1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase.client;
19
20 import static org.junit.Assert.assertEquals;
21 import static org.junit.Assert.assertFalse;
22 import static org.junit.Assert.assertNotNull;
23 import static org.junit.Assert.assertTrue;
24 import static org.junit.Assert.fail;
25 import static org.mockito.Matchers.any;
26 import static org.mockito.Matchers.anyBoolean;
27 import static org.mockito.Matchers.anyInt;
28 import static org.mockito.Mockito.when;
29
30 import java.io.IOException;
31 import java.util.Iterator;
32 import java.util.concurrent.ExecutorService;
33 import java.util.concurrent.Executors;
34
35 import org.apache.hadoop.conf.Configuration;
36 import org.apache.hadoop.hbase.Cell;
37 import org.apache.hadoop.hbase.CellScanner;
38 import org.apache.hadoop.hbase.KeyValue;
39 import org.apache.hadoop.hbase.KeyValue.Type;
40 import org.apache.hadoop.hbase.RegionLocations;
41 import org.apache.hadoop.hbase.TableName;
42 import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
43 import org.apache.hadoop.hbase.testclassification.SmallTests;
44 import org.junit.After;
45 import org.junit.Before;
46 import org.junit.Test;
47 import org.junit.experimental.categories.Category;
48 import org.mockito.InOrder;
49 import org.mockito.Mockito;
50 import org.mockito.invocation.InvocationOnMock;
51 import org.mockito.stubbing.Answer;
52
53
54
55
56 @Category(SmallTests.class)
57 public class TestClientScanner {
58
59 Scan scan;
60 ExecutorService pool;
61 Configuration conf;
62
63 ClusterConnection clusterConn;
64 RpcRetryingCallerFactory rpcFactory;
65 RpcControllerFactory controllerFactory;
66
67 @Before
68 @SuppressWarnings("deprecation")
69 public void setup() throws IOException {
70 clusterConn = Mockito.mock(ClusterConnection.class);
71 rpcFactory = Mockito.mock(RpcRetryingCallerFactory.class);
72 controllerFactory = Mockito.mock(RpcControllerFactory.class);
73 pool = Executors.newSingleThreadExecutor();
74 scan = new Scan();
75 conf = new Configuration();
76 Mockito.when(clusterConn.getConfiguration()).thenReturn(conf);
77 }
78
79 @After
80 public void teardown() {
81 if (null != pool) {
82 pool.shutdownNow();
83 }
84 }
85
86 private static class MockClientScanner extends ClientScanner {
87
88 private boolean rpcFinished = false;
89 private boolean rpcFinishedFired = false;
90
91 public MockClientScanner(final Configuration conf, final Scan scan, final TableName tableName,
92 ClusterConnection connection, RpcRetryingCallerFactory rpcFactory,
93 RpcControllerFactory controllerFactory, ExecutorService pool, int primaryOperationTimeout)
94 throws IOException {
95 super(conf, scan, tableName, connection, rpcFactory, controllerFactory, pool,
96 primaryOperationTimeout);
97 }
98
99 @Override
100 protected boolean nextScanner(int nbRows, final boolean done) throws IOException {
101 if (!rpcFinished) {
102 return super.nextScanner(nbRows, done);
103 }
104
105
106 if (rpcFinishedFired) {
107 throw new RuntimeException("Expected nextScanner to only be called once after " +
108 " short-circuit was triggered.");
109 }
110 rpcFinishedFired = true;
111 return false;
112 }
113
114 @Override
115 protected ScannerCallableWithReplicas getScannerCallable(byte [] localStartKey,
116 int nbRows) {
117 scan.setStartRow(localStartKey);
118 ScannerCallable s =
119 new ScannerCallable(getConnection(), getTable(), scan, this.scanMetrics,
120 this.rpcControllerFactory);
121 s.setCaching(nbRows);
122 ScannerCallableWithReplicas sr = new ScannerCallableWithReplicas(getTable(), getConnection(),
123 s, pool, primaryOperationTimeout, scan,
124 getRetries(), scannerTimeout, caching, conf, caller);
125 return sr;
126 }
127
128 public void setRpcFinished(boolean rpcFinished) {
129 this.rpcFinished = rpcFinished;
130 }
131 }
132
133 @Test
134 @SuppressWarnings("unchecked")
135 public void testNoResultsHint() throws IOException {
136 final Result[] results = new Result[1];
137 KeyValue kv1 = new KeyValue("row".getBytes(), "cf".getBytes(), "cq".getBytes(), 1,
138 Type.Maximum);
139 results[0] = Result.create(new Cell[] {kv1});
140
141 RpcRetryingCaller<Result[]> caller = Mockito.mock(RpcRetryingCaller.class);
142
143 Mockito.when(rpcFactory.<Result[]> newCaller()).thenReturn(caller);
144 Mockito.when(caller.callWithoutRetries(Mockito.any(RetryingCallable.class),
145 Mockito.anyInt())).thenAnswer(new Answer<Result[]>() {
146 private int count = 0;
147 @Override
148 public Result[] answer(InvocationOnMock invocation) throws Throwable {
149 ScannerCallableWithReplicas callable = invocation.getArgumentAt(0,
150 ScannerCallableWithReplicas.class);
151 switch (count) {
152 case 0:
153 case 2:
154 case 3:
155 count++;
156 return null;
157 case 1:
158 count++;
159 callable.setHasMoreResultsContext(false);
160 return results;
161 default:
162 throw new RuntimeException("Expected only 2 invocations");
163 }
164 }
165 });
166
167
168 scan.setCaching(100);
169 scan.setMaxResultSize(1000*1000);
170
171 try (MockClientScanner scanner = new MockClientScanner(conf, scan, TableName.valueOf("table"),
172 clusterConn, rpcFactory, controllerFactory, pool, Integer.MAX_VALUE)) {
173
174 scanner.setRpcFinished(true);
175
176 InOrder inOrder = Mockito.inOrder(caller);
177
178 scanner.loadCache();
179
180
181
182
183 inOrder.verify(caller, Mockito.times(3)).callWithoutRetries(
184 Mockito.any(RetryingCallable.class), Mockito.anyInt());
185
186 assertEquals(1, scanner.cache.size());
187 Result r = scanner.cache.poll();
188 assertNotNull(r);
189 CellScanner cs = r.cellScanner();
190 assertTrue(cs.advance());
191 assertEquals(kv1, cs.current());
192 assertFalse(cs.advance());
193 }
194 }
195
196 @Test
197 @SuppressWarnings("unchecked")
198 public void testSizeLimit() throws IOException {
199 final Result[] results = new Result[1];
200 KeyValue kv1 = new KeyValue("row".getBytes(), "cf".getBytes(), "cq".getBytes(), 1,
201 Type.Maximum);
202 results[0] = Result.create(new Cell[] {kv1});
203
204 RpcRetryingCaller<Result[]> caller = Mockito.mock(RpcRetryingCaller.class);
205
206 Mockito.when(rpcFactory.<Result[]> newCaller()).thenReturn(caller);
207 Mockito.when(caller.callWithoutRetries(Mockito.any(RetryingCallable.class),
208 Mockito.anyInt())).thenAnswer(new Answer<Result[]>() {
209 private int count = 0;
210 @Override
211 public Result[] answer(InvocationOnMock invocation) throws Throwable {
212 ScannerCallableWithReplicas callable = invocation.getArgumentAt(0,
213 ScannerCallableWithReplicas.class);
214 switch (count) {
215 case 0:
216 case 2:
217 count++;
218 return null;
219 case 1:
220 count++;
221 callable.setHasMoreResultsContext(true);
222
223 callable.setServerHasMoreResults(true);
224 return results;
225 default:
226 throw new RuntimeException("Expected only 2 invocations");
227 }
228 }
229 });
230
231 Mockito.when(rpcFactory.<Result[]> newCaller()).thenReturn(caller);
232
233
234 scan.setCaching(100);
235
236 scan.setMaxResultSize(1);
237
238 try (MockClientScanner scanner = new MockClientScanner(conf, scan, TableName.valueOf("table"),
239 clusterConn, rpcFactory, controllerFactory, pool, Integer.MAX_VALUE)) {
240
241
242 Mockito.verify(caller).callWithoutRetries(Mockito.any(RetryingCallable.class),
243 Mockito.anyInt());
244
245 InOrder inOrder = Mockito.inOrder(caller);
246
247 scanner.loadCache();
248
249 inOrder.verify(caller, Mockito.times(2)).callWithoutRetries(
250 Mockito.any(RetryingCallable.class), Mockito.anyInt());
251
252 assertEquals(1, scanner.cache.size());
253 Result r = scanner.cache.poll();
254 assertNotNull(r);
255 CellScanner cs = r.cellScanner();
256 assertTrue(cs.advance());
257 assertEquals(kv1, cs.current());
258 assertFalse(cs.advance());
259 }
260 }
261
262 @Test
263 @SuppressWarnings("unchecked")
264 public void testCacheLimit() throws IOException {
265 KeyValue kv1 = new KeyValue("row1".getBytes(), "cf".getBytes(), "cq".getBytes(), 1,
266 Type.Maximum), kv2 = new KeyValue("row2".getBytes(), "cf".getBytes(), "cq".getBytes(), 1,
267 Type.Maximum), kv3 = new KeyValue("row3".getBytes(), "cf".getBytes(), "cq".getBytes(), 1,
268 Type.Maximum);
269 final Result[] results = new Result[] {Result.create(new Cell[] {kv1}),
270 Result.create(new Cell[] {kv2}), Result.create(new Cell[] {kv3})};
271
272 RpcRetryingCaller<Result[]> caller = Mockito.mock(RpcRetryingCaller.class);
273
274 Mockito.when(rpcFactory.<Result[]> newCaller()).thenReturn(caller);
275 Mockito.when(caller.callWithoutRetries(Mockito.any(RetryingCallable.class),
276 Mockito.anyInt())).thenAnswer(new Answer<Result[]>() {
277 private int count = 0;
278 @Override
279 public Result[] answer(InvocationOnMock invocation) throws Throwable {
280 ScannerCallableWithReplicas callable = invocation.getArgumentAt(0,
281 ScannerCallableWithReplicas.class);
282 switch (count) {
283 case 0:
284 case 2:
285 count++;
286 return null;
287 case 1:
288 count++;
289 callable.setHasMoreResultsContext(true);
290
291 callable.setServerHasMoreResults(true);
292 return results;
293 default:
294 throw new RuntimeException("Expected only 2 invocations");
295 }
296 }
297 });
298
299 Mockito.when(rpcFactory.<Result[]> newCaller()).thenReturn(caller);
300
301
302 scan.setCaching(1);
303
304 scan.setMaxResultSize(1000*1000);
305
306 try (MockClientScanner scanner = new MockClientScanner(conf, scan, TableName.valueOf("table"),
307 clusterConn, rpcFactory, controllerFactory, pool, Integer.MAX_VALUE)) {
308
309
310 Mockito.verify(caller).callWithoutRetries(Mockito.any(RetryingCallable.class),
311 Mockito.anyInt());
312
313 InOrder inOrder = Mockito.inOrder(caller);
314
315 scanner.loadCache();
316
317
318
319 inOrder.verify(caller, Mockito.times(2)).callWithoutRetries(
320 Mockito.any(RetryingCallable.class), Mockito.anyInt());
321
322 assertEquals(3, scanner.cache.size());
323 Result r = scanner.cache.poll();
324 assertNotNull(r);
325 CellScanner cs = r.cellScanner();
326 assertTrue(cs.advance());
327 assertEquals(kv1, cs.current());
328 assertFalse(cs.advance());
329
330 r = scanner.cache.poll();
331 assertNotNull(r);
332 cs = r.cellScanner();
333 assertTrue(cs.advance());
334 assertEquals(kv2, cs.current());
335 assertFalse(cs.advance());
336
337 r = scanner.cache.poll();
338 assertNotNull(r);
339 cs = r.cellScanner();
340 assertTrue(cs.advance());
341 assertEquals(kv3, cs.current());
342 assertFalse(cs.advance());
343 }
344 }
345
346 @Test
347 @SuppressWarnings("unchecked")
348 public void testNoMoreResults() throws IOException {
349 final Result[] results = new Result[1];
350 KeyValue kv1 = new KeyValue("row".getBytes(), "cf".getBytes(), "cq".getBytes(), 1,
351 Type.Maximum);
352 results[0] = Result.create(new Cell[] {kv1});
353
354 RpcRetryingCaller<Result[]> caller = Mockito.mock(RpcRetryingCaller.class);
355
356 Mockito.when(rpcFactory.<Result[]> newCaller()).thenReturn(caller);
357 Mockito.when(caller.callWithoutRetries(Mockito.any(RetryingCallable.class),
358 Mockito.anyInt())).thenAnswer(new Answer<Result[]>() {
359 private int count = 0;
360 @Override
361 public Result[] answer(InvocationOnMock invocation) throws Throwable {
362 ScannerCallableWithReplicas callable = invocation.getArgumentAt(0,
363 ScannerCallableWithReplicas.class);
364 switch (count) {
365 case 0:
366 case 2:
367 count++;
368 return null;
369 case 1:
370 count++;
371 callable.setHasMoreResultsContext(true);
372 callable.setServerHasMoreResults(false);
373 return results;
374 default:
375 throw new RuntimeException("Expected only 2 invocations");
376 }
377 }
378 });
379
380 Mockito.when(rpcFactory.<Result[]> newCaller()).thenReturn(caller);
381
382
383 scan.setCaching(100);
384 scan.setMaxResultSize(1000*1000);
385
386 try (MockClientScanner scanner = new MockClientScanner(conf, scan, TableName.valueOf("table"),
387 clusterConn, rpcFactory, controllerFactory, pool, Integer.MAX_VALUE)) {
388
389
390 Mockito.verify(caller).callWithoutRetries(Mockito.any(RetryingCallable.class),
391 Mockito.anyInt());
392
393 scanner.setRpcFinished(true);
394
395 InOrder inOrder = Mockito.inOrder(caller);
396
397 scanner.loadCache();
398
399 inOrder.verify(caller, Mockito.times(2)).callWithoutRetries(
400 Mockito.any(RetryingCallable.class), Mockito.anyInt());
401
402 assertEquals(1, scanner.cache.size());
403 Result r = scanner.cache.poll();
404 assertNotNull(r);
405 CellScanner cs = r.cellScanner();
406 assertTrue(cs.advance());
407 assertEquals(kv1, cs.current());
408 assertFalse(cs.advance());
409 }
410 }
411
412 @Test
413 @SuppressWarnings("unchecked")
414 public void testMoreResults() throws IOException {
415 final Result[] results1 = new Result[1];
416 KeyValue kv1 = new KeyValue("row".getBytes(), "cf".getBytes(), "cq".getBytes(), 1,
417 Type.Maximum);
418 results1[0] = Result.create(new Cell[] {kv1});
419
420 final Result[] results2 = new Result[1];
421 KeyValue kv2 = new KeyValue("row2".getBytes(), "cf".getBytes(), "cq".getBytes(), 1,
422 Type.Maximum);
423 results2[0] = Result.create(new Cell[] {kv2});
424
425
426 RpcRetryingCaller<Result[]> caller = Mockito.mock(RpcRetryingCaller.class);
427
428 Mockito.when(rpcFactory.<Result[]> newCaller()).thenReturn(caller);
429 Mockito.when(caller.callWithoutRetries(Mockito.any(RetryingCallable.class),
430 Mockito.anyInt())).thenAnswer(new Answer<Result[]>() {
431 private int count = 0;
432 @Override
433 public Result[] answer(InvocationOnMock invocation) throws Throwable {
434 ScannerCallableWithReplicas callable = invocation.getArgumentAt(0,
435 ScannerCallableWithReplicas.class);
436 switch (count) {
437 case 0:
438 case 3:
439 count++;
440 return null;
441 case 1:
442 count++;
443 callable.setHasMoreResultsContext(true);
444 callable.setServerHasMoreResults(true);
445 return results1;
446 case 2:
447 count++;
448
449 callable.setHasMoreResultsContext(true);
450 callable.setServerHasMoreResults(false);
451 return results2;
452 default:
453 throw new RuntimeException("Expected only 2 invocations");
454 }
455 }
456 });
457
458
459 scan.setCaching(100);
460 scan.setMaxResultSize(1000*1000);
461
462 try (MockClientScanner scanner = new MockClientScanner(conf, scan, TableName.valueOf("table"),
463 clusterConn, rpcFactory, controllerFactory, pool, Integer.MAX_VALUE)) {
464
465
466 Mockito.verify(caller).callWithoutRetries(Mockito.any(RetryingCallable.class),
467 Mockito.anyInt());
468
469 InOrder inOrder = Mockito.inOrder(caller);
470 scanner.setRpcFinished(true);
471
472 scanner.loadCache();
473
474 inOrder.verify(caller, Mockito.times(3)).callWithoutRetries(
475 Mockito.any(RetryingCallable.class), Mockito.anyInt());
476
477 assertEquals(2, scanner.cache.size());
478 Result r = scanner.cache.poll();
479 assertNotNull(r);
480 CellScanner cs = r.cellScanner();
481 assertTrue(cs.advance());
482 assertEquals(kv1, cs.current());
483 assertFalse(cs.advance());
484
485 r = scanner.cache.poll();
486 assertNotNull(r);
487 cs = r.cellScanner();
488 assertTrue(cs.advance());
489 assertEquals(kv2, cs.current());
490 assertFalse(cs.advance());
491 }
492 }
493
494
495
496
497
498 @Test (timeout = 30000)
499 public void testExceptionsFromReplicasArePropagated() throws IOException {
500 scan.setConsistency(Consistency.TIMELINE);
501
502
503
504 rpcFactory = new MockRpcRetryingCallerFactory(conf);
505 conf.set(RpcRetryingCallerFactory.CUSTOM_CALLER_CONF_KEY,
506 MockRpcRetryingCallerFactory.class.getName());
507
508
509 when(clusterConn.locateRegion((TableName)any(), (byte[])any(), anyBoolean(),
510 anyBoolean(), anyInt())).thenReturn(new RegionLocations(null, null, null));
511
512 try (MockClientScanner scanner = new MockClientScanner(conf, scan, TableName.valueOf("table"),
513 clusterConn, rpcFactory, new RpcControllerFactory(conf), pool, Integer.MAX_VALUE)) {
514 Iterator<Result> iter = scanner.iterator();
515 while (iter.hasNext()) {
516 iter.next();
517 }
518 fail("Should have failed with RetriesExhaustedException");
519 } catch (RetriesExhaustedException expected) {
520
521 }
522 }
523
524 public static class MockRpcRetryingCallerFactory extends RpcRetryingCallerFactory {
525
526 public MockRpcRetryingCallerFactory(Configuration conf) {
527 super(conf);
528 }
529
530 @Override
531 public <T> RpcRetryingCaller<T> newCaller() {
532 return new RpcRetryingCaller<T>(0, 0, 0) {
533 @Override
534 public void cancel() {
535 }
536 @Override
537 public T callWithRetries(RetryingCallable<T> callable, int callTimeout)
538 throws IOException, RuntimeException {
539 throw new IOException("Scanner exception");
540 }
541
542 @Override
543 public T callWithoutRetries(RetryingCallable<T> callable, int callTimeout)
544 throws IOException, RuntimeException {
545 try {
546 return callable.call(callTimeout);
547 } catch (IOException e) {
548 throw e;
549 } catch (Exception e) {
550 throw new RuntimeException(e);
551 }
552 }
553 };
554 }
555
556 }
557
558 }