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