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.assertTrue;
21 import static org.junit.Assert.fail;
22
23 import java.io.IOException;
24 import java.net.SocketTimeoutException;
25 import java.util.Comparator;
26 import java.util.HashMap;
27 import java.util.Map;
28 import java.util.Random;
29 import java.util.SortedMap;
30 import java.util.concurrent.ConcurrentSkipListMap;
31 import java.util.concurrent.ExecutorService;
32 import java.util.concurrent.Executors;
33 import java.util.concurrent.atomic.AtomicInteger;
34 import java.util.concurrent.atomic.AtomicLong;
35
36 import org.apache.hadoop.hbase.util.ByteStringer;
37 import org.apache.commons.lang.NotImplementedException;
38 import org.apache.commons.logging.Log;
39 import org.apache.commons.logging.LogFactory;
40 import org.apache.hadoop.conf.Configuration;
41 import org.apache.hadoop.conf.Configured;
42 import org.apache.hadoop.hbase.HBaseConfiguration;
43 import org.apache.hadoop.hbase.HConstants;
44 import org.apache.hadoop.hbase.HRegionInfo;
45 import org.apache.hadoop.hbase.HRegionLocation;
46 import org.apache.hadoop.hbase.KeyValue;
47 import org.apache.hadoop.hbase.RegionTooBusyException;
48 import org.apache.hadoop.hbase.ServerName;
49 import org.apache.hadoop.hbase.testclassification.SmallTests;
50 import org.apache.hadoop.hbase.TableName;
51 import org.apache.hadoop.hbase.protobuf.generated.CellProtos;
52 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos;
53 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.BulkLoadHFileRequest;
54 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.BulkLoadHFileResponse;
55 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService;
56 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService.BlockingInterface;
57 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.CoprocessorServiceRequest;
58 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.CoprocessorServiceResponse;
59 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.GetRequest;
60 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.GetResponse;
61 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MultiRequest;
62 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MultiResponse;
63 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutateRequest;
64 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.MutateResponse;
65 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.RegionAction;
66 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.RegionActionResult;
67 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ResultOrException;
68 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanRequest;
69 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ScanResponse;
70 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionSpecifier.RegionSpecifierType;
71 import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
72 import org.apache.hadoop.hbase.security.User;
73 import org.apache.hadoop.hbase.util.Bytes;
74 import org.apache.hadoop.hbase.util.Pair;
75 import org.apache.hadoop.hbase.util.Threads;
76 import org.apache.hadoop.util.Tool;
77 import org.apache.hadoop.util.ToolRunner;
78 import org.junit.Before;
79 import org.junit.Ignore;
80 import org.junit.Test;
81 import org.junit.experimental.categories.Category;
82 import org.mockito.Mockito;
83
84 import com.google.common.base.Stopwatch;
85 import com.google.protobuf.ByteString;
86 import com.google.protobuf.RpcController;
87 import com.google.protobuf.ServiceException;
88
89
90
91
92
93 @Category(SmallTests.class)
94 public class TestClientNoCluster extends Configured implements Tool {
95 private static final Log LOG = LogFactory.getLog(TestClientNoCluster.class);
96 private Configuration conf;
97 public static final ServerName META_SERVERNAME =
98 ServerName.valueOf("meta.example.org", 60010, 12345);
99
100 @Before
101 public void setUp() throws Exception {
102 this.conf = HBaseConfiguration.create();
103
104
105
106 this.conf.set("hbase.client.registry.impl", SimpleRegistry.class.getName());
107 }
108
109
110
111
112 static class SimpleRegistry implements Registry {
113 final ServerName META_HOST = META_SERVERNAME;
114
115 @Override
116 public void init(HConnection connection) {
117 }
118
119 @Override
120 public HRegionLocation getMetaRegionLocation() throws IOException {
121 return new HRegionLocation(HRegionInfo.FIRST_META_REGIONINFO, META_HOST);
122 }
123
124 @Override
125 public String getClusterId() {
126 return HConstants.CLUSTER_ID_DEFAULT;
127 }
128
129 @Override
130 public boolean isTableOnlineState(TableName tableName, boolean enabled)
131 throws IOException {
132 return enabled;
133 }
134
135 @Override
136 public int getCurrentNrHRS() throws IOException {
137 return 1;
138 }
139 }
140
141
142
143
144
145 @Ignore
146 @Test
147 public void testTimeoutAndRetries() throws IOException {
148 Configuration localConfig = HBaseConfiguration.create(this.conf);
149
150 localConfig.set("hbase.client.connection.impl", RpcTimeoutConnection.class.getName());
151 HTable table = new HTable(localConfig, TableName.META_TABLE_NAME);
152 Throwable t = null;
153 LOG.info("Start");
154 try {
155
156 table.exists(new Get(Bytes.toBytes("abc")));
157 } catch (SocketTimeoutException e) {
158
159 LOG.info("Got expected exception", e);
160 t = e;
161 } catch (RetriesExhaustedException e) {
162
163 fail();
164 } finally {
165 table.close();
166 }
167 LOG.info("Stop");
168 assertTrue(t != null);
169 }
170
171
172
173
174
175 @Test
176 public void testRocTimeout() throws IOException {
177 Configuration localConfig = HBaseConfiguration.create(this.conf);
178
179 localConfig.set("hbase.client.connection.impl", RpcTimeoutConnection.class.getName());
180 int pause = 10;
181 localConfig.setInt("hbase.client.pause", pause);
182 localConfig.setInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER, 10);
183
184
185
186
187 localConfig.setInt(HConstants.HBASE_CLIENT_META_OPERATION_TIMEOUT, pause - 1);
188 HTable table = new HTable(localConfig, TableName.META_TABLE_NAME);
189 Throwable t = null;
190 try {
191
192 table.exists(new Get(Bytes.toBytes("abc")));
193 } catch (SocketTimeoutException e) {
194
195 LOG.info("Got expected exception", e);
196 t = e;
197 } catch (RetriesExhaustedException e) {
198
199 fail();
200 } finally {
201 table.close();
202 }
203 assertTrue(t != null);
204 }
205
206 @Test
207 public void testDoNotRetryMetaScanner() throws IOException {
208 this.conf.set("hbase.client.connection.impl",
209 RegionServerStoppedOnScannerOpenConnection.class.getName());
210 MetaScanner.metaScan(this.conf, null);
211 }
212
213 @Test
214 public void testDoNotRetryOnScanNext() throws IOException {
215 this.conf.set("hbase.client.connection.impl",
216 RegionServerStoppedOnScannerOpenConnection.class.getName());
217
218
219
220 HTable table = new HTable(this.conf, TableName.META_TABLE_NAME);
221 ResultScanner scanner = table.getScanner(HConstants.CATALOG_FAMILY);
222 try {
223 Result result = null;
224 while ((result = scanner.next()) != null) {
225 LOG.info(result);
226 }
227 } finally {
228 scanner.close();
229 table.close();
230 }
231 }
232
233 @Test
234 public void testRegionServerStoppedOnScannerOpen() throws IOException {
235 this.conf.set("hbase.client.connection.impl",
236 RegionServerStoppedOnScannerOpenConnection.class.getName());
237
238
239
240 HTable table = new HTable(this.conf, TableName.META_TABLE_NAME);
241 ResultScanner scanner = table.getScanner(HConstants.CATALOG_FAMILY);
242 try {
243 Result result = null;
244 while ((result = scanner.next()) != null) {
245 LOG.info(result);
246 }
247 } finally {
248 scanner.close();
249 table.close();
250 }
251 }
252
253
254
255
256 static class ScanOpenNextThenExceptionThenRecoverConnection
257 extends HConnectionManager.HConnectionImplementation {
258 final ClientService.BlockingInterface stub;
259
260 ScanOpenNextThenExceptionThenRecoverConnection(Configuration conf,
261 boolean managed, ExecutorService pool) throws IOException {
262 super(conf, managed);
263
264
265 this.stub = Mockito.mock(ClientService.BlockingInterface.class);
266 long sid = 12345L;
267 try {
268 Mockito.when(stub.scan((RpcController)Mockito.any(),
269 (ClientProtos.ScanRequest)Mockito.any())).
270 thenReturn(ClientProtos.ScanResponse.newBuilder().setScannerId(sid).build()).
271 thenThrow(new ServiceException(new RegionServerStoppedException("From Mockito"))).
272 thenReturn(ClientProtos.ScanResponse.newBuilder().setScannerId(sid).
273 setMoreResults(false).build());
274 } catch (ServiceException e) {
275 throw new IOException(e);
276 }
277 }
278
279 @Override
280 public BlockingInterface getClient(ServerName sn) throws IOException {
281 return this.stub;
282 }
283 }
284
285
286
287
288 static class RegionServerStoppedOnScannerOpenConnection
289 extends HConnectionManager.HConnectionImplementation {
290 final ClientService.BlockingInterface stub;
291
292 RegionServerStoppedOnScannerOpenConnection(Configuration conf, boolean managed,
293 ExecutorService pool, User user) throws IOException {
294 super(conf, managed);
295
296
297 this.stub = Mockito.mock(ClientService.BlockingInterface.class);
298 long sid = 12345L;
299 try {
300 Mockito.when(stub.scan((RpcController)Mockito.any(),
301 (ClientProtos.ScanRequest)Mockito.any())).
302 thenReturn(ClientProtos.ScanResponse.newBuilder().setScannerId(sid).build()).
303 thenThrow(new ServiceException(new RegionServerStoppedException("From Mockito"))).
304 thenReturn(ClientProtos.ScanResponse.newBuilder().setScannerId(sid).
305 setMoreResults(false).build());
306 } catch (ServiceException e) {
307 throw new IOException(e);
308 }
309 }
310
311 @Override
312 public BlockingInterface getClient(ServerName sn) throws IOException {
313 return this.stub;
314 }
315 }
316
317
318
319
320 static class RpcTimeoutConnection
321 extends HConnectionManager.HConnectionImplementation {
322 final ClientService.BlockingInterface stub;
323
324 RpcTimeoutConnection(Configuration conf, boolean managed, ExecutorService pool, User user)
325 throws IOException {
326 super(conf, managed);
327
328 this.stub = Mockito.mock(ClientService.BlockingInterface.class);
329 try {
330 Mockito.when(stub.get((RpcController)Mockito.any(),
331 (ClientProtos.GetRequest)Mockito.any())).
332 thenThrow(new ServiceException(new RegionServerStoppedException("From Mockito")));
333 } catch (ServiceException e) {
334 throw new IOException(e);
335 }
336 }
337
338 @Override
339 public BlockingInterface getClient(ServerName sn) throws IOException {
340 return this.stub;
341 }
342 }
343
344
345
346
347 static class ManyServersManyRegionsConnection
348 extends HConnectionManager.HConnectionImplementation {
349
350 final Map<ServerName, ClientService.BlockingInterface> serversByClient;
351
352
353
354
355 final SortedMap<byte [], Pair<HRegionInfo, ServerName>> meta;
356 final AtomicLong sequenceids = new AtomicLong(0);
357 private final Configuration conf;
358
359 ManyServersManyRegionsConnection(Configuration conf, boolean managed,
360 ExecutorService pool, User user)
361 throws IOException {
362 super(conf, managed, pool, user);
363 int serverCount = conf.getInt("hbase.test.servers", 10);
364 this.serversByClient =
365 new HashMap<ServerName, ClientService.BlockingInterface>(serverCount);
366 this.meta = makeMeta(Bytes.toBytes(
367 conf.get("hbase.test.tablename", Bytes.toString(BIG_USER_TABLE))),
368 conf.getInt("hbase.test.regions", 100),
369 conf.getLong("hbase.test.namespace.span", 1000),
370 serverCount);
371 this.conf = conf;
372 }
373
374 @Override
375 public ClientService.BlockingInterface getClient(ServerName sn) throws IOException {
376
377 ClientService.BlockingInterface stub = null;
378 synchronized (this.serversByClient) {
379 stub = this.serversByClient.get(sn);
380 if (stub == null) {
381 stub = new FakeServer(this.conf, meta, sequenceids);
382 this.serversByClient.put(sn, stub);
383 }
384 }
385 return stub;
386 }
387 }
388
389 static MultiResponse doMultiResponse(final SortedMap<byte [], Pair<HRegionInfo, ServerName>> meta,
390 final AtomicLong sequenceids, final MultiRequest request) {
391
392 ClientProtos.MultiResponse.Builder builder = ClientProtos.MultiResponse.newBuilder();
393
394 RegionActionResult.Builder regionActionResultBuilder =
395 RegionActionResult.newBuilder();
396 ResultOrException.Builder roeBuilder = ResultOrException.newBuilder();
397 for (RegionAction regionAction: request.getRegionActionList()) {
398 regionActionResultBuilder.clear();
399
400 for (ClientProtos.Action action: regionAction.getActionList()) {
401 roeBuilder.clear();
402
403 roeBuilder.setResult(ClientProtos.Result.getDefaultInstance());
404 roeBuilder.setIndex(action.getIndex());
405 regionActionResultBuilder.addResultOrException(roeBuilder.build());
406 }
407 builder.addRegionActionResult(regionActionResultBuilder.build());
408 }
409 return builder.build();
410 }
411
412
413
414
415
416
417 static class FakeServer implements ClientService.BlockingInterface {
418 private AtomicInteger multiInvocationsCount = new AtomicInteger(0);
419 private final SortedMap<byte [], Pair<HRegionInfo, ServerName>> meta;
420 private final AtomicLong sequenceids;
421 private final long multiPause;
422 private final int tooManyMultiRequests;
423
424 FakeServer(final Configuration c, final SortedMap<byte [], Pair<HRegionInfo, ServerName>> meta,
425 final AtomicLong sequenceids) {
426 this.meta = meta;
427 this.sequenceids = sequenceids;
428
429
430
431 this.multiPause = c.getLong("hbase.test.multi.pause.when.done", 0);
432 this.tooManyMultiRequests = c.getInt("hbase.test.multi.too.many", 3);
433 }
434
435 @Override
436 public GetResponse get(RpcController controller, GetRequest request)
437 throws ServiceException {
438 boolean metaRegion = isMetaRegion(request.getRegion().getValue().toByteArray(),
439 request.getRegion().getType());
440 if (!metaRegion) {
441 return doGetResponse(request);
442 }
443 return doMetaGetResponse(meta, request);
444 }
445
446 private GetResponse doGetResponse(GetRequest request) {
447 ClientProtos.Result.Builder resultBuilder = ClientProtos.Result.newBuilder();
448 ByteString row = request.getGet().getRow();
449 resultBuilder.addCell(getStartCode(row));
450 GetResponse.Builder builder = GetResponse.newBuilder();
451 builder.setResult(resultBuilder.build());
452 return builder.build();
453 }
454
455 @Override
456 public MutateResponse mutate(RpcController controller,
457 MutateRequest request) throws ServiceException {
458 throw new NotImplementedException();
459 }
460
461 @Override
462 public ScanResponse scan(RpcController controller,
463 ScanRequest request) throws ServiceException {
464
465
466 return doMetaScanResponse(meta, sequenceids, request);
467 }
468
469 @Override
470 public BulkLoadHFileResponse bulkLoadHFile(
471 RpcController controller, BulkLoadHFileRequest request)
472 throws ServiceException {
473 throw new NotImplementedException();
474 }
475
476 @Override
477 public CoprocessorServiceResponse execService(
478 RpcController controller, CoprocessorServiceRequest request)
479 throws ServiceException {
480 throw new NotImplementedException();
481 }
482
483 @Override
484 public MultiResponse multi(RpcController controller, MultiRequest request)
485 throws ServiceException {
486 int concurrentInvocations = this.multiInvocationsCount.incrementAndGet();
487 try {
488 if (concurrentInvocations >= tooManyMultiRequests) {
489 throw new ServiceException(new RegionTooBusyException("concurrentInvocations=" +
490 concurrentInvocations));
491 }
492 Threads.sleep(multiPause);
493 return doMultiResponse(meta, sequenceids, request);
494 } finally {
495 this.multiInvocationsCount.decrementAndGet();
496 }
497 }
498
499 @Override
500 public CoprocessorServiceResponse execRegionServerService(RpcController controller,
501 CoprocessorServiceRequest request) throws ServiceException {
502 throw new NotImplementedException();
503 }
504 }
505
506 static ScanResponse doMetaScanResponse(final SortedMap<byte [], Pair<HRegionInfo, ServerName>> meta,
507 final AtomicLong sequenceids, final ScanRequest request) {
508 ScanResponse.Builder builder = ScanResponse.newBuilder();
509 int max = request.getNumberOfRows();
510 int count = 0;
511 Map<byte [], Pair<HRegionInfo, ServerName>> tail =
512 request.hasScan()? meta.tailMap(request.getScan().getStartRow().toByteArray()): meta;
513 ClientProtos.Result.Builder resultBuilder = ClientProtos.Result.newBuilder();
514 for (Map.Entry<byte [], Pair<HRegionInfo, ServerName>> e: tail.entrySet()) {
515
516 if (max <= 0) break;
517 if (++count > max) break;
518 HRegionInfo hri = e.getValue().getFirst();
519 ByteString row = ByteStringer.wrap(hri.getRegionName());
520 resultBuilder.clear();
521 resultBuilder.addCell(getRegionInfo(row, hri));
522 resultBuilder.addCell(getServer(row, e.getValue().getSecond()));
523 resultBuilder.addCell(getStartCode(row));
524 builder.addResults(resultBuilder.build());
525
526 if (hri.getEndKey().length <= 0) builder.setMoreResults(false);
527 else builder.setMoreResults(true);
528 }
529
530 builder.setScannerId(request.hasScannerId()?
531 request.getScannerId(): sequenceids.incrementAndGet());
532 return builder.build();
533 }
534
535 static GetResponse doMetaGetResponse(final SortedMap<byte [], Pair<HRegionInfo, ServerName>> meta,
536 final GetRequest request) {
537 ClientProtos.Result.Builder resultBuilder = ClientProtos.Result.newBuilder();
538 ByteString row = request.getGet().getRow();
539 Pair<HRegionInfo, ServerName> p = meta.get(row.toByteArray());
540 if (p == null) {
541 if (request.getGet().getClosestRowBefore()) {
542 byte [] bytes = row.toByteArray();
543 SortedMap<byte [], Pair<HRegionInfo, ServerName>> head =
544 bytes != null? meta.headMap(bytes): meta;
545 p = head == null? null: head.get(head.lastKey());
546 }
547 }
548 if (p != null) {
549 resultBuilder.addCell(getRegionInfo(row, p.getFirst()));
550 resultBuilder.addCell(getServer(row, p.getSecond()));
551 }
552 resultBuilder.addCell(getStartCode(row));
553 GetResponse.Builder builder = GetResponse.newBuilder();
554 builder.setResult(resultBuilder.build());
555 return builder.build();
556 }
557
558
559
560
561
562
563 static boolean isMetaRegion(final byte [] name, final RegionSpecifierType type) {
564 switch (type) {
565 case REGION_NAME:
566 return Bytes.equals(HRegionInfo.FIRST_META_REGIONINFO.getRegionName(), name);
567 case ENCODED_REGION_NAME:
568 return Bytes.equals(HRegionInfo.FIRST_META_REGIONINFO.getEncodedNameAsBytes(), name);
569 default: throw new UnsupportedOperationException();
570 }
571 }
572
573 private final static ByteString CATALOG_FAMILY_BYTESTRING =
574 ByteStringer.wrap(HConstants.CATALOG_FAMILY);
575 private final static ByteString REGIONINFO_QUALIFIER_BYTESTRING =
576 ByteStringer.wrap(HConstants.REGIONINFO_QUALIFIER);
577 private final static ByteString SERVER_QUALIFIER_BYTESTRING =
578 ByteStringer.wrap(HConstants.SERVER_QUALIFIER);
579
580 static CellProtos.Cell.Builder getBaseCellBuilder(final ByteString row) {
581 CellProtos.Cell.Builder cellBuilder = CellProtos.Cell.newBuilder();
582 cellBuilder.setRow(row);
583 cellBuilder.setFamily(CATALOG_FAMILY_BYTESTRING);
584 cellBuilder.setTimestamp(System.currentTimeMillis());
585 return cellBuilder;
586 }
587
588 static CellProtos.Cell getRegionInfo(final ByteString row, final HRegionInfo hri) {
589 CellProtos.Cell.Builder cellBuilder = getBaseCellBuilder(row);
590 cellBuilder.setQualifier(REGIONINFO_QUALIFIER_BYTESTRING);
591 cellBuilder.setValue(ByteStringer.wrap(hri.toByteArray()));
592 return cellBuilder.build();
593 }
594
595 static CellProtos.Cell getServer(final ByteString row, final ServerName sn) {
596 CellProtos.Cell.Builder cellBuilder = getBaseCellBuilder(row);
597 cellBuilder.setQualifier(SERVER_QUALIFIER_BYTESTRING);
598 cellBuilder.setValue(ByteString.copyFromUtf8(sn.getHostAndPort()));
599 return cellBuilder.build();
600 }
601
602 static CellProtos.Cell getStartCode(final ByteString row) {
603 CellProtos.Cell.Builder cellBuilder = getBaseCellBuilder(row);
604 cellBuilder.setQualifier(ByteStringer.wrap(HConstants.STARTCODE_QUALIFIER));
605
606 cellBuilder.setValue(ByteStringer.wrap(Bytes.toBytes(META_SERVERNAME.getStartcode())));
607 return cellBuilder.build();
608 }
609
610 private static final byte [] BIG_USER_TABLE = Bytes.toBytes("t");
611
612
613
614
615
616
617
618
619 private static byte [] format(final long number) {
620 byte [] b = new byte[10];
621 long d = number;
622 for (int i = b.length - 1; i >= 0; i--) {
623 b[i] = (byte)((d % 10) + '0');
624 d /= 10;
625 }
626 return b;
627 }
628
629
630
631
632
633
634 private static HRegionInfo [] makeHRegionInfos(final byte [] tableName, final int count,
635 final long namespaceSpan) {
636 byte [] startKey = HConstants.EMPTY_BYTE_ARRAY;
637 byte [] endKey = HConstants.EMPTY_BYTE_ARRAY;
638 long interval = namespaceSpan / count;
639 HRegionInfo [] hris = new HRegionInfo[count];
640 for (int i = 0; i < count; i++) {
641 if (i == 0) {
642 endKey = format(interval);
643 } else {
644 startKey = endKey;
645 if (i == count - 1) endKey = HConstants.EMPTY_BYTE_ARRAY;
646 else endKey = format((i + 1) * interval);
647 }
648 hris[i] = new HRegionInfo(TableName.valueOf(tableName), startKey, endKey);
649 }
650 return hris;
651 }
652
653
654
655
656
657 private static ServerName [] makeServerNames(final int count) {
658 ServerName [] sns = new ServerName[count];
659 for (int i = 0; i < count; i++) {
660 sns[i] = ServerName.valueOf("" + i + ".example.org", 60010, i);
661 }
662 return sns;
663 }
664
665
666
667
668 private static class MetaRowsComparator implements Comparator<byte []> {
669 private final KeyValue.KVComparator delegate = new KeyValue.MetaComparator();
670 @Override
671 public int compare(byte[] left, byte[] right) {
672 return delegate.compareRows(left, 0, left.length, right, 0, right.length);
673 }
674 }
675
676
677
678
679
680
681 static SortedMap<byte [], Pair<HRegionInfo, ServerName>> makeMeta(final byte [] tableName,
682 final int regionCount, final long namespaceSpan, final int serverCount) {
683
684 SortedMap<byte [], Pair<HRegionInfo, ServerName>> meta =
685 new ConcurrentSkipListMap<byte[], Pair<HRegionInfo,ServerName>>(new MetaRowsComparator());
686 HRegionInfo [] hris = makeHRegionInfos(tableName, regionCount, namespaceSpan);
687 ServerName [] serverNames = makeServerNames(serverCount);
688 int per = regionCount / serverCount;
689 int count = 0;
690 for (HRegionInfo hri: hris) {
691 Pair<HRegionInfo, ServerName> p =
692 new Pair<HRegionInfo, ServerName>(hri, serverNames[count++ / per]);
693 meta.put(hri.getRegionName(), p);
694 }
695 return meta;
696 }
697
698
699
700
701
702
703
704
705
706 static void cycle(int id, final Configuration c, final HConnection sharedConnection) throws IOException {
707 HTableInterface table = sharedConnection.getTable(BIG_USER_TABLE);
708 table.setAutoFlushTo(false);
709 long namespaceSpan = c.getLong("hbase.test.namespace.span", 1000000);
710 long startTime = System.currentTimeMillis();
711 final int printInterval = 100000;
712 Random rd = new Random(id);
713 boolean get = c.getBoolean("hbase.test.do.gets", false);
714 try {
715 Stopwatch stopWatch = new Stopwatch();
716 stopWatch.start();
717 for (int i = 0; i < namespaceSpan; i++) {
718 byte [] b = format(rd.nextLong());
719 if (get){
720 Get g = new Get(b);
721 table.get(g);
722 } else {
723 Put p = new Put(b);
724 p.add(HConstants.CATALOG_FAMILY, b, b);
725 table.put(p);
726 }
727 if (i % printInterval == 0) {
728 LOG.info("Put " + printInterval + "/" + stopWatch.elapsedMillis());
729 stopWatch.reset();
730 stopWatch.start();
731 }
732 }
733 LOG.info("Finished a cycle putting " + namespaceSpan + " in " +
734 (System.currentTimeMillis() - startTime) + "ms");
735 } finally {
736 table.close();
737 }
738 }
739
740 @Override
741 public int run(String[] arg0) throws Exception {
742 int errCode = 0;
743
744
745 final int servers = 1;
746
747 final int regions = 100000;
748
749 final long namespaceSpan = 50000000;
750
751
752 final long multiPause = 0;
753
754 if ((namespaceSpan < regions) || (regions < servers)) {
755 throw new IllegalArgumentException("namespaceSpan=" + namespaceSpan + " must be > regions=" +
756 regions + " which must be > servers=" + servers);
757 }
758
759
760 getConf().set("hbase.client.connection.impl",
761 ManyServersManyRegionsConnection.class.getName());
762
763 getConf().set("hbase.client.registry.impl", SimpleRegistry.class.getName());
764
765
766
767 getConf().setInt("hbase.client.start.log.errors.counter", 0);
768
769
770 getConf().setInt("hbase.test.regions", regions);
771 getConf().setLong("hbase.test.namespace.span", namespaceSpan);
772 getConf().setLong("hbase.test.servers", servers);
773 getConf().set("hbase.test.tablename", Bytes.toString(BIG_USER_TABLE));
774 getConf().setLong("hbase.test.multi.pause.when.done", multiPause);
775
776 getConf().setInt("hbase.test.multi.too.many", 10);
777 final int clients = 2;
778
779
780
781 final ExecutorService pool = Executors.newCachedThreadPool(Threads.getNamedThreadFactory("p"));
782
783
784 final HConnection sharedConnection = HConnectionManager.createConnection(getConf()
785 try {
786 Thread [] ts = new Thread[clients];
787 for (int j = 0; j < ts.length; j++) {
788 final int id = j;
789 ts[j] = new Thread("" + j) {
790 final Configuration c = getConf();
791
792 @Override
793 public void run() {
794 try {
795 cycle(id, c, sharedConnection);
796 } catch (IOException e) {
797 e.printStackTrace();
798 }
799 }
800 };
801 ts[j].start();
802 }
803 for (int j = 0; j < ts.length; j++) {
804 ts[j].join();
805 }
806 } finally {
807 sharedConnection.close();
808 }
809 return errCode;
810 }
811
812
813
814
815
816
817 public static void main(String[] args) throws Exception {
818 System.exit(ToolRunner.run(HBaseConfiguration.create(), new TestClientNoCluster(), args));
819 }
820 }