1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20 package org.apache.hadoop.hbase.client;
21
22 import static org.junit.Assert.assertEquals;
23 import static org.junit.Assert.assertFalse;
24 import static org.junit.Assert.assertNotEquals;
25 import static org.junit.Assert.assertTrue;
26 import static org.junit.Assert.fail;
27
28 import java.io.IOException;
29 import java.io.InterruptedIOException;
30 import java.util.ArrayList;
31 import java.util.Arrays;
32 import java.util.Collections;
33 import java.util.HashMap;
34 import java.util.Iterator;
35 import java.util.LinkedList;
36 import java.util.List;
37 import java.util.Map;
38 import java.util.Random;
39 import java.util.Set;
40 import java.util.TreeSet;
41 import java.util.concurrent.BlockingQueue;
42 import java.util.concurrent.BrokenBarrierException;
43 import java.util.concurrent.CyclicBarrier;
44 import java.util.concurrent.ExecutorService;
45 import java.util.concurrent.Future;
46 import java.util.concurrent.LinkedBlockingQueue;
47 import java.util.concurrent.RejectedExecutionException;
48 import java.util.concurrent.SynchronousQueue;
49 import java.util.concurrent.ThreadFactory;
50 import java.util.concurrent.ThreadPoolExecutor;
51 import java.util.concurrent.TimeUnit;
52 import java.util.concurrent.atomic.AtomicBoolean;
53 import java.util.concurrent.atomic.AtomicInteger;
54 import java.util.concurrent.atomic.AtomicLong;
55
56 import org.apache.commons.logging.Log;
57 import org.apache.commons.logging.LogFactory;
58 import org.apache.hadoop.conf.Configuration;
59 import org.apache.hadoop.hbase.CallQueueTooBigException;
60 import org.apache.hadoop.hbase.Cell;
61 import org.apache.hadoop.hbase.HConstants;
62 import org.apache.hadoop.hbase.HRegionInfo;
63 import org.apache.hadoop.hbase.HRegionLocation;
64 import org.apache.hadoop.hbase.RegionLocations;
65 import org.apache.hadoop.hbase.ServerName;
66 import org.apache.hadoop.hbase.TableName;
67 import org.apache.hadoop.hbase.client.AsyncProcess.AsyncRequestFuture;
68 import org.apache.hadoop.hbase.client.AsyncProcess.AsyncRequestFutureImpl;
69 import org.apache.hadoop.hbase.client.AsyncProcess.ListRowAccess;
70 import org.apache.hadoop.hbase.client.AsyncProcess.RequestSizeChecker;
71 import org.apache.hadoop.hbase.client.AsyncProcess.RowChecker;
72 import org.apache.hadoop.hbase.client.AsyncProcess.RowChecker.ReturnCode;
73 import org.apache.hadoop.hbase.client.AsyncProcess.RowCheckerHost;
74 import org.apache.hadoop.hbase.client.AsyncProcess.SubmittedSizeChecker;
75 import org.apache.hadoop.hbase.client.AsyncProcess.TaskCountChecker;
76 import org.apache.hadoop.hbase.client.backoff.ClientBackoffPolicy;
77 import org.apache.hadoop.hbase.client.backoff.ServerStatistics;
78 import org.apache.hadoop.hbase.client.coprocessor.Batch;
79 import org.apache.hadoop.hbase.client.coprocessor.Batch.Callback;
80 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
81 import org.apache.hadoop.hbase.testclassification.MediumTests;
82 import org.apache.hadoop.hbase.util.Bytes;
83 import org.apache.hadoop.hbase.util.Threads;
84 import org.junit.Assert;
85 import org.junit.BeforeClass;
86 import org.junit.Rule;
87 import org.junit.Test;
88 import org.junit.experimental.categories.Category;
89 import org.junit.rules.Timeout;
90 import org.mockito.Mockito;
91
92 @Category(MediumTests.class)
93 public class TestAsyncProcess {
94 private final static Log LOG = LogFactory.getLog(TestAsyncProcess.class);
95 private static final TableName DUMMY_TABLE =
96 TableName.valueOf("DUMMY_TABLE");
97 private static final byte[] DUMMY_BYTES_1 = Bytes.toBytes("DUMMY_BYTES_1");
98 private static final byte[] DUMMY_BYTES_2 = Bytes.toBytes("DUMMY_BYTES_2");
99 private static final byte[] DUMMY_BYTES_3 = Bytes.toBytes("DUMMY_BYTES_3");
100 private static final byte[] FAILS = Bytes.toBytes("FAILS");
101 private static final Configuration conf = new Configuration();
102
103 private static ServerName sn = ServerName.valueOf("s1:1,1");
104 private static ServerName sn2 = ServerName.valueOf("s2:2,2");
105 private static ServerName sn3 = ServerName.valueOf("s3:3,3");
106 private static HRegionInfo hri1 =
107 new HRegionInfo(DUMMY_TABLE, DUMMY_BYTES_1, DUMMY_BYTES_2, false, 1);
108 private static HRegionInfo hri2 =
109 new HRegionInfo(DUMMY_TABLE, DUMMY_BYTES_2, HConstants.EMPTY_END_ROW, false, 2);
110 private static HRegionInfo hri3 =
111 new HRegionInfo(DUMMY_TABLE, DUMMY_BYTES_3, HConstants.EMPTY_END_ROW, false, 3);
112 private static HRegionLocation loc1 = new HRegionLocation(hri1, sn);
113 private static HRegionLocation loc2 = new HRegionLocation(hri2, sn);
114 private static HRegionLocation loc3 = new HRegionLocation(hri3, sn2);
115
116
117 private static HRegionInfo hri1r1 = RegionReplicaUtil.getRegionInfoForReplica(hri1, 1),
118 hri1r2 = RegionReplicaUtil.getRegionInfoForReplica(hri1, 2);
119 private static HRegionInfo hri2r1 = RegionReplicaUtil.getRegionInfoForReplica(hri2, 1);
120 private static RegionLocations hrls1 = new RegionLocations(new HRegionLocation(hri1, sn),
121 new HRegionLocation(hri1r1, sn2), new HRegionLocation(hri1r2, sn3));
122 private static RegionLocations hrls2 = new RegionLocations(new HRegionLocation(hri2, sn2),
123 new HRegionLocation(hri2r1, sn3));
124 private static RegionLocations hrls3 = new RegionLocations(new HRegionLocation(hri3, sn3), null);
125
126 private static final String success = "success";
127 private static Exception failure = new Exception("failure");
128
129 private static int NB_RETRIES = 3;
130
131 @BeforeClass
132 public static void beforeClass(){
133 conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, NB_RETRIES);
134 }
135
136 static class CountingThreadFactory implements ThreadFactory {
137 final AtomicInteger nbThreads;
138 ThreadFactory realFactory = Threads.newDaemonThreadFactory("test-TestAsyncProcess");
139 @Override
140 public Thread newThread(Runnable r) {
141 nbThreads.incrementAndGet();
142 return realFactory.newThread(r);
143 }
144
145 CountingThreadFactory(AtomicInteger nbThreads){
146 this.nbThreads = nbThreads;
147 }
148 }
149
150 static class MyAsyncProcess extends AsyncProcess {
151 final AtomicInteger nbMultiResponse = new AtomicInteger();
152 final AtomicInteger nbActions = new AtomicInteger();
153 public List<AsyncRequestFuture> allReqs = new ArrayList<AsyncRequestFuture>();
154 public AtomicInteger callsCt = new AtomicInteger();
155 private static int rpcTimeout = conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY, HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
156 private long previousTimeout = -1;
157
158 @Override
159 <Res> AsyncRequestFutureImpl<Res> createAsyncRequestFuture(TableName tableName,
160 List<Action<Row>> actions, long nonceGroup, ExecutorService pool,
161 Batch.Callback<Res> callback, Object[] results, boolean needResults,
162 PayloadCarryingServerCallable callable, int operationTimeout, int rpcTimeout) {
163
164 MyAsyncRequestFutureImpl<Res> r = new MyAsyncRequestFutureImpl(
165 DUMMY_TABLE, actions, nonceGroup, getPool(pool), needResults, results, callback, callable,
166 operationTimeout, rpcTimeout);
167 allReqs.add(r);
168 callsCt.incrementAndGet();
169 return r;
170 }
171
172 public MyAsyncProcess(ClusterConnection hc, Configuration conf) {
173 this(hc, conf, new AtomicInteger());
174 }
175
176 public MyAsyncProcess(ClusterConnection hc, Configuration conf, AtomicInteger nbThreads) {
177 super(hc, conf, new ThreadPoolExecutor(1, 20, 60, TimeUnit.SECONDS,
178 new SynchronousQueue<Runnable>(), new CountingThreadFactory(nbThreads)),
179 new RpcRetryingCallerFactory(conf), false, new RpcControllerFactory(conf), rpcTimeout);
180 }
181
182 public MyAsyncProcess(
183 ClusterConnection hc, Configuration conf, boolean useGlobalErrors) {
184 super(hc, conf, new ThreadPoolExecutor(1, 20, 60, TimeUnit.SECONDS,
185 new SynchronousQueue<Runnable>(), new CountingThreadFactory(new AtomicInteger())),
186 new RpcRetryingCallerFactory(conf), useGlobalErrors, new RpcControllerFactory(conf), rpcTimeout);
187 }
188
189 public MyAsyncProcess(ClusterConnection hc, Configuration conf, boolean useGlobalErrors,
190 @SuppressWarnings("unused") boolean dummy) {
191 super(hc, conf, new ThreadPoolExecutor(1, 20, 60, TimeUnit.SECONDS,
192 new SynchronousQueue<Runnable>(), new CountingThreadFactory(new AtomicInteger())) {
193 @Override
194 public void execute(Runnable command) {
195 throw new RejectedExecutionException("test under failure");
196 }
197 },
198 new RpcRetryingCallerFactory(conf), useGlobalErrors, new RpcControllerFactory(conf), rpcTimeout);
199 }
200
201 @Override
202 public <Res> AsyncRequestFuture submit(TableName tableName, RowAccess<? extends Row> rows,
203 boolean atLeastOne, Callback<Res> callback, boolean needResults)
204 throws InterruptedIOException {
205
206 return super.submit(DUMMY_TABLE, rows, atLeastOne, callback, true);
207 }
208 @Override
209 public <Res> AsyncRequestFuture submit(TableName tableName, List<? extends Row> rows,
210 boolean atLeastOne, Callback<Res> callback, boolean needResults)
211 throws InterruptedIOException {
212
213 return super.submit(DUMMY_TABLE, rows, atLeastOne, callback, true);
214 }
215 @Override
216 public <CResult> AsyncRequestFuture submitAll(ExecutorService pool, TableName tableName,
217 List<? extends Row> rows, Batch.Callback<CResult> callback, Object[] results,
218 PayloadCarryingServerCallable callable, int operationTimeout, int rpcTimeout) {
219 previousTimeout = rpcTimeout;
220 return super.submitAll(pool, tableName, rows, callback, results, callable, operationTimeout,
221 rpcTimeout);
222 }
223
224 @Override
225 protected void updateStats(ServerName server, Map<byte[], MultiResponse.RegionResult> results) {
226
227 }
228 @Override
229 protected RpcRetryingCaller<MultiResponse> createCaller(
230 PayloadCarryingServerCallable callable, int rpcTimeout) {
231 callsCt.incrementAndGet();
232 MultiServerCallable callable1 = (MultiServerCallable) callable;
233 final MultiResponse mr = createMultiResponse(
234 callable1.getMulti(), nbMultiResponse, nbActions, new ResponseGenerator() {
235 @Override
236 public void addResponse(MultiResponse mr, byte[] regionName, Action<Row> a) {
237 if (Arrays.equals(FAILS, a.getAction().getRow())) {
238 mr.add(regionName, a.getOriginalIndex(), failure);
239 } else {
240 mr.add(regionName, a.getOriginalIndex(), success);
241 }
242 }
243 });
244
245 return new RpcRetryingCaller<MultiResponse>(100, 500, 10, 9) {
246 @Override
247 public MultiResponse callWithoutRetries(RetryingCallable<MultiResponse> callable,
248 int callTimeout)
249 throws IOException, RuntimeException {
250 try {
251
252
253 Thread.sleep(1000);
254 } catch (InterruptedException e) {
255
256 }
257 return mr;
258 }
259 };
260 }
261
262 class MyAsyncRequestFutureImpl<Res> extends AsyncRequestFutureImpl<Res> {
263
264 private final Map<ServerName, List<Long>> heapSizesByServer = new HashMap<>();
265
266 MyAsyncRequestFutureImpl(TableName tableName, List<Action<Row>> actions, long nonceGroup,
267 ExecutorService pool, boolean needResults, Object[] results,
268 Batch.Callback<Res> callback, PayloadCarryingServerCallable callable,
269 int operationTimeout, int rpcTimeout) {
270 super(tableName, actions, nonceGroup, pool, needResults, results, callback, callable, operationTimeout, rpcTimeout);
271 }
272
273 Map<ServerName, List<Long>> getRequestHeapSize() {
274 return heapSizesByServer;
275 }
276
277 @Override
278 SingleServerRequestRunnable createSingleServerRequest(
279 MultiAction<Row> multiAction, int numAttempt, ServerName server,
280 Set<PayloadCarryingServerCallable> callsInProgress) {
281 SingleServerRequestRunnable rq = new SingleServerRequestRunnable(
282 multiAction, numAttempt, server, callsInProgress);
283 List<Long> heapCount = heapSizesByServer.get(server);
284 if (heapCount == null) {
285 heapCount = new ArrayList<>();
286 heapSizesByServer.put(server, heapCount);
287 }
288 heapCount.add(heapSizeOf(multiAction));
289 return rq;
290 }
291
292 long heapSizeOf(MultiAction<Row> multiAction) {
293 long sum = 0;
294 for (List<Action<Row>> actions : multiAction.actions.values()) {
295 for (Action action : actions) {
296 Row row = action.getAction();
297 if (row instanceof Mutation) {
298 sum += ((Mutation) row).heapSize();
299 }
300 }
301 }
302 return sum;
303 }
304 }
305 }
306
307 static class CallerWithFailure extends RpcRetryingCaller<MultiResponse>{
308
309 private final IOException e;
310
311 public CallerWithFailure(IOException e) {
312 super(100, 500, 100, 9);
313 this.e = e;
314 }
315
316 @Override
317 public MultiResponse callWithoutRetries(RetryingCallable<MultiResponse> callable,
318 int callTimeout)
319 throws IOException, RuntimeException {
320 throw e;
321 }
322 }
323
324
325 static class AsyncProcessWithFailure extends MyAsyncProcess {
326
327 private final IOException ioe;
328
329 public AsyncProcessWithFailure(ClusterConnection hc, Configuration conf, IOException ioe) {
330 super(hc, conf, true);
331 this.ioe = ioe;
332 serverTrackerTimeout = 1;
333 }
334
335 @Override
336 protected RpcRetryingCaller<MultiResponse> createCaller(
337 PayloadCarryingServerCallable callable, int rpcTimeout) {
338 callsCt.incrementAndGet();
339 return new CallerWithFailure(ioe);
340 }
341 }
342
343
344
345 static class MyClientBackoffPolicy implements ClientBackoffPolicy {
346 private final Map<ServerName, AtomicInteger> count = new HashMap<>();
347 @Override
348 public long getBackoffTime(ServerName serverName, byte[] region, ServerStatistics stats) {
349 AtomicInteger inc = count.get(serverName);
350 if (inc == null) {
351 inc = new AtomicInteger(0);
352 count.put(serverName, inc);
353 }
354 return inc.getAndIncrement();
355 }
356 }
357
358 static class MyAsyncProcessWithReplicas extends MyAsyncProcess {
359 private Set<byte[]> failures = new TreeSet<byte[]>(new Bytes.ByteArrayComparator());
360 private long primarySleepMs = 0, replicaSleepMs = 0;
361 private Map<ServerName, Long> customPrimarySleepMs = new HashMap<ServerName, Long>();
362 private final AtomicLong replicaCalls = new AtomicLong(0);
363
364 public void addFailures(HRegionInfo... hris) {
365 for (HRegionInfo hri : hris) {
366 failures.add(hri.getRegionName());
367 }
368 }
369
370 public long getReplicaCallCount() {
371 return replicaCalls.get();
372 }
373
374 public void setPrimaryCallDelay(ServerName server, long primaryMs) {
375 customPrimarySleepMs.put(server, primaryMs);
376 }
377
378 public MyAsyncProcessWithReplicas(ClusterConnection hc, Configuration conf) {
379 super(hc, conf);
380 }
381
382 public void setCallDelays(long primaryMs, long replicaMs) {
383 this.primarySleepMs = primaryMs;
384 this.replicaSleepMs = replicaMs;
385 }
386
387 @Override
388 protected RpcRetryingCaller<MultiResponse> createCaller(
389 PayloadCarryingServerCallable payloadCallable, int rpcTimeout) {
390 MultiServerCallable<Row> callable = (MultiServerCallable) payloadCallable;
391 final MultiResponse mr = createMultiResponse(
392 callable.getMulti(), nbMultiResponse, nbActions, new ResponseGenerator() {
393 @Override
394 public void addResponse(MultiResponse mr, byte[] regionName, Action<Row> a) {
395 if (failures.contains(regionName)) {
396 mr.add(regionName, a.getOriginalIndex(), failure);
397 } else {
398 boolean isStale = !RegionReplicaUtil.isDefaultReplica(a.getReplicaId());
399 mr.add(regionName, a.getOriginalIndex(),
400 Result.create(new Cell[0], null, isStale));
401 }
402 }
403 });
404
405 final boolean isDefault = RegionReplicaUtil.isDefaultReplica(
406 callable.getMulti().actions.values().iterator().next().iterator().next().getReplicaId());
407 final ServerName server = ((MultiServerCallable<?>)callable).getServerName();
408 String debugMsg = "Call to " + server + ", primary=" + isDefault + " with "
409 + callable.getMulti().actions.size() + " entries: ";
410 for (byte[] region : callable.getMulti().actions.keySet()) {
411 debugMsg += "[" + Bytes.toStringBinary(region) + "], ";
412 }
413 LOG.debug(debugMsg);
414 if (!isDefault) {
415 replicaCalls.incrementAndGet();
416 }
417
418 return new RpcRetryingCaller<MultiResponse>(100, 500, 10, 9) {
419 @Override
420 public MultiResponse callWithoutRetries(RetryingCallable<MultiResponse> callable,
421 int callTimeout)
422 throws IOException, RuntimeException {
423 long sleep = -1;
424 if (isDefault) {
425 Long customSleep = customPrimarySleepMs.get(server);
426 sleep = (customSleep == null ? primarySleepMs : customSleep.longValue());
427 } else {
428 sleep = replicaSleepMs;
429 }
430 if (sleep != 0) {
431 try {
432 Thread.sleep(sleep);
433 } catch (InterruptedException e) {
434 }
435 }
436 return mr;
437 }
438 };
439 }
440 }
441
442 static MultiResponse createMultiResponse(final MultiAction<Row> multi,
443 AtomicInteger nbMultiResponse, AtomicInteger nbActions, ResponseGenerator gen) {
444 final MultiResponse mr = new MultiResponse();
445 nbMultiResponse.incrementAndGet();
446 for (Map.Entry<byte[], List<Action<Row>>> entry : multi.actions.entrySet()) {
447 byte[] regionName = entry.getKey();
448 for (Action<Row> a : entry.getValue()) {
449 nbActions.incrementAndGet();
450 gen.addResponse(mr, regionName, a);
451 }
452 }
453 return mr;
454 }
455
456 private static interface ResponseGenerator {
457 void addResponse(final MultiResponse mr, byte[] regionName, Action<Row> a);
458 }
459
460
461
462
463 static class MyConnectionImpl extends ConnectionManager.HConnectionImplementation {
464 public static class TestRegistry implements Registry {
465 @Override
466 public void init(Connection connection) {}
467
468 @Override
469 public RegionLocations getMetaRegionLocation() throws IOException {
470 return null;
471 }
472
473 @Override
474 public String getClusterId() {
475 return "testClusterId";
476 }
477
478 @Override
479 public boolean isTableOnlineState(TableName tableName, boolean enabled) throws IOException {
480 return false;
481 }
482
483 @Override
484 public int getCurrentNrHRS() throws IOException {
485 return 1;
486 }
487 }
488
489 final AtomicInteger nbThreads = new AtomicInteger(0);
490
491 protected MyConnectionImpl(Configuration conf) throws IOException {
492 super(setupConf(conf), false);
493 }
494
495 private static Configuration setupConf(Configuration conf) {
496 conf.setClass(RegistryFactory.REGISTRY_IMPL_CONF_KEY, TestRegistry.class, Registry.class);
497 return conf;
498 }
499
500 @Override
501 public RegionLocations locateRegion(TableName tableName,
502 byte[] row, boolean useCache, boolean retry, int replicaId) throws IOException {
503 return new RegionLocations(loc1);
504 }
505
506 @Override
507 public boolean hasCellBlockSupport() {
508 return false;
509 }
510 }
511
512
513
514
515 static class MyConnectionImpl2 extends MyConnectionImpl {
516 List<HRegionLocation> hrl;
517 final boolean usedRegions[];
518
519 protected MyConnectionImpl2(List<HRegionLocation> hrl) throws IOException {
520 super(conf);
521 this.hrl = hrl;
522 this.usedRegions = new boolean[hrl.size()];
523 }
524
525 @Override
526 public RegionLocations locateRegion(TableName tableName,
527 byte[] row, boolean useCache, boolean retry, int replicaId) throws IOException {
528 int i = 0;
529 for (HRegionLocation hr : hrl){
530 if (Arrays.equals(row, hr.getRegionInfo().getStartKey())) {
531 usedRegions[i] = true;
532 return new RegionLocations(hr);
533 }
534 i++;
535 }
536 return null;
537 }
538 }
539
540 @Rule
541 public Timeout timeout = Timeout.millis(10000);
542
543 @Test
544 public void testSubmit() throws Exception {
545 ClusterConnection hc = createHConnection();
546 AsyncProcess ap = new MyAsyncProcess(hc, conf);
547
548 List<Put> puts = new ArrayList<Put>();
549 puts.add(createPut(1, true));
550
551 ap.submit(DUMMY_TABLE, puts, false, null, false);
552 Assert.assertTrue(puts.isEmpty());
553 }
554 @Test
555 public void testListRowAccess() {
556 int count = 10;
557 List<String> values = new LinkedList<>();
558 for (int i = 0; i != count; ++i) {
559 values.add(String.valueOf(i));
560 }
561
562 ListRowAccess<String> taker = new ListRowAccess(values);
563 assertEquals(count, taker.size());
564
565 int restoreCount = 0;
566 int takeCount = 0;
567 Iterator<String> it = taker.iterator();
568 while (it.hasNext()) {
569 String v = it.next();
570 assertEquals(String.valueOf(takeCount), v);
571 ++takeCount;
572 it.remove();
573 if (Math.random() >= 0.5) {
574 break;
575 }
576 }
577 assertEquals(count, taker.size() + takeCount);
578
579 it = taker.iterator();
580 while (it.hasNext()) {
581 String v = it.next();
582 assertEquals(String.valueOf(takeCount), v);
583 ++takeCount;
584 it.remove();
585 }
586 assertEquals(0, taker.size());
587 assertEquals(count, takeCount);
588 }
589 private static long calculateRequestCount(long putSizePerServer, long maxHeapSizePerRequest) {
590 if (putSizePerServer <= maxHeapSizePerRequest) {
591 return 1;
592 } else if (putSizePerServer % maxHeapSizePerRequest == 0) {
593 return putSizePerServer / maxHeapSizePerRequest;
594 } else {
595 return putSizePerServer / maxHeapSizePerRequest + 1;
596 }
597 }
598
599 @Test
600 public void testSubmitSameSizeOfRequest() throws Exception {
601 long writeBuffer = 2 * 1024 * 1024;
602 long putsHeapSize = writeBuffer;
603 doSubmitRequest(writeBuffer, putsHeapSize);
604 }
605 @Test
606 public void testIllegalArgument() throws IOException {
607 ClusterConnection conn = createHConnection();
608 final long maxHeapSizePerRequest = conn.getConfiguration().getLong(AsyncProcess.HBASE_CLIENT_MAX_PERREQUEST_HEAPSIZE,
609 AsyncProcess.DEFAULT_HBASE_CLIENT_MAX_PERREQUEST_HEAPSIZE);
610 conn.getConfiguration().setLong(AsyncProcess.HBASE_CLIENT_MAX_PERREQUEST_HEAPSIZE, -1);
611 try {
612 MyAsyncProcess ap = new MyAsyncProcess(conn, conf, true);
613 fail("The maxHeapSizePerRequest must be bigger than zero");
614 } catch (IllegalArgumentException e) {
615 }
616 conn.getConfiguration().setLong(AsyncProcess.HBASE_CLIENT_MAX_PERREQUEST_HEAPSIZE, maxHeapSizePerRequest);
617 }
618 @Test
619 public void testSubmitLargeRequestWithUnlimitedSize() throws Exception {
620 long maxHeapSizePerRequest = Long.MAX_VALUE;
621 long putsHeapSize = 2 * 1024 * 1024;
622 doSubmitRequest(maxHeapSizePerRequest, putsHeapSize);
623 }
624
625 @Test(timeout=300000)
626 public void testSubmitRandomSizeRequest() throws Exception {
627 Random rn = new Random();
628 final long limit = 10 * 1024 * 1024;
629 final int requestCount = 1 + (int) (rn.nextDouble() * 3);
630 long n = rn.nextLong();
631 if (n < 0) {
632 n = -n;
633 } else if (n == 0) {
634 n = 1;
635 }
636 long putsHeapSize = n % limit;
637 long maxHeapSizePerRequest = putsHeapSize / requestCount;
638 LOG.info("[testSubmitRandomSizeRequest] maxHeapSizePerRequest=" + maxHeapSizePerRequest +
639 ", putsHeapSize=" + putsHeapSize);
640 doSubmitRequest(maxHeapSizePerRequest, putsHeapSize);
641 }
642
643 @Test
644 public void testSubmitSmallRequest() throws Exception {
645 long maxHeapSizePerRequest = 2 * 1024 * 1024;
646 long putsHeapSize = 100;
647 doSubmitRequest(maxHeapSizePerRequest, putsHeapSize);
648 }
649
650 @Test(timeout=120000)
651 public void testSubmitLargeRequest() throws Exception {
652 long maxHeapSizePerRequest = 2 * 1024 * 1024;
653 long putsHeapSize = maxHeapSizePerRequest * 2;
654 doSubmitRequest(maxHeapSizePerRequest, putsHeapSize);
655 }
656
657 private void doSubmitRequest(long maxHeapSizePerRequest, long putsHeapSize) throws Exception {
658 ClusterConnection conn = createHConnection();
659 final long defaultHeapSizePerRequest = conn.getConfiguration().getLong(AsyncProcess.HBASE_CLIENT_MAX_PERREQUEST_HEAPSIZE,
660 AsyncProcess.DEFAULT_HBASE_CLIENT_MAX_PERREQUEST_HEAPSIZE);
661 conn.getConfiguration().setLong(AsyncProcess.HBASE_CLIENT_MAX_PERREQUEST_HEAPSIZE, maxHeapSizePerRequest);
662 BufferedMutatorParams bufferParam = new BufferedMutatorParams(DUMMY_TABLE);
663
664
665 long putSizeSN = 0;
666 long putSizeSN2 = 0;
667 List<Put> puts = new ArrayList<>();
668 while ((putSizeSN + putSizeSN2) <= putsHeapSize) {
669 Put put1 = new Put(DUMMY_BYTES_1);
670 put1.addColumn(DUMMY_BYTES_1, DUMMY_BYTES_1, DUMMY_BYTES_1);
671 Put put2 = new Put(DUMMY_BYTES_2);
672 put2.addColumn(DUMMY_BYTES_2, DUMMY_BYTES_2, DUMMY_BYTES_2);
673 Put put3 = new Put(DUMMY_BYTES_3);
674 put3.addColumn(DUMMY_BYTES_3, DUMMY_BYTES_3, DUMMY_BYTES_3);
675 putSizeSN += (put1.heapSize() + put2.heapSize());
676 putSizeSN2 += put3.heapSize();
677 puts.add(put1);
678 puts.add(put2);
679 puts.add(put3);
680 }
681
682 int minCountSnRequest = (int) calculateRequestCount(putSizeSN, maxHeapSizePerRequest);
683 int minCountSn2Request = (int) calculateRequestCount(putSizeSN2, maxHeapSizePerRequest);
684 LOG.info("Total put count:" + puts.size() + ", putSizeSN:"+ putSizeSN + ", putSizeSN2:" + putSizeSN2
685 + ", maxHeapSizePerRequest:" + maxHeapSizePerRequest
686 + ", minCountSnRequest:" + minCountSnRequest
687 + ", minCountSn2Request:" + minCountSn2Request);
688 try (HTable ht = new HTable(conn, bufferParam)) {
689 MyAsyncProcess ap = new MyAsyncProcess(conn, conf, true);
690 ht.mutator.ap = ap;
691
692 Assert.assertEquals(0L, ht.mutator.currentWriteBufferSize.get());
693 ht.put(puts);
694 List<AsyncRequestFuture> reqs = ap.allReqs;
695
696 int actualSnReqCount = 0;
697 int actualSn2ReqCount = 0;
698 for (AsyncRequestFuture req : reqs) {
699 if (!(req instanceof AsyncRequestFutureImpl)) {
700 continue;
701 }
702 MyAsyncProcess.MyAsyncRequestFutureImpl ars = (MyAsyncProcess.MyAsyncRequestFutureImpl) req;
703 if (ars.getRequestHeapSize().containsKey(sn)) {
704 ++actualSnReqCount;
705 }
706 if (ars.getRequestHeapSize().containsKey(sn2)) {
707 ++actualSn2ReqCount;
708 }
709 }
710
711 assertEquals(true, minCountSnRequest <= actualSnReqCount);
712 assertEquals(true, minCountSn2Request <= actualSn2ReqCount);
713 Map<ServerName, Long> sizePerServers = new HashMap<>();
714 for (AsyncRequestFuture req : reqs) {
715 if (!(req instanceof AsyncRequestFutureImpl)) {
716 continue;
717 }
718 MyAsyncProcess.MyAsyncRequestFutureImpl ars = (MyAsyncProcess.MyAsyncRequestFutureImpl) req;
719 Map<ServerName, List<Long>> requestHeapSize = ars.getRequestHeapSize();
720 for (Map.Entry<ServerName, List<Long>> entry : requestHeapSize.entrySet()) {
721 long sum = 0;
722 for (long size : entry.getValue()) {
723 assertEquals(true, size <= maxHeapSizePerRequest);
724 sum += size;
725 }
726 assertEquals(true, sum <= maxHeapSizePerRequest);
727 long value = sizePerServers.containsKey(entry.getKey()) ? sizePerServers.get(entry.getKey()) : 0L;
728 sizePerServers.put(entry.getKey(), value + sum);
729 }
730 }
731 assertEquals(true, sizePerServers.containsKey(sn));
732 assertEquals(true, sizePerServers.containsKey(sn2));
733 assertEquals(false, sizePerServers.containsKey(sn3));
734 assertEquals(putSizeSN, (long) sizePerServers.get(sn));
735 assertEquals(putSizeSN2, (long) sizePerServers.get(sn2));
736 }
737
738 conn.getConfiguration().setLong(AsyncProcess.HBASE_CLIENT_MAX_PERREQUEST_HEAPSIZE, defaultHeapSizePerRequest);
739 }
740 @Test
741 public void testSubmitWithCB() throws Exception {
742 ClusterConnection hc = createHConnection();
743 final AtomicInteger updateCalled = new AtomicInteger(0);
744 Batch.Callback<Object> cb = new Batch.Callback<Object>() {
745 @Override
746 public void update(byte[] region, byte[] row, Object result) {
747 updateCalled.incrementAndGet();
748 }
749 };
750 AsyncProcess ap = new MyAsyncProcess(hc, conf);
751
752 List<Put> puts = new ArrayList<Put>();
753 puts.add(createPut(1, true));
754
755 final AsyncRequestFuture ars = ap.submit(DUMMY_TABLE, puts, false, cb, false);
756 Assert.assertTrue(puts.isEmpty());
757 ars.waitUntilDone();
758 Assert.assertEquals(1, updateCalled.get());
759 }
760
761 @Test
762 public void testSubmitBusyRegion() throws Exception {
763 ClusterConnection hc = createHConnection();
764 AsyncProcess ap = new MyAsyncProcess(hc, conf);
765
766 List<Put> puts = new ArrayList<Put>();
767 puts.add(createPut(1, true));
768
769 for (int i = 0; i != ap.maxConcurrentTasksPerRegion; ++i) {
770 ap.incTaskCounters(Collections.singletonList(hri1.getRegionName()), sn);
771 }
772 ap.submit(DUMMY_TABLE, puts, false, null, false);
773 Assert.assertEquals(puts.size(), 1);
774
775 ap.decTaskCounters(Collections.singletonList(hri1.getRegionName()), sn);
776 ap.submit(DUMMY_TABLE, puts, false, null, false);
777 Assert.assertEquals(0, puts.size());
778 }
779
780
781 @Test
782 public void testSubmitBusyRegionServer() throws Exception {
783 ClusterConnection hc = createHConnection();
784 AsyncProcess ap = new MyAsyncProcess(hc, conf);
785
786 ap.taskCounterPerServer.put(sn2, new AtomicInteger(ap.maxConcurrentTasksPerServer));
787
788 List<Put> puts = new ArrayList<Put>();
789 puts.add(createPut(1, true));
790 puts.add(createPut(3, true));
791 puts.add(createPut(1, true));
792 puts.add(createPut(2, true));
793
794 ap.submit(DUMMY_TABLE, puts, false, null, false);
795 Assert.assertEquals(" puts=" + puts, 1, puts.size());
796
797 ap.taskCounterPerServer.put(sn2, new AtomicInteger(ap.maxConcurrentTasksPerServer - 1));
798 ap.submit(DUMMY_TABLE, puts, false, null, false);
799 Assert.assertTrue(puts.isEmpty());
800 }
801
802 @Test
803 public void testFail() throws Exception {
804 MyAsyncProcess ap = new MyAsyncProcess(createHConnection(), conf, false);
805
806 List<Put> puts = new ArrayList<Put>();
807 Put p = createPut(1, false);
808 puts.add(p);
809
810 AsyncRequestFuture ars = ap.submit(DUMMY_TABLE, puts, false, null, true);
811 Assert.assertEquals(0, puts.size());
812 ars.waitUntilDone();
813 verifyResult(ars, false);
814 Assert.assertEquals(NB_RETRIES + 1, ap.callsCt.get());
815
816 Assert.assertEquals(1, ars.getErrors().exceptions.size());
817 Assert.assertTrue("was: " + ars.getErrors().exceptions.get(0),
818 failure.equals(ars.getErrors().exceptions.get(0)));
819 Assert.assertTrue("was: " + ars.getErrors().exceptions.get(0),
820 failure.equals(ars.getErrors().exceptions.get(0)));
821
822 Assert.assertEquals(1, ars.getFailedOperations().size());
823 Assert.assertTrue("was: " + ars.getFailedOperations().get(0),
824 p.equals(ars.getFailedOperations().get(0)));
825 }
826
827
828 @Test
829 public void testSubmitTrue() throws IOException {
830 final AsyncProcess ap = new MyAsyncProcess(createHConnection(), conf, false);
831 ap.tasksInProgress.incrementAndGet();
832 final AtomicInteger ai = new AtomicInteger(ap.maxConcurrentTasksPerRegion);
833 ap.taskCounterPerRegion.put(hri1.getRegionName(), ai);
834
835 final AtomicBoolean checkPoint = new AtomicBoolean(false);
836 final AtomicBoolean checkPoint2 = new AtomicBoolean(false);
837
838 Thread t = new Thread(){
839 @Override
840 public void run(){
841 Threads.sleep(1000);
842 Assert.assertFalse(checkPoint.get());
843 ai.decrementAndGet();
844 ap.tasksInProgress.decrementAndGet();
845 checkPoint2.set(true);
846 }
847 };
848
849 List<Put> puts = new ArrayList<Put>();
850 Put p = createPut(1, true);
851 puts.add(p);
852
853 ap.submit(DUMMY_TABLE, puts, false, null, false);
854 Assert.assertFalse(puts.isEmpty());
855
856 t.start();
857
858 ap.submit(DUMMY_TABLE, puts, true, null, false);
859 Assert.assertTrue(puts.isEmpty());
860
861 checkPoint.set(true);
862 while (!checkPoint2.get()){
863 Threads.sleep(1);
864 }
865 }
866
867 @Test
868 public void testFailAndSuccess() throws Exception {
869 MyAsyncProcess ap = new MyAsyncProcess(createHConnection(), conf, false);
870
871 List<Put> puts = new ArrayList<Put>();
872 puts.add(createPut(1, false));
873 puts.add(createPut(1, true));
874 puts.add(createPut(1, true));
875
876 AsyncRequestFuture ars = ap.submit(DUMMY_TABLE, puts, false, null, true);
877 Assert.assertTrue(puts.isEmpty());
878 ars.waitUntilDone();
879 verifyResult(ars, false, true, true);
880 Assert.assertEquals(NB_RETRIES + 1, ap.callsCt.get());
881 ap.callsCt.set(0);
882 Assert.assertEquals(1, ars.getErrors().actions.size());
883
884 puts.add(createPut(1, true));
885
886 ap.waitUntilDone();
887 ars = ap.submit(DUMMY_TABLE, puts, false, null, true);
888 Assert.assertEquals(0, puts.size());
889 ars.waitUntilDone();
890 Assert.assertEquals(2, ap.callsCt.get());
891 verifyResult(ars, true);
892 }
893
894 @Test
895 public void testFlush() throws Exception {
896 MyAsyncProcess ap = new MyAsyncProcess(createHConnection(), conf, false);
897
898 List<Put> puts = new ArrayList<Put>();
899 puts.add(createPut(1, false));
900 puts.add(createPut(1, true));
901 puts.add(createPut(1, true));
902
903 AsyncRequestFuture ars = ap.submit(DUMMY_TABLE, puts, false, null, true);
904 ars.waitUntilDone();
905 verifyResult(ars, false, true, true);
906 Assert.assertEquals(NB_RETRIES + 1, ap.callsCt.get());
907
908 Assert.assertEquals(1, ars.getFailedOperations().size());
909 }
910
911 @Test
912 public void testTaskCountWithoutClientBackoffPolicy() throws IOException, InterruptedException {
913 ClusterConnection hc = createHConnection();
914 MyAsyncProcess ap = new MyAsyncProcess(hc, conf, false);
915 testTaskCount(ap);
916 }
917
918 @Test
919 public void testTaskCountWithClientBackoffPolicy() throws IOException, InterruptedException {
920 Configuration copyConf = new Configuration(conf);
921 copyConf.setBoolean(HConstants.ENABLE_CLIENT_BACKPRESSURE, true);
922 MyClientBackoffPolicy bp = new MyClientBackoffPolicy();
923 ClusterConnection hc = createHConnection();
924 Mockito.when(hc.getConfiguration()).thenReturn(copyConf);
925 Mockito.when(hc.getStatisticsTracker()).thenReturn(ServerStatisticTracker.create(copyConf));
926 Mockito.when(hc.getBackoffPolicy()).thenReturn(bp);
927 MyAsyncProcess ap = new MyAsyncProcess(hc, copyConf, false);
928 testTaskCount(ap);
929 }
930
931 private void testTaskCount(AsyncProcess ap) throws InterruptedIOException, InterruptedException {
932 List<Put> puts = new ArrayList<>();
933 for (int i = 0; i != 3; ++i) {
934 puts.add(createPut(1, true));
935 puts.add(createPut(2, true));
936 puts.add(createPut(3, true));
937 }
938 ap.submit(DUMMY_TABLE, puts, true, null, false);
939 ap.waitUntilDone();
940
941 TimeUnit.SECONDS.sleep(1);
942 assertEquals(0, ap.tasksInProgress.get());
943 for (AtomicInteger count : ap.taskCounterPerRegion.values()) {
944 assertEquals(0, count.get());
945 }
946 for (AtomicInteger count : ap.taskCounterPerServer.values()) {
947 assertEquals(0, count.get());
948 }
949 }
950
951 @Test
952 public void testMaxTask() throws Exception {
953 final AsyncProcess ap = new MyAsyncProcess(createHConnection(), conf, false);
954
955 for (int i = 0; i < 1000; i++) {
956 ap.incTaskCounters(Collections.singletonList(Bytes.toBytes("dummy")), sn);
957 }
958
959 final Thread myThread = Thread.currentThread();
960
961 Thread t = new Thread() {
962 @Override
963 public void run() {
964 Threads.sleep(2000);
965 myThread.interrupt();
966 }
967 };
968
969 List<Put> puts = new ArrayList<Put>();
970 puts.add(createPut(1, true));
971
972 t.start();
973
974 try {
975 ap.submit(DUMMY_TABLE, puts, false, null, false);
976 Assert.fail("We should have been interrupted.");
977 } catch (InterruptedIOException expected) {
978 }
979
980 final long sleepTime = 2000;
981
982 Thread t2 = new Thread() {
983 @Override
984 public void run() {
985 Threads.sleep(sleepTime);
986 while (ap.tasksInProgress.get() > 0) {
987 ap.decTaskCounters(Collections.singletonList(Bytes.toBytes("dummy")), sn);
988 }
989 }
990 };
991 t2.start();
992
993 long start = System.currentTimeMillis();
994 ap.submit(DUMMY_TABLE, new ArrayList<Row>(), false, null, false);
995 long end = System.currentTimeMillis();
996
997
998 Assert.assertTrue(start + 100L + sleepTime > end);
999 }
1000
1001 private static ClusterConnection createHConnection() throws IOException {
1002 ClusterConnection hc = createHConnectionCommon();
1003 setMockLocation(hc, DUMMY_BYTES_1, new RegionLocations(loc1));
1004 setMockLocation(hc, DUMMY_BYTES_2, new RegionLocations(loc2));
1005 setMockLocation(hc, DUMMY_BYTES_3, new RegionLocations(loc3));
1006 Mockito.when(hc.locateRegions(Mockito.eq(DUMMY_TABLE), Mockito.anyBoolean(), Mockito.anyBoolean()))
1007 .thenReturn(Arrays.asList(loc1, loc2, loc3));
1008 setMockLocation(hc, FAILS, new RegionLocations(loc2));
1009 return hc;
1010 }
1011
1012 private static ClusterConnection createHConnectionWithReplicas() throws IOException {
1013 ClusterConnection hc = createHConnectionCommon();
1014 setMockLocation(hc, DUMMY_BYTES_1, hrls1);
1015 setMockLocation(hc, DUMMY_BYTES_2, hrls2);
1016 setMockLocation(hc, DUMMY_BYTES_3, hrls3);
1017 List<HRegionLocation> locations = new ArrayList<HRegionLocation>();
1018 for (HRegionLocation loc : hrls1.getRegionLocations()) {
1019 locations.add(loc);
1020 }
1021 for (HRegionLocation loc : hrls2.getRegionLocations()) {
1022 locations.add(loc);
1023 }
1024 for (HRegionLocation loc : hrls3.getRegionLocations()) {
1025 locations.add(loc);
1026 }
1027 Mockito.when(hc.locateRegions(Mockito.eq(DUMMY_TABLE), Mockito.anyBoolean(), Mockito.anyBoolean()))
1028 .thenReturn(locations);
1029 return hc;
1030 }
1031
1032 private static void setMockLocation(ClusterConnection hc, byte[] row,
1033 RegionLocations result) throws IOException {
1034 Mockito.when(hc.locateRegion(Mockito.eq(DUMMY_TABLE), Mockito.eq(row),
1035 Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.anyInt())).thenReturn(result);
1036 Mockito.when(hc.locateRegion(Mockito.eq(DUMMY_TABLE), Mockito.eq(row),
1037 Mockito.anyBoolean(), Mockito.anyBoolean())).thenReturn(result);
1038 }
1039
1040 private static ClusterConnection createHConnectionCommon() {
1041 ClusterConnection hc = Mockito.mock(ClusterConnection.class);
1042 NonceGenerator ng = Mockito.mock(NonceGenerator.class);
1043 Mockito.when(ng.getNonceGroup()).thenReturn(HConstants.NO_NONCE);
1044 Mockito.when(hc.getNonceGenerator()).thenReturn(ng);
1045 Mockito.when(hc.getConfiguration()).thenReturn(conf);
1046 return hc;
1047 }
1048
1049 @Test
1050 public void testHTablePutSuccess() throws Exception {
1051 BufferedMutatorImpl ht = Mockito.mock(BufferedMutatorImpl.class);
1052 ht.ap = new MyAsyncProcess(createHConnection(), conf, true);
1053
1054 Put put = createPut(1, true);
1055
1056 Assert.assertEquals(0, ht.getWriteBufferSize());
1057 ht.mutate(put);
1058 Assert.assertEquals(0, ht.getWriteBufferSize());
1059 }
1060
1061 private void doHTableFailedPut(boolean bufferOn) throws Exception {
1062 ClusterConnection conn = createHConnection();
1063 HTable ht = new HTable(conn, new BufferedMutatorParams(DUMMY_TABLE));
1064 MyAsyncProcess ap = new MyAsyncProcess(conn, conf, true);
1065 ht.mutator.ap = ap;
1066 if (bufferOn) {
1067 ht.setWriteBufferSize(1024L * 1024L);
1068 } else {
1069 ht.setWriteBufferSize(0L);
1070 }
1071
1072 Put put = createPut(1, false);
1073
1074 Assert.assertEquals(0L, ht.mutator.currentWriteBufferSize.get());
1075 try {
1076 ht.put(put);
1077 if (bufferOn) {
1078 ht.flushCommits();
1079 }
1080 Assert.fail();
1081 } catch (RetriesExhaustedException expected) {
1082 }
1083 Assert.assertEquals(0L, ht.mutator.currentWriteBufferSize.get());
1084
1085 AsyncRequestFuture ars = null;
1086 for (AsyncRequestFuture someReqs : ap.allReqs) {
1087 if (someReqs.getResults().length == 0) continue;
1088 Assert.assertTrue(ars == null);
1089 ars = someReqs;
1090 }
1091 Assert.assertTrue(ars != null);
1092 verifyResult(ars, false);
1093
1094
1095 ht.close();
1096 }
1097
1098 @Test
1099 public void testHTableFailedPutWithBuffer() throws Exception {
1100 doHTableFailedPut(true);
1101 }
1102
1103 @Test
1104 public void testHTableFailedPutWithoutBuffer() throws Exception {
1105 doHTableFailedPut(false);
1106 }
1107
1108 @Test
1109 public void testHTableFailedPutAndNewPut() throws Exception {
1110 ClusterConnection conn = createHConnection();
1111 BufferedMutatorImpl mutator = new BufferedMutatorImpl(conn, null, null,
1112 new BufferedMutatorParams(DUMMY_TABLE).writeBufferSize(0));
1113 MyAsyncProcess ap = new MyAsyncProcess(conn, conf, true);
1114 mutator.ap = ap;
1115
1116 Put p = createPut(1, false);
1117 mutator.mutate(p);
1118
1119 ap.waitUntilDone();
1120
1121
1122
1123
1124
1125
1126 p = createPut(1, true);
1127 Assert.assertEquals(0, mutator.getWriteBuffer().size());
1128 try {
1129 mutator.mutate(p);
1130 Assert.fail();
1131 } catch (RetriesExhaustedException expected) {
1132 }
1133 Assert.assertEquals("the put should not been inserted.", 0, mutator.getWriteBuffer().size());
1134 }
1135
1136 @Test
1137 public void testTaskCheckerHost() throws IOException {
1138 final int maxTotalConcurrentTasks = 100;
1139 final int maxConcurrentTasksPerServer = 2;
1140 final int maxConcurrentTasksPerRegion = 1;
1141 final AtomicLong tasksInProgress = new AtomicLong(0);
1142 final Map<ServerName, AtomicInteger> taskCounterPerServer = new HashMap<>();
1143 final Map<byte[], AtomicInteger> taskCounterPerRegion = new HashMap<>();
1144 TaskCountChecker countChecker = new TaskCountChecker(
1145 maxTotalConcurrentTasks,
1146 maxConcurrentTasksPerServer,
1147 maxConcurrentTasksPerRegion,
1148 tasksInProgress, taskCounterPerServer, taskCounterPerRegion);
1149 final long maxHeapSizePerRequest = 2 * 1024 * 1024;
1150
1151 RequestSizeChecker sizeChecker = new RequestSizeChecker(maxHeapSizePerRequest);
1152 RowCheckerHost checkerHost = new RowCheckerHost(Arrays.asList(countChecker, sizeChecker));
1153
1154 ReturnCode loc1Code = checkerHost.canTakeOperation(loc1, maxHeapSizePerRequest);
1155 assertEquals(RowChecker.ReturnCode.INCLUDE, loc1Code);
1156
1157 ReturnCode loc1Code_2 = checkerHost.canTakeOperation(loc1, maxHeapSizePerRequest);
1158
1159 assertNotEquals(RowChecker.ReturnCode.INCLUDE, loc1Code_2);
1160
1161 ReturnCode loc2Code = checkerHost.canTakeOperation(loc2, maxHeapSizePerRequest);
1162
1163 assertNotEquals(RowChecker.ReturnCode.INCLUDE, loc2Code);
1164
1165
1166 taskCounterPerRegion.put(loc3.getRegionInfo().getRegionName(), new AtomicInteger(100));
1167 taskCounterPerServer.put(loc3.getServerName(), new AtomicInteger(100));
1168
1169 ReturnCode loc3Code = checkerHost.canTakeOperation(loc3, 1L);
1170
1171 assertNotEquals(RowChecker.ReturnCode.INCLUDE, loc3Code);
1172
1173
1174 taskCounterPerRegion.put(loc3.getRegionInfo().getRegionName(), new AtomicInteger(0));
1175 taskCounterPerServer.put(loc3.getServerName(), new AtomicInteger(0));
1176
1177 ReturnCode loc3Code_2 = checkerHost.canTakeOperation(loc3, 1L);
1178 assertEquals(RowChecker.ReturnCode.INCLUDE, loc3Code_2);
1179 }
1180
1181 @Test
1182 public void testRequestSizeCheckerr() throws IOException {
1183 final long maxHeapSizePerRequest = 2 * 1024 * 1024;
1184 final ClusterConnection conn = createHConnection();
1185 RequestSizeChecker checker = new RequestSizeChecker(maxHeapSizePerRequest);
1186
1187
1188 for (int i = 0; i != 10; ++i) {
1189 ReturnCode code = checker.canTakeOperation(loc1, maxHeapSizePerRequest);
1190 assertEquals(RowChecker.ReturnCode.INCLUDE, code);
1191 code = checker.canTakeOperation(loc2, maxHeapSizePerRequest);
1192 assertEquals(RowChecker.ReturnCode.INCLUDE, code);
1193 }
1194
1195
1196 ReturnCode acceptCode = checker.canTakeOperation(loc1, maxHeapSizePerRequest);
1197 assertEquals(RowChecker.ReturnCode.INCLUDE, acceptCode);
1198 checker.notifyFinal(acceptCode, loc1, maxHeapSizePerRequest);
1199
1200
1201 for (int i = 0; i != 10; ++i) {
1202 ReturnCode code = checker.canTakeOperation(loc1, maxHeapSizePerRequest);
1203 assertNotEquals(RowChecker.ReturnCode.INCLUDE, code);
1204 code = checker.canTakeOperation(loc2, maxHeapSizePerRequest);
1205 assertNotEquals(RowChecker.ReturnCode.INCLUDE, code);
1206 }
1207
1208
1209 for (int i = 0; i != 10; ++i) {
1210 ReturnCode code = checker.canTakeOperation(loc3, maxHeapSizePerRequest);
1211 assertEquals(RowChecker.ReturnCode.INCLUDE, code);
1212 }
1213
1214 checker.reset();
1215 for (int i = 0; i != 10; ++i) {
1216 ReturnCode code = checker.canTakeOperation(loc1, maxHeapSizePerRequest);
1217 assertEquals(RowChecker.ReturnCode.INCLUDE, code);
1218 code = checker.canTakeOperation(loc2, maxHeapSizePerRequest);
1219 assertEquals(RowChecker.ReturnCode.INCLUDE, code);
1220 }
1221 }
1222
1223 @Test
1224 public void testSubmittedSizeChecker() {
1225 final long maxHeapSizeSubmit = 2 * 1024 * 1024;
1226 SubmittedSizeChecker checker = new SubmittedSizeChecker(maxHeapSizeSubmit);
1227
1228 for (int i = 0; i != 10; ++i) {
1229 ReturnCode include = checker.canTakeOperation(loc1, 100000);
1230 assertEquals(ReturnCode.INCLUDE, include);
1231 }
1232
1233 for (int i = 0; i != 10; ++i) {
1234 checker.notifyFinal(ReturnCode.INCLUDE, loc1, maxHeapSizeSubmit);
1235 }
1236
1237 for (int i = 0; i != 10; ++i) {
1238 ReturnCode include = checker.canTakeOperation(loc1, 100000);
1239 assertEquals(ReturnCode.END, include);
1240 }
1241 for (int i = 0; i != 10; ++i) {
1242 ReturnCode include = checker.canTakeOperation(loc2, 100000);
1243 assertEquals(ReturnCode.END, include);
1244 }
1245 checker.reset();
1246 for (int i = 0; i != 10; ++i) {
1247 ReturnCode include = checker.canTakeOperation(loc1, 100000);
1248 assertEquals(ReturnCode.INCLUDE, include);
1249 }
1250 }
1251 @Test
1252 public void testTaskCountChecker() throws InterruptedIOException {
1253 long rowSize = 12345;
1254 int maxTotalConcurrentTasks = 100;
1255 int maxConcurrentTasksPerServer = 2;
1256 int maxConcurrentTasksPerRegion = 1;
1257 AtomicLong tasksInProgress = new AtomicLong(0);
1258 Map<ServerName, AtomicInteger> taskCounterPerServer = new HashMap<>();
1259 Map<byte[], AtomicInteger> taskCounterPerRegion = new HashMap<>();
1260 TaskCountChecker checker = new TaskCountChecker(
1261 maxTotalConcurrentTasks,
1262 maxConcurrentTasksPerServer,
1263 maxConcurrentTasksPerRegion,
1264 tasksInProgress, taskCounterPerServer, taskCounterPerRegion);
1265
1266
1267 for (int i = 0; i != 10; ++i) {
1268 ReturnCode code = checker.canTakeOperation(loc1, rowSize);
1269 assertEquals(RowChecker.ReturnCode.INCLUDE, code);
1270 }
1271
1272 ReturnCode code = checker.canTakeOperation(loc1, rowSize);
1273 assertEquals(RowChecker.ReturnCode.INCLUDE, code);
1274 checker.notifyFinal(code, loc1, rowSize);
1275
1276
1277 taskCounterPerRegion.put(loc1.getRegionInfo().getRegionName(), new AtomicInteger(100));
1278 taskCounterPerServer.put(loc1.getServerName(), new AtomicInteger(100));
1279
1280
1281 for (int i = 0; i != maxConcurrentTasksPerRegion * 5; ++i) {
1282 ReturnCode includeCode = checker.canTakeOperation(loc1, rowSize);
1283 assertEquals(RowChecker.ReturnCode.INCLUDE, includeCode);
1284 checker.notifyFinal(includeCode, loc1, rowSize);
1285 }
1286
1287
1288 taskCounterPerRegion.put(loc3.getRegionInfo().getRegionName(), new AtomicInteger(100));
1289 taskCounterPerServer.put(loc3.getServerName(), new AtomicInteger(100));
1290
1291
1292 for (int i = 0; i != maxConcurrentTasksPerRegion * 5; ++i) {
1293 ReturnCode excludeCode = checker.canTakeOperation(loc3, rowSize);
1294 assertNotEquals(RowChecker.ReturnCode.INCLUDE, excludeCode);
1295 checker.notifyFinal(excludeCode, loc3, rowSize);
1296 }
1297
1298
1299 taskCounterPerRegion.put(loc3.getRegionInfo().getRegionName(), new AtomicInteger(0));
1300 taskCounterPerServer.put(loc3.getServerName(), new AtomicInteger(0));
1301
1302
1303 ReturnCode code3 = checker.canTakeOperation(loc3, rowSize);
1304 assertEquals(RowChecker.ReturnCode.INCLUDE, code3);
1305 checker.notifyFinal(code3, loc3, rowSize);
1306
1307
1308 for (int i = 0; i != maxConcurrentTasksPerRegion * 5; ++i) {
1309 ReturnCode includeCode = checker.canTakeOperation(loc3, rowSize);
1310 assertEquals(RowChecker.ReturnCode.INCLUDE, includeCode);
1311 checker.notifyFinal(includeCode, loc3, rowSize);
1312 }
1313
1314 checker.reset();
1315
1316
1317
1318 for (int i = 0; i != maxConcurrentTasksPerRegion * 5; ++i) {
1319 ReturnCode includeCode = checker.canTakeOperation(loc1, rowSize);
1320 assertNotEquals(RowChecker.ReturnCode.INCLUDE, includeCode);
1321 checker.notifyFinal(includeCode, loc1, rowSize);
1322 }
1323 }
1324
1325 @Test
1326 public void testBatch() throws IOException, InterruptedException {
1327 ClusterConnection conn = new MyConnectionImpl(conf);
1328 HTable ht = new HTable(conn, new BufferedMutatorParams(DUMMY_TABLE));
1329 ht.multiAp = new MyAsyncProcess(conn, conf, false);
1330
1331 List<Put> puts = new ArrayList<Put>();
1332 puts.add(createPut(1, true));
1333 puts.add(createPut(1, true));
1334 puts.add(createPut(1, true));
1335 puts.add(createPut(1, true));
1336 puts.add(createPut(1, false));
1337 puts.add(createPut(1, true));
1338 puts.add(createPut(1, false));
1339
1340 Object[] res = new Object[puts.size()];
1341 try {
1342 ht.batch(puts, res);
1343 Assert.fail();
1344 } catch (RetriesExhaustedException expected) {
1345 }
1346
1347 Assert.assertEquals(success, res[0]);
1348 Assert.assertEquals(success, res[1]);
1349 Assert.assertEquals(success, res[2]);
1350 Assert.assertEquals(success, res[3]);
1351 Assert.assertEquals(failure, res[4]);
1352 Assert.assertEquals(success, res[5]);
1353 Assert.assertEquals(failure, res[6]);
1354 }
1355 @Test
1356 public void testErrorsServers() throws IOException {
1357 Configuration configuration = new Configuration(conf);
1358 ClusterConnection conn = new MyConnectionImpl(configuration);
1359 BufferedMutatorImpl mutator =
1360 new BufferedMutatorImpl(conn, null, null, new BufferedMutatorParams(DUMMY_TABLE));
1361 configuration.setBoolean(ConnectionManager.RETRIES_BY_SERVER_KEY, true);
1362
1363 MyAsyncProcess ap = new MyAsyncProcess(conn, configuration, true);
1364 mutator.ap = ap;
1365
1366 Assert.assertNotNull(mutator.ap.createServerErrorTracker());
1367 Assert.assertTrue(mutator.ap.serverTrackerTimeout > 200);
1368 mutator.ap.serverTrackerTimeout = 1;
1369
1370 Put p = createPut(1, false);
1371 mutator.mutate(p);
1372
1373 try {
1374 mutator.flush();
1375 Assert.fail();
1376 } catch (RetriesExhaustedWithDetailsException expected) {
1377 }
1378
1379 Assert.assertEquals(NB_RETRIES + 1, ap.callsCt.get());
1380 }
1381
1382 @Test
1383 public void testReadAndWriteTimeout() throws IOException {
1384 final long readTimeout = 10 * 1000;
1385 final long writeTimeout = 20 * 1000;
1386 Configuration copyConf = new Configuration(conf);
1387 copyConf.setLong(HConstants.HBASE_RPC_READ_TIMEOUT_KEY, readTimeout);
1388 copyConf.setLong(HConstants.HBASE_RPC_WRITE_TIMEOUT_KEY, writeTimeout);
1389 ClusterConnection conn = createHConnection();
1390 Mockito.when(conn.getConfiguration()).thenReturn(copyConf);
1391 BufferedMutatorParams bufferParam = new BufferedMutatorParams(DUMMY_TABLE);
1392 try (HTable ht = new HTable(conn, bufferParam)) {
1393 MyAsyncProcess ap = new MyAsyncProcess(conn, copyConf, true);
1394 ht.multiAp = ap;
1395 List<Get> gets = new LinkedList<>();
1396 gets.add(new Get(DUMMY_BYTES_1));
1397 gets.add(new Get(DUMMY_BYTES_2));
1398 try {
1399 ht.get(gets);
1400 } catch (ClassCastException e) {
1401
1402 }
1403 assertEquals(readTimeout, ap.previousTimeout);
1404 ap.previousTimeout = -1;
1405
1406 try {
1407 ht.existsAll(gets);
1408 } catch (ClassCastException e) {
1409
1410 }
1411 assertEquals(readTimeout, ap.previousTimeout);
1412 ap.previousTimeout = -1;
1413
1414 List<Delete> deletes = new LinkedList<>();
1415 deletes.add(new Delete(DUMMY_BYTES_1));
1416 deletes.add(new Delete(DUMMY_BYTES_2));
1417 ht.delete(deletes);
1418 assertEquals(writeTimeout, ap.previousTimeout);
1419 }
1420 }
1421
1422 @Test
1423 public void testGlobalErrors() throws IOException {
1424 ClusterConnection conn = new MyConnectionImpl(conf);
1425 BufferedMutatorImpl mutator = (BufferedMutatorImpl) conn.getBufferedMutator(DUMMY_TABLE);
1426 AsyncProcessWithFailure ap = new AsyncProcessWithFailure(conn, conf, new IOException("test"));
1427 mutator.ap = ap;
1428
1429 Assert.assertNotNull(mutator.ap.createServerErrorTracker());
1430
1431 Put p = createPut(1, true);
1432 mutator.mutate(p);
1433
1434 try {
1435 mutator.flush();
1436 Assert.fail();
1437 } catch (RetriesExhaustedWithDetailsException expected) {
1438 }
1439
1440 Assert.assertEquals(NB_RETRIES + 1, ap.callsCt.get());
1441 }
1442
1443
1444 @Test
1445 public void testCallQueueTooLarge() throws IOException {
1446 ClusterConnection conn = new MyConnectionImpl(conf);
1447 BufferedMutatorImpl mutator = (BufferedMutatorImpl) conn.getBufferedMutator(DUMMY_TABLE);
1448 AsyncProcessWithFailure ap = new AsyncProcessWithFailure(conn, conf, new CallQueueTooBigException());
1449 mutator.ap = ap;
1450
1451 Assert.assertNotNull(mutator.ap.createServerErrorTracker());
1452
1453 Put p = createPut(1, true);
1454 mutator.mutate(p);
1455
1456 try {
1457 mutator.flush();
1458 Assert.fail();
1459 } catch (RetriesExhaustedWithDetailsException expected) {
1460 }
1461
1462 Assert.assertEquals(NB_RETRIES + 1, ap.callsCt.get());
1463 }
1464
1465
1466
1467
1468 @Test
1469 public void testThreadCreation() throws Exception {
1470 final int NB_REGS = 100;
1471 List<HRegionLocation> hrls = new ArrayList<HRegionLocation>(NB_REGS);
1472 List<Get> gets = new ArrayList<Get>(NB_REGS);
1473 for (int i = 0; i < NB_REGS; i++) {
1474 HRegionInfo hri = new HRegionInfo(
1475 DUMMY_TABLE, Bytes.toBytes(i * 10L), Bytes.toBytes(i * 10L + 9L), false, i);
1476 HRegionLocation hrl = new HRegionLocation(hri, i % 2 == 0 ? sn : sn2);
1477 hrls.add(hrl);
1478
1479 Get get = new Get(Bytes.toBytes(i * 10L));
1480 gets.add(get);
1481 }
1482
1483 MyConnectionImpl2 con = new MyConnectionImpl2(hrls);
1484 HTable ht = new HTable(con, new BufferedMutatorParams(DUMMY_TABLE));
1485 MyAsyncProcess ap = new MyAsyncProcess(con, conf, con.nbThreads);
1486 ht.multiAp = ap;
1487
1488 ht.batch(gets, new Object[gets.size()]);
1489
1490 Assert.assertEquals(NB_REGS, ap.nbActions.get());
1491 Assert.assertEquals("1 multi response per server", 2, ap.nbMultiResponse.get());
1492 Assert.assertEquals("1 thread per server", 2, con.nbThreads.get());
1493
1494 int nbReg = 0;
1495 for (int i =0; i<NB_REGS; i++){
1496 if (con.usedRegions[i]) nbReg++;
1497 }
1498 Assert.assertEquals("nbReg=" + nbReg, NB_REGS, nbReg);
1499 }
1500
1501 @Test
1502 public void testReplicaReplicaSuccess() throws Exception {
1503
1504
1505 MyAsyncProcessWithReplicas ap = createReplicaAp(10, 1000, 0);
1506 List<Get> rows = makeTimelineGets(DUMMY_BYTES_1, DUMMY_BYTES_2, DUMMY_BYTES_3);
1507 AsyncRequestFuture ars = ap.submitAll(DUMMY_TABLE, rows, null, new Object[3]);
1508 verifyReplicaResult(ars, RR.TRUE, RR.TRUE, RR.FALSE);
1509 Assert.assertEquals(2, ap.getReplicaCallCount());
1510 }
1511
1512 @Test
1513 public void testReplicaPrimarySuccessWoReplicaCalls() throws Exception {
1514
1515 MyAsyncProcessWithReplicas ap = createReplicaAp(1000, 10, 0);
1516 List<Get> rows = makeTimelineGets(DUMMY_BYTES_1, DUMMY_BYTES_2, DUMMY_BYTES_3);
1517 AsyncRequestFuture ars = ap.submitAll(DUMMY_TABLE, rows, null, new Object[3]);
1518 verifyReplicaResult(ars, RR.FALSE, RR.FALSE, RR.FALSE);
1519 Assert.assertEquals(0, ap.getReplicaCallCount());
1520 }
1521
1522 @Test
1523 public void testReplicaParallelCallsSucceed() throws Exception {
1524
1525 MyAsyncProcessWithReplicas ap = createReplicaAp(0, 0, 0);
1526 List<Get> rows = makeTimelineGets(DUMMY_BYTES_1, DUMMY_BYTES_2);
1527 AsyncRequestFuture ars = ap.submitAll(DUMMY_TABLE, rows, null, new Object[2]);
1528 verifyReplicaResult(ars, RR.DONT_CARE, RR.DONT_CARE);
1529 long replicaCalls = ap.getReplicaCallCount();
1530 Assert.assertTrue(replicaCalls >= 0);
1531 Assert.assertTrue(replicaCalls <= 2);
1532 }
1533
1534 @Test
1535 public void testReplicaPartialReplicaCall() throws Exception {
1536
1537
1538
1539 MyAsyncProcessWithReplicas ap = createReplicaAp(1000, 0, 0);
1540 ap.setPrimaryCallDelay(sn2, 2000);
1541 List<Get> rows = makeTimelineGets(DUMMY_BYTES_1, DUMMY_BYTES_2);
1542 AsyncRequestFuture ars = ap.submitAll(DUMMY_TABLE, rows, null, new Object[2]);
1543 verifyReplicaResult(ars, RR.FALSE, RR.TRUE);
1544 Assert.assertEquals(1, ap.getReplicaCallCount());
1545 }
1546
1547 @Test
1548 public void testReplicaMainFailsBeforeReplicaCalls() throws Exception {
1549
1550
1551
1552 MyAsyncProcessWithReplicas ap = createReplicaAp(1000, 0, 0, 1);
1553 ap.addFailures(hri1, hri2);
1554 List<Get> rows = makeTimelineGets(DUMMY_BYTES_1, DUMMY_BYTES_2);
1555 AsyncRequestFuture ars = ap.submitAll(DUMMY_TABLE, rows, null, new Object[2]);
1556 verifyReplicaResult(ars, RR.FAILED, RR.FAILED);
1557 Assert.assertEquals(0, ap.getReplicaCallCount());
1558 }
1559
1560 @Test
1561 public void testReplicaReplicaSuccessWithParallelFailures() throws Exception {
1562
1563
1564 MyAsyncProcessWithReplicas ap = createReplicaAp(0, 1000, 1000, 1);
1565 ap.addFailures(hri1, hri1r2, hri2);
1566 List<Get> rows = makeTimelineGets(DUMMY_BYTES_1, DUMMY_BYTES_2);
1567 AsyncRequestFuture ars = ap.submitAll(DUMMY_TABLE, rows, null, new Object[2]);
1568 verifyReplicaResult(ars, RR.TRUE, RR.TRUE);
1569 Assert.assertEquals(2, ap.getReplicaCallCount());
1570 }
1571
1572 @Test
1573 public void testReplicaAllCallsFailForOneRegion() throws Exception {
1574
1575
1576 MyAsyncProcessWithReplicas ap = createReplicaAp(500, 1000, 0, 1);
1577 ap.addFailures(hri1, hri1r1, hri1r2, hri2r1);
1578 List<Get> rows = makeTimelineGets(DUMMY_BYTES_1, DUMMY_BYTES_2);
1579 AsyncRequestFuture ars = ap.submitAll(DUMMY_TABLE, rows, null, new Object[2]);
1580 verifyReplicaResult(ars, RR.FAILED, RR.FALSE);
1581
1582 Assert.assertEquals(3, ars.getErrors().getNumExceptions());
1583 for (int i = 0; i < ars.getErrors().getNumExceptions(); ++i) {
1584 Assert.assertArrayEquals(DUMMY_BYTES_1, ars.getErrors().getRow(i).getRow());
1585 }
1586 }
1587
1588 private MyAsyncProcessWithReplicas createReplicaAp(
1589 int replicaAfterMs, int primaryMs, int replicaMs) throws Exception {
1590 return createReplicaAp(replicaAfterMs, primaryMs, replicaMs, -1);
1591 }
1592
1593 private MyAsyncProcessWithReplicas createReplicaAp(
1594 int replicaAfterMs, int primaryMs, int replicaMs, int retries) throws Exception {
1595
1596
1597 Configuration conf = new Configuration();
1598 ClusterConnection conn = createHConnectionWithReplicas();
1599 conf.setInt(AsyncProcess.PRIMARY_CALL_TIMEOUT_KEY, replicaAfterMs * 1000);
1600 if (retries > 0) {
1601 conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, retries);
1602 }
1603 MyAsyncProcessWithReplicas ap = new MyAsyncProcessWithReplicas(conn, conf);
1604 ap.setCallDelays(primaryMs, replicaMs);
1605 return ap;
1606 }
1607
1608 private static List<Get> makeTimelineGets(byte[]... rows) {
1609 List<Get> result = new ArrayList<Get>();
1610 for (byte[] row : rows) {
1611 Get get = new Get(row);
1612 get.setConsistency(Consistency.TIMELINE);
1613 result.add(get);
1614 }
1615 return result;
1616 }
1617
1618 private void verifyResult(AsyncRequestFuture ars, boolean... expected) throws Exception {
1619 Object[] actual = ars.getResults();
1620 Assert.assertEquals(expected.length, actual.length);
1621 for (int i = 0; i < expected.length; ++i) {
1622 Assert.assertEquals(expected[i], !(actual[i] instanceof Throwable));
1623 }
1624 }
1625
1626
1627 private enum RR {
1628 TRUE,
1629 FALSE,
1630 DONT_CARE,
1631 FAILED
1632 }
1633
1634 private void verifyReplicaResult(AsyncRequestFuture ars, RR... expecteds) throws Exception {
1635 Object[] actuals = ars.getResults();
1636 Assert.assertEquals(expecteds.length, actuals.length);
1637 for (int i = 0; i < expecteds.length; ++i) {
1638 Object actual = actuals[i];
1639 RR expected = expecteds[i];
1640 Assert.assertEquals(actual.toString(), expected == RR.FAILED, actual instanceof Throwable);
1641 if (expected != RR.FAILED && expected != RR.DONT_CARE) {
1642 Assert.assertEquals(expected == RR.TRUE, ((Result)actual).isStale());
1643 }
1644 }
1645 }
1646
1647
1648
1649
1650
1651
1652 private Put createPut(int regCnt, boolean success) {
1653 Put p;
1654 if (!success) {
1655 p = new Put(FAILS);
1656 } else switch (regCnt){
1657 case 1 :
1658 p = new Put(DUMMY_BYTES_1);
1659 break;
1660 case 2:
1661 p = new Put(DUMMY_BYTES_2);
1662 break;
1663 case 3:
1664 p = new Put(DUMMY_BYTES_3);
1665 break;
1666 default:
1667 throw new IllegalArgumentException("unknown " + regCnt);
1668 }
1669
1670 p.add(DUMMY_BYTES_1, DUMMY_BYTES_1, DUMMY_BYTES_1);
1671
1672 return p;
1673 }
1674
1675 static class MyThreadPoolExecutor extends ThreadPoolExecutor {
1676 public MyThreadPoolExecutor(int coreThreads, int maxThreads, long keepAliveTime,
1677 TimeUnit timeunit, BlockingQueue<Runnable> blockingqueue) {
1678 super(coreThreads, maxThreads, keepAliveTime, timeunit, blockingqueue);
1679 }
1680
1681 @Override
1682 public Future submit(Runnable runnable) {
1683 throw new OutOfMemoryError("OutOfMemory error thrown by means");
1684 }
1685 }
1686
1687 static class AsyncProcessForThrowableCheck extends AsyncProcess {
1688 private static int rpcTimeout = conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY, HConstants.DEFAULT_HBASE_RPC_TIMEOUT);
1689
1690 public AsyncProcessForThrowableCheck(ClusterConnection hc, Configuration conf,
1691 ExecutorService pool) {
1692 super(hc, conf, pool, new RpcRetryingCallerFactory(conf), false, new RpcControllerFactory(
1693 conf), rpcTimeout);
1694 }
1695 }
1696
1697 @Test
1698 public void testUncheckedException() throws Exception {
1699
1700 ClusterConnection hc = createHConnection();
1701 MyThreadPoolExecutor myPool =
1702 new MyThreadPoolExecutor(1, 20, 60, TimeUnit.SECONDS,
1703 new LinkedBlockingQueue<Runnable>(200));
1704 AsyncProcess ap = new AsyncProcessForThrowableCheck(hc, conf, myPool);
1705
1706 List<Put> puts = new ArrayList<Put>();
1707 puts.add(createPut(1, true));
1708
1709 ap.submit(DUMMY_TABLE, puts, false, null, false);
1710 Assert.assertTrue(puts.isEmpty());
1711 }
1712
1713 @Test
1714 public void testWaitForMaximumCurrentTasks() throws Exception {
1715 final AtomicLong tasks = new AtomicLong(0);
1716 final AtomicInteger max = new AtomicInteger(0);
1717 final CyclicBarrier barrier = new CyclicBarrier(2);
1718 final AsyncProcess ap = new MyAsyncProcess(createHConnection(), conf);
1719 Runnable runnable = new Runnable() {
1720 @Override
1721 public void run() {
1722 try {
1723 barrier.await();
1724 ap.waitForMaximumCurrentTasks(max.get(), tasks, 1, null);
1725 } catch (InterruptedIOException e) {
1726 Assert.fail(e.getMessage());
1727 } catch (InterruptedException e) {
1728
1729 e.printStackTrace();
1730 } catch (BrokenBarrierException e) {
1731
1732 e.printStackTrace();
1733 }
1734 }
1735 };
1736
1737 Thread t = new Thread(runnable);
1738 t.start();
1739 barrier.await();
1740 t.join();
1741
1742 barrier.reset();
1743 tasks.set(1000000);
1744 t = new Thread(runnable);
1745 t.start();
1746 barrier.await();
1747 while (tasks.get() > 0) {
1748 assertTrue(t.isAlive());
1749 tasks.set(tasks.get() - 1);
1750 }
1751 t.join();
1752 }
1753
1754
1755
1756
1757
1758
1759 @Test
1760 public void testRetryPauseWithCallQueueTooBigException() throws Exception {
1761 Configuration myConf = new Configuration(conf);
1762 final long specialPause = 500L;
1763 final int tries = 2;
1764 myConf.setLong(HConstants.HBASE_CLIENT_PAUSE_FOR_CQTBE, specialPause);
1765 myConf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, tries);
1766 ClusterConnection conn = new MyConnectionImpl(myConf);
1767 BufferedMutatorImpl mutator = (BufferedMutatorImpl) conn.getBufferedMutator(DUMMY_TABLE);
1768 AsyncProcessWithFailure ap =
1769 new AsyncProcessWithFailure(conn, myConf, new CallQueueTooBigException());
1770 mutator.ap = ap;
1771
1772 Assert.assertNotNull(mutator.ap.createServerErrorTracker());
1773
1774 Put p = createPut(1, true);
1775 mutator.mutate(p);
1776
1777 long startTime = System.currentTimeMillis();
1778 try {
1779 mutator.flush();
1780 Assert.fail();
1781 } catch (RetriesExhaustedWithDetailsException expected) {
1782 }
1783 long actualSleep = System.currentTimeMillis() - startTime;
1784 long expectedSleep = 0L;
1785 for (int i = 0; i < tries - 1; i++) {
1786 expectedSleep += ConnectionUtils.getPauseTime(specialPause, i);
1787
1788 actualSleep += (long) (specialPause * 0.01f);
1789 }
1790 LOG.debug("Expected to sleep " + expectedSleep + "ms, actually slept " + actualSleep + "ms");
1791 Assert.assertTrue("Expected to sleep " + expectedSleep + " but actually " + actualSleep + "ms",
1792 actualSleep >= expectedSleep);
1793
1794
1795 final long normalPause =
1796 myConf.getLong(HConstants.HBASE_CLIENT_PAUSE, HConstants.DEFAULT_HBASE_CLIENT_PAUSE);
1797 ap = new AsyncProcessWithFailure(conn, myConf, new IOException());
1798 mutator.ap = ap;
1799 Assert.assertNotNull(mutator.ap.createServerErrorTracker());
1800 mutator.mutate(p);
1801 startTime = System.currentTimeMillis();
1802 try {
1803 mutator.flush();
1804 Assert.fail();
1805 } catch (RetriesExhaustedWithDetailsException expected) {
1806 }
1807 actualSleep = System.currentTimeMillis() - startTime;
1808 expectedSleep = 0L;
1809 for (int i = 0; i < tries - 1; i++) {
1810 expectedSleep += ConnectionUtils.getPauseTime(normalPause, i);
1811 }
1812
1813 expectedSleep += normalPause;
1814 LOG.debug("Expected to sleep " + expectedSleep + "ms, actually slept " + actualSleep + "ms");
1815 Assert.assertTrue("Slept for too long: " + actualSleep + "ms", actualSleep <= expectedSleep);
1816 }
1817
1818 @Test
1819 public void testQueueRowAccess() throws Exception {
1820 ClusterConnection conn = createHConnection();
1821 BufferedMutatorImpl mutator = new BufferedMutatorImpl(conn, null, null,
1822 new BufferedMutatorParams(DUMMY_TABLE).writeBufferSize(100000));
1823 Put p0 = new Put(DUMMY_BYTES_1).addColumn(DUMMY_BYTES_1, DUMMY_BYTES_1, DUMMY_BYTES_1);
1824 Put p1 = new Put(DUMMY_BYTES_2).addColumn(DUMMY_BYTES_2, DUMMY_BYTES_2, DUMMY_BYTES_2);
1825 mutator.mutate(p0);
1826 BufferedMutatorImpl.QueueRowAccess ra0 = mutator.createQueueRowAccess();
1827
1828 assertEquals(0, mutator.undealtMutationCount.get());
1829 mutator.mutate(p1);
1830 assertEquals(1, mutator.undealtMutationCount.get());
1831 BufferedMutatorImpl.QueueRowAccess ra1 = mutator.createQueueRowAccess();
1832
1833 assertEquals(0, mutator.undealtMutationCount.get());
1834 assertEquals(1, ra0.size());
1835 assertEquals(1, ra1.size());
1836 Iterator<Row> iter0 = ra0.iterator();
1837 Iterator<Row> iter1 = ra1.iterator();
1838 assertTrue(iter0.hasNext());
1839 assertTrue(iter1.hasNext());
1840
1841 assertTrue(iter0.next() == p0);
1842 assertEquals(1, mutator.writeAsyncBuffer.size());
1843 assertEquals(p1.heapSize(), mutator.currentWriteBufferSize.get());
1844 assertTrue(iter1.next() == p1);
1845 assertEquals(0, mutator.writeAsyncBuffer.size());
1846 assertEquals(0, mutator.currentWriteBufferSize.get());
1847 assertFalse(iter0.hasNext());
1848 assertFalse(iter1.hasNext());
1849
1850 iter0.remove();
1851 ra0.close();
1852 assertEquals(0, mutator.undealtMutationCount.get());
1853 assertEquals(0, mutator.writeAsyncBuffer.size());
1854 assertEquals(0, mutator.currentWriteBufferSize.get());
1855
1856 ra1.close();
1857 assertEquals(1, mutator.undealtMutationCount.get());
1858 assertEquals(1, mutator.writeAsyncBuffer.size());
1859 assertEquals(p1.heapSize(), mutator.currentWriteBufferSize.get());
1860 }
1861 }