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
23 import static org.junit.Assert.assertTrue;
24
25 import java.io.IOException;
26 import java.io.InterruptedIOException;
27 import java.util.ArrayList;
28 import java.util.Arrays;
29 import java.util.HashMap;
30 import java.util.List;
31 import java.util.Map;
32 import java.util.Set;
33 import java.util.TreeSet;
34 import java.util.concurrent.BlockingQueue;
35 import java.util.concurrent.BrokenBarrierException;
36 import java.util.concurrent.CyclicBarrier;
37 import java.util.concurrent.ExecutorService;
38 import java.util.concurrent.Future;
39 import java.util.concurrent.LinkedBlockingQueue;
40 import java.util.concurrent.RejectedExecutionException;
41 import java.util.concurrent.SynchronousQueue;
42 import java.util.concurrent.ThreadFactory;
43 import java.util.concurrent.ThreadPoolExecutor;
44 import java.util.concurrent.TimeUnit;
45 import java.util.concurrent.atomic.AtomicBoolean;
46 import java.util.concurrent.atomic.AtomicInteger;
47 import java.util.concurrent.atomic.AtomicLong;
48
49 import org.apache.commons.logging.Log;
50 import org.apache.commons.logging.LogFactory;
51 import org.apache.hadoop.conf.Configuration;
52 import org.apache.hadoop.hbase.CallQueueTooBigException;
53 import org.apache.hadoop.hbase.Cell;
54 import org.apache.hadoop.hbase.HConstants;
55 import org.apache.hadoop.hbase.HRegionInfo;
56 import org.apache.hadoop.hbase.HRegionLocation;
57 import org.apache.hadoop.hbase.RegionLocations;
58 import org.apache.hadoop.hbase.ServerName;
59 import org.apache.hadoop.hbase.TableName;
60 import org.apache.hadoop.hbase.client.AsyncProcess.AsyncRequestFuture;
61 import org.apache.hadoop.hbase.client.coprocessor.Batch;
62 import org.apache.hadoop.hbase.client.coprocessor.Batch.Callback;
63 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
64 import org.apache.hadoop.hbase.testclassification.MediumTests;
65 import org.apache.hadoop.hbase.util.Bytes;
66 import org.apache.hadoop.hbase.util.Threads;
67 import org.junit.Assert;
68 import org.junit.BeforeClass;
69 import org.junit.Rule;
70 import org.junit.Test;
71 import org.junit.experimental.categories.Category;
72 import org.junit.rules.Timeout;
73 import org.mockito.Mockito;
74
75 @Category(MediumTests.class)
76 public class TestAsyncProcess {
77 private final static Log LOG = LogFactory.getLog(TestAsyncProcess.class);
78 private static final TableName DUMMY_TABLE =
79 TableName.valueOf("DUMMY_TABLE");
80 private static final byte[] DUMMY_BYTES_1 = "DUMMY_BYTES_1".getBytes();
81 private static final byte[] DUMMY_BYTES_2 = "DUMMY_BYTES_2".getBytes();
82 private static final byte[] DUMMY_BYTES_3 = "DUMMY_BYTES_3".getBytes();
83 private static final byte[] FAILS = "FAILS".getBytes();
84 private static final Configuration conf = new Configuration();
85
86 private static ServerName sn = ServerName.valueOf("s1:1,1");
87 private static ServerName sn2 = ServerName.valueOf("s2:2,2");
88 private static ServerName sn3 = ServerName.valueOf("s3:3,3");
89 private static HRegionInfo hri1 =
90 new HRegionInfo(DUMMY_TABLE, DUMMY_BYTES_1, DUMMY_BYTES_2, false, 1);
91 private static HRegionInfo hri2 =
92 new HRegionInfo(DUMMY_TABLE, DUMMY_BYTES_2, HConstants.EMPTY_END_ROW, false, 2);
93 private static HRegionInfo hri3 =
94 new HRegionInfo(DUMMY_TABLE, DUMMY_BYTES_3, HConstants.EMPTY_END_ROW, false, 3);
95 private static HRegionLocation loc1 = new HRegionLocation(hri1, sn);
96 private static HRegionLocation loc2 = new HRegionLocation(hri2, sn);
97 private static HRegionLocation loc3 = new HRegionLocation(hri3, sn2);
98
99
100 private static HRegionInfo hri1r1 = RegionReplicaUtil.getRegionInfoForReplica(hri1, 1),
101 hri1r2 = RegionReplicaUtil.getRegionInfoForReplica(hri1, 2);
102 private static HRegionInfo hri2r1 = RegionReplicaUtil.getRegionInfoForReplica(hri2, 1);
103 private static RegionLocations hrls1 = new RegionLocations(new HRegionLocation(hri1, sn),
104 new HRegionLocation(hri1r1, sn2), new HRegionLocation(hri1r2, sn3));
105 private static RegionLocations hrls2 = new RegionLocations(new HRegionLocation(hri2, sn2),
106 new HRegionLocation(hri2r1, sn3));
107 private static RegionLocations hrls3 = new RegionLocations(new HRegionLocation(hri3, sn3), null);
108
109 private static final String success = "success";
110 private static Exception failure = new Exception("failure");
111
112 private static int NB_RETRIES = 3;
113
114 @BeforeClass
115 public static void beforeClass(){
116 conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, NB_RETRIES);
117 }
118
119 static class CountingThreadFactory implements ThreadFactory {
120 final AtomicInteger nbThreads;
121 ThreadFactory realFactory = Threads.newDaemonThreadFactory("test-TestAsyncProcess");
122 @Override
123 public Thread newThread(Runnable r) {
124 nbThreads.incrementAndGet();
125 return realFactory.newThread(r);
126 }
127
128 CountingThreadFactory(AtomicInteger nbThreads){
129 this.nbThreads = nbThreads;
130 }
131 }
132
133 static class MyAsyncProcess extends AsyncProcess {
134 final AtomicInteger nbMultiResponse = new AtomicInteger();
135 final AtomicInteger nbActions = new AtomicInteger();
136 public List<AsyncRequestFuture> allReqs = new ArrayList<AsyncRequestFuture>();
137 public AtomicInteger callsCt = new AtomicInteger();
138
139 @Override
140 protected <Res> AsyncRequestFutureImpl<Res> createAsyncRequestFuture(TableName tableName,
141 List<Action<Row>> actions, long nonceGroup, ExecutorService pool,
142 Batch.Callback<Res> callback, Object[] results, boolean needResults) {
143
144 AsyncRequestFutureImpl<Res> r = super.createAsyncRequestFuture(
145 DUMMY_TABLE, actions, nonceGroup, pool, callback, results, needResults);
146 allReqs.add(r);
147 callsCt.incrementAndGet();
148 return r;
149 }
150
151 public MyAsyncProcess(ClusterConnection hc, Configuration conf) {
152 this(hc, conf, new AtomicInteger());
153 }
154
155 public MyAsyncProcess(ClusterConnection hc, Configuration conf, AtomicInteger nbThreads) {
156 super(hc, conf, new ThreadPoolExecutor(1, 20, 60, TimeUnit.SECONDS,
157 new SynchronousQueue<Runnable>(), new CountingThreadFactory(nbThreads)),
158 new RpcRetryingCallerFactory(conf), false, new RpcControllerFactory(conf),
159 conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
160 HConstants.DEFAULT_HBASE_RPC_TIMEOUT));
161 }
162
163 public MyAsyncProcess(
164 ClusterConnection hc, Configuration conf, boolean useGlobalErrors) {
165 super(hc, conf, new ThreadPoolExecutor(1, 20, 60, TimeUnit.SECONDS,
166 new SynchronousQueue<Runnable>(), new CountingThreadFactory(new AtomicInteger())),
167 new RpcRetryingCallerFactory(conf), useGlobalErrors, new RpcControllerFactory(conf),
168 conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
169 HConstants.DEFAULT_HBASE_RPC_TIMEOUT));
170 }
171
172 public MyAsyncProcess(ClusterConnection hc, Configuration conf, boolean useGlobalErrors,
173 @SuppressWarnings("unused") boolean dummy) {
174 super(hc, conf, new ThreadPoolExecutor(1, 20, 60, TimeUnit.SECONDS,
175 new SynchronousQueue<Runnable>(), new CountingThreadFactory(new AtomicInteger())) {
176 @Override
177 public void execute(Runnable command) {
178 throw new RejectedExecutionException("test under failure");
179 }
180 },
181 new RpcRetryingCallerFactory(conf), useGlobalErrors, new RpcControllerFactory(conf),
182 conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
183 HConstants.DEFAULT_HBASE_RPC_TIMEOUT));
184 }
185
186 @Override
187 public <Res> AsyncRequestFuture submit(TableName tableName, List<? extends Row> rows,
188 boolean atLeastOne, Callback<Res> callback, boolean needResults)
189 throws InterruptedIOException {
190
191 return super.submit(DUMMY_TABLE, rows, atLeastOne, callback, true);
192 }
193
194 @Override
195 protected RpcRetryingCaller<MultiResponse> createCaller(
196 PayloadCarryingServerCallable callable, int rpcTimeout) {
197 callsCt.incrementAndGet();
198 MultiServerCallable callable1 = (MultiServerCallable) callable;
199 final MultiResponse mr = createMultiResponse(
200 callable1.getMulti(), nbMultiResponse, nbActions, new ResponseGenerator() {
201 @Override
202 public void addResponse(MultiResponse mr, byte[] regionName, Action<Row> a) {
203 if (Arrays.equals(FAILS, a.getAction().getRow())) {
204 mr.add(regionName, a.getOriginalIndex(), failure);
205 } else {
206 mr.add(regionName, a.getOriginalIndex(), success);
207 }
208 }
209 });
210
211 return new RpcRetryingCaller<MultiResponse>(100, 10, 9) {
212 @Override
213 public MultiResponse callWithoutRetries(RetryingCallable<MultiResponse> callable,
214 int callTimeout)
215 throws IOException, RuntimeException {
216 try {
217
218
219 Thread.sleep(1000);
220 } catch (InterruptedException e) {
221
222 }
223 return mr;
224 }
225 };
226 }
227 }
228
229 static class CallerWithFailure extends RpcRetryingCaller<MultiResponse>{
230
231 private final IOException e;
232
233 public CallerWithFailure(IOException e) {
234 super(100, 100, 9);
235 this.e = e;
236 }
237
238 @Override
239 public MultiResponse callWithoutRetries(RetryingCallable<MultiResponse> callable,
240 int callTimeout)
241 throws IOException, RuntimeException {
242 throw e;
243 }
244 }
245
246
247 static class AsyncProcessWithFailure extends MyAsyncProcess {
248
249 private final IOException ioe;
250
251 public AsyncProcessWithFailure(ClusterConnection hc, Configuration conf, IOException ioe) {
252 super(hc, conf, true);
253 this.ioe = ioe;
254 serverTrackerTimeout = 1;
255 }
256
257 @Override
258 protected RpcRetryingCaller<MultiResponse> createCaller(
259 PayloadCarryingServerCallable callable, int rpcTimeout) {
260 callsCt.incrementAndGet();
261 return new CallerWithFailure(ioe);
262 }
263 }
264
265 class MyAsyncProcessWithReplicas extends MyAsyncProcess {
266 private Set<byte[]> failures = new TreeSet<byte[]>(new Bytes.ByteArrayComparator());
267 private long primarySleepMs = 0, replicaSleepMs = 0;
268 private Map<ServerName, Long> customPrimarySleepMs = new HashMap<ServerName, Long>();
269 private final AtomicLong replicaCalls = new AtomicLong(0);
270
271 public void addFailures(HRegionInfo... hris) {
272 for (HRegionInfo hri : hris) {
273 failures.add(hri.getRegionName());
274 }
275 }
276
277 public long getReplicaCallCount() {
278 return replicaCalls.get();
279 }
280
281 public void setPrimaryCallDelay(ServerName server, long primaryMs) {
282 customPrimarySleepMs.put(server, primaryMs);
283 }
284
285 public MyAsyncProcessWithReplicas(ClusterConnection hc, Configuration conf) {
286 super(hc, conf);
287 }
288
289 public void setCallDelays(long primaryMs, long replicaMs) {
290 this.primarySleepMs = primaryMs;
291 this.replicaSleepMs = replicaMs;
292 }
293
294 @Override
295 protected RpcRetryingCaller<MultiResponse> createCaller(
296 PayloadCarryingServerCallable payloadCallable, int rpcTimeout) {
297 MultiServerCallable<Row> callable = (MultiServerCallable) payloadCallable;
298 final MultiResponse mr = createMultiResponse(
299 callable.getMulti(), nbMultiResponse, nbActions, new ResponseGenerator() {
300 @Override
301 public void addResponse(MultiResponse mr, byte[] regionName, Action<Row> a) {
302 if (failures.contains(regionName)) {
303 mr.add(regionName, a.getOriginalIndex(), failure);
304 } else {
305 boolean isStale = !RegionReplicaUtil.isDefaultReplica(a.getReplicaId());
306 mr.add(regionName, a.getOriginalIndex(),
307 Result.create(new Cell[0], null, isStale));
308 }
309 }
310 });
311
312 final boolean isDefault = RegionReplicaUtil.isDefaultReplica(
313 callable.getMulti().actions.values().iterator().next().iterator().next().getReplicaId());
314 final ServerName server = ((MultiServerCallable<?>)callable).getServerName();
315 String debugMsg = "Call to " + server + ", primary=" + isDefault + " with "
316 + callable.getMulti().actions.size() + " entries: ";
317 for (byte[] region : callable.getMulti().actions.keySet()) {
318 debugMsg += "[" + Bytes.toStringBinary(region) + "], ";
319 }
320 LOG.debug(debugMsg);
321 if (!isDefault) {
322 replicaCalls.incrementAndGet();
323 }
324
325 return new RpcRetryingCaller<MultiResponse>(100, 10, 9) {
326 @Override
327 public MultiResponse callWithoutRetries(RetryingCallable<MultiResponse> callable,
328 int callTimeout)
329 throws IOException, RuntimeException {
330 long sleep = -1;
331 if (isDefault) {
332 Long customSleep = customPrimarySleepMs.get(server);
333 sleep = (customSleep == null ? primarySleepMs : customSleep.longValue());
334 } else {
335 sleep = replicaSleepMs;
336 }
337 if (sleep != 0) {
338 try {
339 Thread.sleep(sleep);
340 } catch (InterruptedException e) {
341 }
342 }
343 return mr;
344 }
345 };
346 }
347 }
348
349 static MultiResponse createMultiResponse(final MultiAction<Row> multi,
350 AtomicInteger nbMultiResponse, AtomicInteger nbActions, ResponseGenerator gen) {
351 final MultiResponse mr = new MultiResponse();
352 nbMultiResponse.incrementAndGet();
353 for (Map.Entry<byte[], List<Action<Row>>> entry : multi.actions.entrySet()) {
354 byte[] regionName = entry.getKey();
355 for (Action<Row> a : entry.getValue()) {
356 nbActions.incrementAndGet();
357 gen.addResponse(mr, regionName, a);
358 }
359 }
360 return mr;
361 }
362
363 private static interface ResponseGenerator {
364 void addResponse(final MultiResponse mr, byte[] regionName, Action<Row> a);
365 }
366
367
368
369
370 static class MyConnectionImpl extends ConnectionManager.HConnectionImplementation {
371 public static class TestRegistry implements Registry {
372 @Override
373 public void init(Connection connection) {}
374
375 @Override
376 public RegionLocations getMetaRegionLocation() throws IOException {
377 return null;
378 }
379
380 @Override
381 public String getClusterId() {
382 return "testClusterId";
383 }
384
385 @Override
386 public boolean isTableOnlineState(TableName tableName, boolean enabled) throws IOException {
387 return false;
388 }
389
390 @Override
391 public int getCurrentNrHRS() throws IOException {
392 return 1;
393 }
394 }
395
396 final AtomicInteger nbThreads = new AtomicInteger(0);
397
398 protected MyConnectionImpl(Configuration conf) throws IOException {
399 super(setupConf(conf), false);
400 }
401
402 private static Configuration setupConf(Configuration conf) {
403 conf.setClass(RegistryFactory.REGISTRY_IMPL_CONF_KEY, TestRegistry.class, Registry.class);
404 return conf;
405 }
406
407 @Override
408 public RegionLocations locateRegion(TableName tableName,
409 byte[] row, boolean useCache, boolean retry, int replicaId) throws IOException {
410 return new RegionLocations(loc1);
411 }
412
413 @Override
414 public boolean hasCellBlockSupport() {
415 return false;
416 }
417 }
418
419
420
421
422 static class MyConnectionImpl2 extends MyConnectionImpl {
423 List<HRegionLocation> hrl;
424 final boolean usedRegions[];
425
426 protected MyConnectionImpl2(List<HRegionLocation> hrl) throws IOException {
427 super(conf);
428 this.hrl = hrl;
429 this.usedRegions = new boolean[hrl.size()];
430 }
431
432 @Override
433 public RegionLocations locateRegion(TableName tableName,
434 byte[] row, boolean useCache, boolean retry, int replicaId) throws IOException {
435 int i = 0;
436 for (HRegionLocation hr : hrl){
437 if (Arrays.equals(row, hr.getRegionInfo().getStartKey())) {
438 usedRegions[i] = true;
439 return new RegionLocations(hr);
440 }
441 i++;
442 }
443 return null;
444 }
445 }
446
447 @Rule
448 public Timeout timeout = Timeout.millis(10000);
449
450 @Test
451 public void testSubmit() throws Exception {
452 ClusterConnection hc = createHConnection();
453 AsyncProcess ap = new MyAsyncProcess(hc, conf);
454
455 List<Put> puts = new ArrayList<Put>();
456 puts.add(createPut(1, true));
457
458 ap.submit(DUMMY_TABLE, puts, false, null, false);
459 Assert.assertTrue(puts.isEmpty());
460 }
461
462 @Test
463 public void testSubmitWithCB() throws Exception {
464 ClusterConnection hc = createHConnection();
465 final AtomicInteger updateCalled = new AtomicInteger(0);
466 Batch.Callback<Object> cb = new Batch.Callback<Object>() {
467 @Override
468 public void update(byte[] region, byte[] row, Object result) {
469 updateCalled.incrementAndGet();
470 }
471 };
472 AsyncProcess ap = new MyAsyncProcess(hc, conf);
473
474 List<Put> puts = new ArrayList<Put>();
475 puts.add(createPut(1, true));
476
477 final AsyncRequestFuture ars = ap.submit(DUMMY_TABLE, puts, false, cb, false);
478 Assert.assertTrue(puts.isEmpty());
479 ars.waitUntilDone();
480 Assert.assertEquals(updateCalled.get(), 1);
481 }
482
483 @Test
484 public void testSubmitBusyRegion() throws Exception {
485 ClusterConnection hc = createHConnection();
486 AsyncProcess ap = new MyAsyncProcess(hc, conf);
487
488 List<Put> puts = new ArrayList<Put>();
489 puts.add(createPut(1, true));
490
491 ap.incTaskCounters(Arrays.asList(hri1.getRegionName()), sn);
492 ap.submit(DUMMY_TABLE, puts, false, null, false);
493 Assert.assertEquals(puts.size(), 1);
494
495 ap.decTaskCounters(Arrays.asList(hri1.getRegionName()), sn);
496 ap.submit(DUMMY_TABLE, puts, false, null, false);
497 Assert.assertEquals(0, puts.size());
498 }
499
500
501 @Test
502 public void testSubmitBusyRegionServer() throws Exception {
503 ClusterConnection hc = createHConnection();
504 AsyncProcess ap = new MyAsyncProcess(hc, conf);
505
506 ap.taskCounterPerServer.put(sn2, new AtomicInteger(ap.maxConcurrentTasksPerServer));
507
508 List<Put> puts = new ArrayList<Put>();
509 puts.add(createPut(1, true));
510 puts.add(createPut(3, true));
511 puts.add(createPut(1, true));
512 puts.add(createPut(2, true));
513
514 ap.submit(DUMMY_TABLE, puts, false, null, false);
515 Assert.assertEquals(" puts=" + puts, 1, puts.size());
516
517 ap.taskCounterPerServer.put(sn2, new AtomicInteger(ap.maxConcurrentTasksPerServer - 1));
518 ap.submit(DUMMY_TABLE, puts, false, null, false);
519 Assert.assertTrue(puts.isEmpty());
520 }
521
522 @Test
523 public void testFail() throws Exception {
524 MyAsyncProcess ap = new MyAsyncProcess(createHConnection(), conf, false);
525
526 List<Put> puts = new ArrayList<Put>();
527 Put p = createPut(1, false);
528 puts.add(p);
529
530 AsyncRequestFuture ars = ap.submit(DUMMY_TABLE, puts, false, null, true);
531 Assert.assertEquals(0, puts.size());
532 ars.waitUntilDone();
533 verifyResult(ars, false);
534 Assert.assertEquals(NB_RETRIES + 1, ap.callsCt.get());
535
536 Assert.assertEquals(1, ars.getErrors().exceptions.size());
537 Assert.assertTrue("was: " + ars.getErrors().exceptions.get(0),
538 failure.equals(ars.getErrors().exceptions.get(0)));
539 Assert.assertTrue("was: " + ars.getErrors().exceptions.get(0),
540 failure.equals(ars.getErrors().exceptions.get(0)));
541
542 Assert.assertEquals(1, ars.getFailedOperations().size());
543 Assert.assertTrue("was: " + ars.getFailedOperations().get(0),
544 p.equals(ars.getFailedOperations().get(0)));
545 }
546
547
548 @Test
549 public void testSubmitTrue() throws IOException {
550 final AsyncProcess ap = new MyAsyncProcess(createHConnection(), conf, false);
551 ap.tasksInProgress.incrementAndGet();
552 final AtomicInteger ai = new AtomicInteger(1);
553 ap.taskCounterPerRegion.put(hri1.getRegionName(), ai);
554
555 final AtomicBoolean checkPoint = new AtomicBoolean(false);
556 final AtomicBoolean checkPoint2 = new AtomicBoolean(false);
557
558 Thread t = new Thread(){
559 @Override
560 public void run(){
561 Threads.sleep(1000);
562 Assert.assertFalse(checkPoint.get());
563 ai.decrementAndGet();
564 ap.tasksInProgress.decrementAndGet();
565 checkPoint2.set(true);
566 }
567 };
568
569 List<Put> puts = new ArrayList<Put>();
570 Put p = createPut(1, true);
571 puts.add(p);
572
573 ap.submit(DUMMY_TABLE, puts, false, null, false);
574 Assert.assertFalse(puts.isEmpty());
575
576 t.start();
577
578 ap.submit(DUMMY_TABLE, puts, true, null, false);
579 Assert.assertTrue(puts.isEmpty());
580
581 checkPoint.set(true);
582 while (!checkPoint2.get()){
583 Threads.sleep(1);
584 }
585 }
586
587 @Test
588 public void testFailAndSuccess() throws Exception {
589 MyAsyncProcess ap = new MyAsyncProcess(createHConnection(), conf, false);
590
591 List<Put> puts = new ArrayList<Put>();
592 puts.add(createPut(1, false));
593 puts.add(createPut(1, true));
594 puts.add(createPut(1, true));
595
596 AsyncRequestFuture ars = ap.submit(DUMMY_TABLE, puts, false, null, true);
597 Assert.assertTrue(puts.isEmpty());
598 ars.waitUntilDone();
599 verifyResult(ars, false, true, true);
600 Assert.assertEquals(NB_RETRIES + 1, ap.callsCt.get());
601 ap.callsCt.set(0);
602 Assert.assertEquals(1, ars.getErrors().actions.size());
603
604 puts.add(createPut(1, true));
605
606 ap.waitUntilDone();
607 ars = ap.submit(DUMMY_TABLE, puts, false, null, true);
608 Assert.assertEquals(0, puts.size());
609 ars.waitUntilDone();
610 Assert.assertEquals(2, ap.callsCt.get());
611 verifyResult(ars, true);
612 }
613
614 @Test
615 public void testFlush() throws Exception {
616 MyAsyncProcess ap = new MyAsyncProcess(createHConnection(), conf, false);
617
618 List<Put> puts = new ArrayList<Put>();
619 puts.add(createPut(1, false));
620 puts.add(createPut(1, true));
621 puts.add(createPut(1, true));
622
623 AsyncRequestFuture ars = ap.submit(DUMMY_TABLE, puts, false, null, true);
624 ars.waitUntilDone();
625 verifyResult(ars, false, true, true);
626 Assert.assertEquals(NB_RETRIES + 1, ap.callsCt.get());
627
628 Assert.assertEquals(1, ars.getFailedOperations().size());
629 }
630
631 @Test
632 public void testMaxTask() throws Exception {
633 final AsyncProcess ap = new MyAsyncProcess(createHConnection(), conf, false);
634
635 for (int i = 0; i < 1000; i++) {
636 ap.incTaskCounters(Arrays.asList("dummy".getBytes()), sn);
637 }
638
639 final Thread myThread = Thread.currentThread();
640
641 Thread t = new Thread() {
642 @Override
643 public void run() {
644 Threads.sleep(2000);
645 myThread.interrupt();
646 }
647 };
648
649 List<Put> puts = new ArrayList<Put>();
650 puts.add(createPut(1, true));
651
652 t.start();
653
654 try {
655 ap.submit(DUMMY_TABLE, puts, false, null, false);
656 Assert.fail("We should have been interrupted.");
657 } catch (InterruptedIOException expected) {
658 }
659
660 final long sleepTime = 2000;
661
662 Thread t2 = new Thread() {
663 @Override
664 public void run() {
665 Threads.sleep(sleepTime);
666 while (ap.tasksInProgress.get() > 0) {
667 ap.decTaskCounters(Arrays.asList("dummy".getBytes()), sn);
668 }
669 }
670 };
671 t2.start();
672
673 long start = System.currentTimeMillis();
674 ap.submit(DUMMY_TABLE, new ArrayList<Row>(), false, null, false);
675 long end = System.currentTimeMillis();
676
677
678 Assert.assertTrue(start + 100L + sleepTime > end);
679 }
680
681 private static ClusterConnection createHConnection() throws IOException {
682 ClusterConnection hc = createHConnectionCommon();
683 setMockLocation(hc, DUMMY_BYTES_1, new RegionLocations(loc1));
684 setMockLocation(hc, DUMMY_BYTES_2, new RegionLocations(loc2));
685 setMockLocation(hc, DUMMY_BYTES_3, new RegionLocations(loc3));
686 setMockLocation(hc, FAILS, new RegionLocations(loc2));
687 return hc;
688 }
689
690 private static ClusterConnection createHConnectionWithReplicas() throws IOException {
691 ClusterConnection hc = createHConnectionCommon();
692 setMockLocation(hc, DUMMY_BYTES_1, hrls1);
693 setMockLocation(hc, DUMMY_BYTES_2, hrls2);
694 setMockLocation(hc, DUMMY_BYTES_3, hrls3);
695 return hc;
696 }
697
698 private static void setMockLocation(ClusterConnection hc, byte[] row,
699 RegionLocations result) throws IOException {
700 Mockito.when(hc.locateRegion(Mockito.eq(DUMMY_TABLE), Mockito.eq(row),
701 Mockito.anyBoolean(), Mockito.anyBoolean(), Mockito.anyInt())).thenReturn(result);
702 }
703
704 private static ClusterConnection createHConnectionCommon() {
705 ClusterConnection hc = Mockito.mock(ClusterConnection.class);
706 NonceGenerator ng = Mockito.mock(NonceGenerator.class);
707 Mockito.when(ng.getNonceGroup()).thenReturn(HConstants.NO_NONCE);
708 Mockito.when(hc.getNonceGenerator()).thenReturn(ng);
709 Mockito.when(hc.getConfiguration()).thenReturn(conf);
710 return hc;
711 }
712
713 @Test
714 public void testHTablePutSuccess() throws Exception {
715 BufferedMutatorImpl ht = Mockito.mock(BufferedMutatorImpl.class);
716 ht.ap = new MyAsyncProcess(createHConnection(), conf, true);
717
718 Put put = createPut(1, true);
719
720 Assert.assertEquals(0, ht.getWriteBufferSize());
721 ht.mutate(put);
722 Assert.assertEquals(0, ht.getWriteBufferSize());
723 }
724
725 private void doHTableFailedPut(boolean bufferOn) throws Exception {
726 ClusterConnection conn = createHConnection();
727 HTable ht = new HTable(conn, new BufferedMutatorParams(DUMMY_TABLE));
728 MyAsyncProcess ap = new MyAsyncProcess(conn, conf, true);
729 ht.mutator.ap = ap;
730 if (bufferOn) {
731 ht.setWriteBufferSize(1024L * 1024L);
732 } else {
733 ht.setWriteBufferSize(0L);
734 }
735
736 Put put = createPut(1, false);
737
738 Assert.assertEquals(0L, ht.mutator.currentWriteBufferSize.get());
739 try {
740 ht.put(put);
741 if (bufferOn) {
742 ht.flushCommits();
743 }
744 Assert.fail();
745 } catch (RetriesExhaustedException expected) {
746 }
747 Assert.assertEquals(0L, ht.mutator.currentWriteBufferSize.get());
748
749 AsyncRequestFuture ars = null;
750 for (AsyncRequestFuture someReqs : ap.allReqs) {
751 if (someReqs.getResults().length == 0) continue;
752 Assert.assertTrue(ars == null);
753 ars = someReqs;
754 }
755 Assert.assertTrue(ars != null);
756 verifyResult(ars, false);
757
758
759 ht.close();
760 }
761
762 @Test
763 public void testHTableFailedPutWithBuffer() throws Exception {
764 doHTableFailedPut(true);
765 }
766
767 @Test
768 public void testHTableFailedPutWithoutBuffer() throws Exception {
769 doHTableFailedPut(false);
770 }
771
772 @Test
773 public void testHTableFailedPutAndNewPut() throws Exception {
774 ClusterConnection conn = createHConnection();
775 BufferedMutatorImpl mutator = new BufferedMutatorImpl(conn, null, null,
776 new BufferedMutatorParams(DUMMY_TABLE).writeBufferSize(0));
777 MyAsyncProcess ap = new MyAsyncProcess(conn, conf, true);
778 mutator.ap = ap;
779
780 Put p = createPut(1, false);
781 mutator.mutate(p);
782
783 ap.waitUntilDone();
784
785
786
787
788
789
790 p = createPut(1, true);
791 Assert.assertEquals(0, mutator.getWriteBuffer().size());
792 try {
793 mutator.mutate(p);
794 Assert.fail();
795 } catch (RetriesExhaustedException expected) {
796 }
797 Assert.assertEquals("the put should not been inserted.", 0, mutator.getWriteBuffer().size());
798 }
799
800 @Test
801 public void testBatch() throws IOException, InterruptedException {
802 ClusterConnection conn = new MyConnectionImpl(conf);
803 HTable ht = new HTable(conn, new BufferedMutatorParams(DUMMY_TABLE));
804 ht.multiAp = new MyAsyncProcess(conn, conf, false);
805
806 List<Put> puts = new ArrayList<Put>();
807 puts.add(createPut(1, true));
808 puts.add(createPut(1, true));
809 puts.add(createPut(1, true));
810 puts.add(createPut(1, true));
811 puts.add(createPut(1, false));
812 puts.add(createPut(1, true));
813 puts.add(createPut(1, false));
814
815 Object[] res = new Object[puts.size()];
816 try {
817 ht.processBatch(puts, res);
818 Assert.fail();
819 } catch (RetriesExhaustedException expected) {
820 }
821
822 Assert.assertEquals(res[0], success);
823 Assert.assertEquals(res[1], success);
824 Assert.assertEquals(res[2], success);
825 Assert.assertEquals(res[3], success);
826 Assert.assertEquals(res[4], failure);
827 Assert.assertEquals(res[5], success);
828 Assert.assertEquals(res[6], failure);
829 }
830
831 @Test
832 public void testErrorsServers() throws IOException {
833 Configuration configuration = new Configuration(conf);
834 ClusterConnection conn = new MyConnectionImpl(configuration);
835 BufferedMutatorImpl mutator =
836 new BufferedMutatorImpl(conn, null, null, new BufferedMutatorParams(DUMMY_TABLE));
837 configuration.setBoolean(ConnectionManager.RETRIES_BY_SERVER_KEY, true);
838
839 MyAsyncProcess ap = new MyAsyncProcess(conn, configuration, true);
840 mutator.ap = ap;
841
842 Assert.assertNotNull(mutator.ap.createServerErrorTracker());
843 Assert.assertTrue(mutator.ap.serverTrackerTimeout > 200);
844 mutator.ap.serverTrackerTimeout = 1;
845
846 Put p = createPut(1, false);
847 mutator.mutate(p);
848
849 try {
850 mutator.flush();
851 Assert.fail();
852 } catch (RetriesExhaustedWithDetailsException expected) {
853 }
854
855 Assert.assertEquals(NB_RETRIES + 1, ap.callsCt.get());
856 }
857
858 @Test
859 public void testGlobalErrors() throws IOException {
860 ClusterConnection conn = new MyConnectionImpl(conf);
861 BufferedMutatorImpl mutator = (BufferedMutatorImpl) conn.getBufferedMutator(DUMMY_TABLE);
862 AsyncProcessWithFailure ap = new AsyncProcessWithFailure(conn, conf, new IOException("test"));
863 mutator.ap = ap;
864
865 Assert.assertNotNull(mutator.ap.createServerErrorTracker());
866
867 Put p = createPut(1, true);
868 mutator.mutate(p);
869
870 try {
871 mutator.flush();
872 Assert.fail();
873 } catch (RetriesExhaustedWithDetailsException expected) {
874 }
875
876 Assert.assertEquals(NB_RETRIES + 1, ap.callsCt.get());
877 }
878
879
880 @Test
881 public void testCallQueueTooLarge() throws IOException {
882 ClusterConnection conn = new MyConnectionImpl(conf);
883 BufferedMutatorImpl mutator = (BufferedMutatorImpl) conn.getBufferedMutator(DUMMY_TABLE);
884 AsyncProcessWithFailure ap = new AsyncProcessWithFailure(conn, conf, new CallQueueTooBigException());
885 mutator.ap = ap;
886
887 Assert.assertNotNull(mutator.ap.createServerErrorTracker());
888
889 Put p = createPut(1, true);
890 mutator.mutate(p);
891
892 try {
893 mutator.flush();
894 Assert.fail();
895 } catch (RetriesExhaustedWithDetailsException expected) {
896 }
897
898 Assert.assertEquals(NB_RETRIES + 1, ap.callsCt.get());
899 }
900
901
902
903
904 @Test
905 public void testThreadCreation() throws Exception {
906 final int NB_REGS = 100;
907 List<HRegionLocation> hrls = new ArrayList<HRegionLocation>(NB_REGS);
908 List<Get> gets = new ArrayList<Get>(NB_REGS);
909 for (int i = 0; i < NB_REGS; i++) {
910 HRegionInfo hri = new HRegionInfo(
911 DUMMY_TABLE, Bytes.toBytes(i * 10L), Bytes.toBytes(i * 10L + 9L), false, i);
912 HRegionLocation hrl = new HRegionLocation(hri, i % 2 == 0 ? sn : sn2);
913 hrls.add(hrl);
914
915 Get get = new Get(Bytes.toBytes(i * 10L));
916 gets.add(get);
917 }
918
919 MyConnectionImpl2 con = new MyConnectionImpl2(hrls);
920 HTable ht = new HTable(con, new BufferedMutatorParams(DUMMY_TABLE));
921 MyAsyncProcess ap = new MyAsyncProcess(con, conf, con.nbThreads);
922 ht.multiAp = ap;
923
924 ht.batch(gets, new Object[gets.size()]);
925
926 Assert.assertEquals(ap.nbActions.get(), NB_REGS);
927 Assert.assertEquals("1 multi response per server", 2, ap.nbMultiResponse.get());
928 Assert.assertEquals("1 thread per server", 2, con.nbThreads.get());
929
930 int nbReg = 0;
931 for (int i =0; i<NB_REGS; i++){
932 if (con.usedRegions[i]) nbReg++;
933 }
934 Assert.assertEquals("nbReg=" + nbReg, nbReg, NB_REGS);
935 }
936
937 @Test
938 public void testReplicaReplicaSuccess() throws Exception {
939
940
941 MyAsyncProcessWithReplicas ap = createReplicaAp(10, 1000, 0);
942 List<Get> rows = makeTimelineGets(DUMMY_BYTES_1, DUMMY_BYTES_2, DUMMY_BYTES_3);
943 AsyncRequestFuture ars = ap.submitAll(DUMMY_TABLE, rows, null, new Object[3]);
944 verifyReplicaResult(ars, RR.TRUE, RR.TRUE, RR.FALSE);
945 Assert.assertEquals(2, ap.getReplicaCallCount());
946 }
947
948 @Test
949 public void testReplicaPrimarySuccessWoReplicaCalls() throws Exception {
950
951 MyAsyncProcessWithReplicas ap = createReplicaAp(1000, 10, 0);
952 List<Get> rows = makeTimelineGets(DUMMY_BYTES_1, DUMMY_BYTES_2, DUMMY_BYTES_3);
953 AsyncRequestFuture ars = ap.submitAll(DUMMY_TABLE, rows, null, new Object[3]);
954 verifyReplicaResult(ars, RR.FALSE, RR.FALSE, RR.FALSE);
955 Assert.assertEquals(0, ap.getReplicaCallCount());
956 }
957
958 @Test
959 public void testReplicaParallelCallsSucceed() throws Exception {
960
961 MyAsyncProcessWithReplicas ap = createReplicaAp(0, 0, 0);
962 List<Get> rows = makeTimelineGets(DUMMY_BYTES_1, DUMMY_BYTES_2);
963 AsyncRequestFuture ars = ap.submitAll(DUMMY_TABLE, rows, null, new Object[2]);
964 verifyReplicaResult(ars, RR.DONT_CARE, RR.DONT_CARE);
965 long replicaCalls = ap.getReplicaCallCount();
966 Assert.assertTrue(replicaCalls >= 0);
967 Assert.assertTrue(replicaCalls <= 2);
968 }
969
970 @Test
971 public void testReplicaPartialReplicaCall() throws Exception {
972
973
974
975 MyAsyncProcessWithReplicas ap = createReplicaAp(1000, 0, 0);
976 ap.setPrimaryCallDelay(sn2, 2000);
977 List<Get> rows = makeTimelineGets(DUMMY_BYTES_1, DUMMY_BYTES_2);
978 AsyncRequestFuture ars = ap.submitAll(DUMMY_TABLE, rows, null, new Object[2]);
979 verifyReplicaResult(ars, RR.FALSE, RR.TRUE);
980 Assert.assertEquals(1, ap.getReplicaCallCount());
981 }
982
983 @Test
984 public void testReplicaMainFailsBeforeReplicaCalls() throws Exception {
985
986
987
988 MyAsyncProcessWithReplicas ap = createReplicaAp(1000, 0, 0, 1);
989 ap.addFailures(hri1, hri2);
990 List<Get> rows = makeTimelineGets(DUMMY_BYTES_1, DUMMY_BYTES_2);
991 AsyncRequestFuture ars = ap.submitAll(DUMMY_TABLE, rows, null, new Object[2]);
992 verifyReplicaResult(ars, RR.FAILED, RR.FAILED);
993 Assert.assertEquals(0, ap.getReplicaCallCount());
994 }
995
996 @Test
997 public void testReplicaReplicaSuccessWithParallelFailures() throws Exception {
998
999
1000 MyAsyncProcessWithReplicas ap = createReplicaAp(0, 1000, 1000, 1);
1001 ap.addFailures(hri1, hri1r2, hri2);
1002 List<Get> rows = makeTimelineGets(DUMMY_BYTES_1, DUMMY_BYTES_2);
1003 AsyncRequestFuture ars = ap.submitAll(DUMMY_TABLE, rows, null, new Object[2]);
1004 verifyReplicaResult(ars, RR.TRUE, RR.TRUE);
1005 Assert.assertEquals(2, ap.getReplicaCallCount());
1006 }
1007
1008 @Test
1009 public void testReplicaAllCallsFailForOneRegion() throws Exception {
1010
1011
1012 MyAsyncProcessWithReplicas ap = createReplicaAp(500, 1000, 0, 1);
1013 ap.addFailures(hri1, hri1r1, hri1r2, hri2r1);
1014 List<Get> rows = makeTimelineGets(DUMMY_BYTES_1, DUMMY_BYTES_2);
1015 AsyncRequestFuture ars = ap.submitAll(DUMMY_TABLE, rows, null, new Object[2]);
1016 verifyReplicaResult(ars, RR.FAILED, RR.FALSE);
1017
1018 Assert.assertEquals(3, ars.getErrors().getNumExceptions());
1019 for (int i = 0; i < ars.getErrors().getNumExceptions(); ++i) {
1020 Assert.assertArrayEquals(DUMMY_BYTES_1, ars.getErrors().getRow(i).getRow());
1021 }
1022 }
1023
1024 private MyAsyncProcessWithReplicas createReplicaAp(
1025 int replicaAfterMs, int primaryMs, int replicaMs) throws Exception {
1026 return createReplicaAp(replicaAfterMs, primaryMs, replicaMs, -1);
1027 }
1028
1029 private MyAsyncProcessWithReplicas createReplicaAp(
1030 int replicaAfterMs, int primaryMs, int replicaMs, int retries) throws Exception {
1031
1032
1033 Configuration conf = new Configuration();
1034 ClusterConnection conn = createHConnectionWithReplicas();
1035 conf.setInt(AsyncProcess.PRIMARY_CALL_TIMEOUT_KEY, replicaAfterMs * 1000);
1036 if (retries > 0) {
1037 conf.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, retries);
1038 }
1039 MyAsyncProcessWithReplicas ap = new MyAsyncProcessWithReplicas(conn, conf);
1040 ap.setCallDelays(primaryMs, replicaMs);
1041 return ap;
1042 }
1043
1044 private static List<Get> makeTimelineGets(byte[]... rows) {
1045 List<Get> result = new ArrayList<Get>();
1046 for (byte[] row : rows) {
1047 Get get = new Get(row);
1048 get.setConsistency(Consistency.TIMELINE);
1049 result.add(get);
1050 }
1051 return result;
1052 }
1053
1054 private void verifyResult(AsyncRequestFuture ars, boolean... expected) throws Exception {
1055 Object[] actual = ars.getResults();
1056 Assert.assertEquals(expected.length, actual.length);
1057 for (int i = 0; i < expected.length; ++i) {
1058 Assert.assertEquals(expected[i], !(actual[i] instanceof Throwable));
1059 }
1060 }
1061
1062
1063 private enum RR {
1064 TRUE,
1065 FALSE,
1066 DONT_CARE,
1067 FAILED
1068 }
1069
1070 private void verifyReplicaResult(AsyncRequestFuture ars, RR... expecteds) throws Exception {
1071 Object[] actuals = ars.getResults();
1072 Assert.assertEquals(expecteds.length, actuals.length);
1073 for (int i = 0; i < expecteds.length; ++i) {
1074 Object actual = actuals[i];
1075 RR expected = expecteds[i];
1076 Assert.assertEquals(actual.toString(), expected == RR.FAILED, actual instanceof Throwable);
1077 if (expected != RR.FAILED && expected != RR.DONT_CARE) {
1078 Assert.assertEquals(expected == RR.TRUE, ((Result)actual).isStale());
1079 }
1080 }
1081 }
1082
1083
1084
1085
1086
1087
1088 private Put createPut(int regCnt, boolean success) {
1089 Put p;
1090 if (!success) {
1091 p = new Put(FAILS);
1092 } else switch (regCnt){
1093 case 1 :
1094 p = new Put(DUMMY_BYTES_1);
1095 break;
1096 case 2:
1097 p = new Put(DUMMY_BYTES_2);
1098 break;
1099 case 3:
1100 p = new Put(DUMMY_BYTES_3);
1101 break;
1102 default:
1103 throw new IllegalArgumentException("unknown " + regCnt);
1104 }
1105
1106 p.add(DUMMY_BYTES_1, DUMMY_BYTES_1, DUMMY_BYTES_1);
1107
1108 return p;
1109 }
1110
1111 static class MyThreadPoolExecutor extends ThreadPoolExecutor {
1112 public MyThreadPoolExecutor(int coreThreads, int maxThreads, long keepAliveTime,
1113 TimeUnit timeunit, BlockingQueue<Runnable> blockingqueue) {
1114 super(coreThreads, maxThreads, keepAliveTime, timeunit, blockingqueue);
1115 }
1116
1117 @Override
1118 public Future submit(Runnable runnable) {
1119 throw new OutOfMemoryError("OutOfMemory error thrown by means");
1120 }
1121 }
1122
1123 static class AsyncProcessForThrowableCheck extends AsyncProcess {
1124 public AsyncProcessForThrowableCheck(ClusterConnection hc, Configuration conf,
1125 ExecutorService pool) {
1126 super(hc, conf, pool, new RpcRetryingCallerFactory(conf), false, new RpcControllerFactory(
1127 conf),
1128 conf.getInt(HConstants.HBASE_RPC_TIMEOUT_KEY,
1129 HConstants.DEFAULT_HBASE_RPC_TIMEOUT));
1130 }
1131 }
1132
1133 @Test
1134 public void testUncheckedException() throws Exception {
1135
1136 ClusterConnection hc = createHConnection();
1137 MyThreadPoolExecutor myPool =
1138 new MyThreadPoolExecutor(1, 20, 60, TimeUnit.SECONDS,
1139 new LinkedBlockingQueue<Runnable>(200));
1140 AsyncProcess ap = new AsyncProcessForThrowableCheck(hc, conf, myPool);
1141
1142 List<Put> puts = new ArrayList<Put>();
1143 puts.add(createPut(1, true));
1144
1145 ap.submit(DUMMY_TABLE, puts, false, null, false);
1146 Assert.assertTrue(puts.isEmpty());
1147 }
1148
1149 @Test
1150 public void testWaitForMaximumCurrentTasks() throws Exception {
1151 final AtomicLong tasks = new AtomicLong(0);
1152 final AtomicInteger max = new AtomicInteger(0);
1153 final CyclicBarrier barrier = new CyclicBarrier(2);
1154 final AsyncProcess ap = new MyAsyncProcess(createHConnection(), conf);
1155 Runnable runnable = new Runnable() {
1156 @Override
1157 public void run() {
1158 try {
1159 barrier.await();
1160 ap.waitForMaximumCurrentTasks(max.get(), tasks, 1, null);
1161 } catch (InterruptedIOException e) {
1162 Assert.fail(e.getMessage());
1163 } catch (InterruptedException e) {
1164
1165 e.printStackTrace();
1166 } catch (BrokenBarrierException e) {
1167
1168 e.printStackTrace();
1169 }
1170 }
1171 };
1172
1173 Thread t = new Thread(runnable);
1174 t.start();
1175 barrier.await();
1176 t.join();
1177
1178 barrier.reset();
1179 tasks.set(1000000);
1180 t = new Thread(runnable);
1181 t.start();
1182 barrier.await();
1183 while (tasks.get() > 0) {
1184 assertTrue(t.isAlive());
1185 tasks.set(tasks.get() - 1);
1186 }
1187 t.join();
1188 }
1189 }