1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18 package org.apache.hadoop.hbase;
19
20 import java.io.File;
21 import java.io.IOException;
22 import java.io.OutputStream;
23 import java.lang.reflect.Field;
24 import java.lang.reflect.Modifier;
25 import java.net.InetAddress;
26 import java.net.ServerSocket;
27 import java.net.Socket;
28 import java.net.UnknownHostException;
29 import java.security.MessageDigest;
30 import java.util.ArrayList;
31 import java.util.Arrays;
32 import java.util.Collection;
33 import java.util.Collections;
34 import java.util.HashSet;
35 import java.util.List;
36 import java.util.Map;
37 import java.util.NavigableSet;
38 import java.util.Random;
39 import java.util.Set;
40 import java.util.TreeSet;
41 import java.util.UUID;
42 import java.util.concurrent.TimeUnit;
43
44 import org.apache.commons.lang.RandomStringUtils;
45 import org.apache.commons.logging.Log;
46 import org.apache.commons.logging.LogFactory;
47 import org.apache.commons.logging.impl.Jdk14Logger;
48 import org.apache.commons.logging.impl.Log4JLogger;
49 import org.apache.hadoop.conf.Configuration;
50 import org.apache.hadoop.fs.FileSystem;
51 import org.apache.hadoop.fs.Path;
52 import org.apache.hadoop.hbase.Waiter.ExplainingPredicate;
53 import org.apache.hadoop.hbase.Waiter.Predicate;
54 import org.apache.hadoop.hbase.classification.InterfaceAudience;
55 import org.apache.hadoop.hbase.classification.InterfaceStability;
56 import org.apache.hadoop.hbase.client.Admin;
57 import org.apache.hadoop.hbase.client.Connection;
58 import org.apache.hadoop.hbase.client.ConnectionFactory;
59 import org.apache.hadoop.hbase.client.Consistency;
60 import org.apache.hadoop.hbase.client.Delete;
61 import org.apache.hadoop.hbase.client.Durability;
62 import org.apache.hadoop.hbase.client.Get;
63 import org.apache.hadoop.hbase.client.HBaseAdmin;
64 import org.apache.hadoop.hbase.client.HConnection;
65 import org.apache.hadoop.hbase.client.HTable;
66 import org.apache.hadoop.hbase.client.Put;
67 import org.apache.hadoop.hbase.client.RegionLocator;
68 import org.apache.hadoop.hbase.client.Result;
69 import org.apache.hadoop.hbase.client.ResultScanner;
70 import org.apache.hadoop.hbase.client.Scan;
71 import org.apache.hadoop.hbase.client.Table;
72 import org.apache.hadoop.hbase.fs.HFileSystem;
73 import org.apache.hadoop.hbase.io.compress.Compression;
74 import org.apache.hadoop.hbase.io.compress.Compression.Algorithm;
75 import org.apache.hadoop.hbase.io.encoding.DataBlockEncoding;
76 import org.apache.hadoop.hbase.io.hfile.ChecksumUtil;
77 import org.apache.hadoop.hbase.io.hfile.HFile;
78 import org.apache.hadoop.hbase.ipc.RpcServerInterface;
79 import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
80 import org.apache.hadoop.hbase.mapreduce.MapreduceTestingShim;
81 import org.apache.hadoop.hbase.master.HMaster;
82 import org.apache.hadoop.hbase.master.RegionStates;
83 import org.apache.hadoop.hbase.master.ServerManager;
84 import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
85 import org.apache.hadoop.hbase.regionserver.BloomType;
86 import org.apache.hadoop.hbase.regionserver.HRegion;
87 import org.apache.hadoop.hbase.regionserver.HRegionServer;
88 import org.apache.hadoop.hbase.regionserver.HStore;
89 import org.apache.hadoop.hbase.regionserver.InternalScanner;
90 import org.apache.hadoop.hbase.regionserver.Region;
91 import org.apache.hadoop.hbase.regionserver.RegionServerServices;
92 import org.apache.hadoop.hbase.regionserver.RegionServerStoppedException;
93 import org.apache.hadoop.hbase.regionserver.wal.MetricsWAL;
94 import org.apache.hadoop.hbase.regionserver.wal.WALActionsListener;
95 import org.apache.hadoop.hbase.security.User;
96 import org.apache.hadoop.hbase.security.visibility.VisibilityLabelsCache;
97 import org.apache.hadoop.hbase.tool.Canary;
98 import org.apache.hadoop.hbase.util.Bytes;
99 import org.apache.hadoop.hbase.util.FSTableDescriptors;
100 import org.apache.hadoop.hbase.util.FSUtils;
101 import org.apache.hadoop.hbase.util.JVMClusterUtil;
102 import org.apache.hadoop.hbase.util.JVMClusterUtil.MasterThread;
103 import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread;
104 import org.apache.hadoop.hbase.util.Pair;
105 import org.apache.hadoop.hbase.util.RegionSplitter;
106 import org.apache.hadoop.hbase.util.RetryCounter;
107 import org.apache.hadoop.hbase.util.Threads;
108 import org.apache.hadoop.hbase.wal.WAL;
109 import org.apache.hadoop.hbase.wal.WALFactory;
110 import org.apache.hadoop.hbase.zookeeper.EmptyWatcher;
111 import org.apache.hadoop.hbase.zookeeper.MiniZooKeeperCluster;
112 import org.apache.hadoop.hbase.zookeeper.ZKAssign;
113 import org.apache.hadoop.hbase.zookeeper.ZKConfig;
114 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
115 import org.apache.hadoop.hdfs.DFSClient;
116 import org.apache.hadoop.hdfs.DistributedFileSystem;
117 import org.apache.hadoop.hdfs.MiniDFSCluster;
118 import org.apache.hadoop.hdfs.server.namenode.EditLogFileOutputStream;
119 import org.apache.hadoop.mapred.JobConf;
120 import org.apache.hadoop.mapred.MiniMRCluster;
121 import org.apache.hadoop.mapred.TaskLog;
122 import org.apache.zookeeper.KeeperException;
123 import org.apache.zookeeper.KeeperException.NodeExistsException;
124 import org.apache.zookeeper.WatchedEvent;
125 import org.apache.zookeeper.ZooKeeper;
126 import org.apache.zookeeper.ZooKeeper.States;
127
128 import static org.junit.Assert.assertEquals;
129 import static org.junit.Assert.assertTrue;
130 import static org.junit.Assert.fail;
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146 @InterfaceAudience.Public
147 @InterfaceStability.Evolving
148 @SuppressWarnings("deprecation")
149 public class HBaseTestingUtility extends HBaseCommonTestingUtility {
150 private MiniZooKeeperCluster zkCluster = null;
151
152 public static final String REGIONS_PER_SERVER_KEY = "hbase.test.regions-per-server";
153
154
155
156
157 public static final int DEFAULT_REGIONS_PER_SERVER = 5;
158
159
160
161
162
163 private boolean passedZkCluster = false;
164 private MiniDFSCluster dfsCluster = null;
165
166 private volatile HBaseCluster hbaseCluster = null;
167 private MiniMRCluster mrCluster = null;
168
169
170 private volatile boolean miniClusterRunning;
171
172 private String hadoopLogDir;
173
174
175 private File clusterTestDir = null;
176
177
178
179 private Path dataTestDirOnTestFS = null;
180
181
182
183
184 private volatile Connection connection;
185
186
187
188
189
190
191
192
193 @Deprecated
194 private static final String TEST_DIRECTORY_KEY = "test.build.data";
195
196
197 private static String FS_URI;
198
199
200 private static final Set<Integer> takenRandomPorts = new HashSet<Integer>();
201
202
203 public static final List<Object[]> COMPRESSION_ALGORITHMS_PARAMETERIZED =
204 Arrays.asList(new Object[][] {
205 { Compression.Algorithm.NONE },
206 { Compression.Algorithm.GZ }
207 });
208
209
210 public static final List<Object[]> BOOLEAN_PARAMETERIZED =
211 Arrays.asList(new Object[][] {
212 { new Boolean(false) },
213 { new Boolean(true) }
214 });
215
216
217 public static final List<Object[]> MEMSTORETS_TAGS_PARAMETRIZED = memStoreTSAndTagsCombination() ;
218
219 public static final Compression.Algorithm[] COMPRESSION_ALGORITHMS ={
220 Compression.Algorithm.NONE, Compression.Algorithm.GZ
221 };
222
223
224
225
226
227 private static List<Object[]> bloomAndCompressionCombinations() {
228 List<Object[]> configurations = new ArrayList<Object[]>();
229 for (Compression.Algorithm comprAlgo :
230 HBaseTestingUtility.COMPRESSION_ALGORITHMS) {
231 for (BloomType bloomType : BloomType.values()) {
232 configurations.add(new Object[] { comprAlgo, bloomType });
233 }
234 }
235 return Collections.unmodifiableList(configurations);
236 }
237
238
239
240
241 private static List<Object[]> memStoreTSAndTagsCombination() {
242 List<Object[]> configurations = new ArrayList<Object[]>();
243 configurations.add(new Object[] { false, false });
244 configurations.add(new Object[] { false, true });
245 configurations.add(new Object[] { true, false });
246 configurations.add(new Object[] { true, true });
247 return Collections.unmodifiableList(configurations);
248 }
249
250 public static final Collection<Object[]> BLOOM_AND_COMPRESSION_COMBINATIONS =
251 bloomAndCompressionCombinations();
252
253 public HBaseTestingUtility() {
254 this(HBaseConfiguration.create());
255 }
256
257 public HBaseTestingUtility(Configuration conf) {
258 super(conf);
259
260
261 ChecksumUtil.generateExceptionForChecksumFailureForTest(true);
262 }
263
264
265
266
267
268
269
270 public static HBaseTestingUtility createLocalHTU() {
271 Configuration c = HBaseConfiguration.create();
272 return createLocalHTU(c);
273 }
274
275
276
277
278
279
280
281
282 public static HBaseTestingUtility createLocalHTU(Configuration c) {
283 HBaseTestingUtility htu = new HBaseTestingUtility(c);
284 String dataTestDir = htu.getDataTestDir().toString();
285 htu.getConfiguration().set(HConstants.HBASE_DIR, dataTestDir);
286 LOG.debug("Setting " + HConstants.HBASE_DIR + " to " + dataTestDir);
287 return htu;
288 }
289
290
291
292
293 public static void closeRegion(final Region r) throws IOException {
294 if (r != null) {
295 ((HRegion)r).close();
296 }
297 }
298
299
300
301
302
303
304
305
306
307
308
309
310 @Override
311 public Configuration getConfiguration() {
312 return super.getConfiguration();
313 }
314
315 public void setHBaseCluster(HBaseCluster hbaseCluster) {
316 this.hbaseCluster = hbaseCluster;
317 }
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335 @Override
336 protected Path setupDataTestDir() {
337 Path testPath = super.setupDataTestDir();
338 if (null == testPath) {
339 return null;
340 }
341
342 createSubDirAndSystemProperty(
343 "hadoop.log.dir",
344 testPath, "hadoop-log-dir");
345
346
347
348 createSubDirAndSystemProperty(
349 "hadoop.tmp.dir",
350 testPath, "hadoop-tmp-dir");
351
352
353 createSubDir(
354 "mapreduce.cluster.local.dir",
355 testPath, "mapred-local-dir");
356
357 return testPath;
358 }
359
360 private void createSubDirAndSystemProperty(
361 String propertyName, Path parent, String subDirName){
362
363 String sysValue = System.getProperty(propertyName);
364
365 if (sysValue != null) {
366
367
368 LOG.info("System.getProperty(\""+propertyName+"\") already set to: "+
369 sysValue + " so I do NOT create it in " + parent);
370 String confValue = conf.get(propertyName);
371 if (confValue != null && !confValue.endsWith(sysValue)){
372 LOG.warn(
373 propertyName + " property value differs in configuration and system: "+
374 "Configuration="+confValue+" while System="+sysValue+
375 " Erasing configuration value by system value."
376 );
377 }
378 conf.set(propertyName, sysValue);
379 } else {
380
381 createSubDir(propertyName, parent, subDirName);
382 System.setProperty(propertyName, conf.get(propertyName));
383 }
384 }
385
386
387
388
389
390
391
392 private Path getBaseTestDirOnTestFS() throws IOException {
393 FileSystem fs = getTestFileSystem();
394 return new Path(fs.getWorkingDirectory(), "test-data");
395 }
396
397
398
399
400 public HTableDescriptor getMetaTableDescriptor() {
401 try {
402 return new FSTableDescriptors(conf).get(TableName.META_TABLE_NAME);
403 } catch (IOException e) {
404 throw new RuntimeException("Unable to create META table descriptor", e);
405 }
406 }
407
408
409
410
411
412
413 Path getClusterTestDir() {
414 if (clusterTestDir == null){
415 setupClusterTestDir();
416 }
417 return new Path(clusterTestDir.getAbsolutePath());
418 }
419
420
421
422
423 private void setupClusterTestDir() {
424 if (clusterTestDir != null) {
425 return;
426 }
427
428
429
430 Path testDir = getDataTestDir("dfscluster_" + UUID.randomUUID().toString());
431 clusterTestDir = new File(testDir.toString()).getAbsoluteFile();
432
433 boolean b = deleteOnExit();
434 if (b) clusterTestDir.deleteOnExit();
435 conf.set(TEST_DIRECTORY_KEY, clusterTestDir.getPath());
436 LOG.info("Created new mini-cluster data directory: " + clusterTestDir + ", deleteOnExit=" + b);
437 }
438
439
440
441
442
443
444
445 public Path getDataTestDirOnTestFS() throws IOException {
446 if (dataTestDirOnTestFS == null) {
447 setupDataTestDirOnTestFS();
448 }
449
450 return dataTestDirOnTestFS;
451 }
452
453
454
455
456
457
458
459
460 public Path getDataTestDirOnTestFS(final String subdirName) throws IOException {
461 return new Path(getDataTestDirOnTestFS(), subdirName);
462 }
463
464
465
466
467
468 private void setupDataTestDirOnTestFS() throws IOException {
469 if (dataTestDirOnTestFS != null) {
470 LOG.warn("Data test on test fs dir already setup in "
471 + dataTestDirOnTestFS.toString());
472 return;
473 }
474 dataTestDirOnTestFS = getNewDataTestDirOnTestFS();
475 }
476
477
478
479
480 private Path getNewDataTestDirOnTestFS() throws IOException {
481
482
483
484
485 FileSystem fs = getTestFileSystem();
486 Path newDataTestDir = null;
487 if (fs.getUri().getScheme().equals(FileSystem.getLocal(conf).getUri().getScheme())) {
488 File dataTestDir = new File(getDataTestDir().toString());
489 if (deleteOnExit()) dataTestDir.deleteOnExit();
490 newDataTestDir = new Path(dataTestDir.getAbsolutePath());
491 } else {
492 Path base = getBaseTestDirOnTestFS();
493 String randomStr = UUID.randomUUID().toString();
494 newDataTestDir = new Path(base, randomStr);
495 if (deleteOnExit()) fs.deleteOnExit(newDataTestDir);
496 }
497 return newDataTestDir;
498 }
499
500
501
502
503
504
505 public boolean cleanupDataTestDirOnTestFS() throws IOException {
506 boolean ret = getTestFileSystem().delete(dataTestDirOnTestFS, true);
507 if (ret)
508 dataTestDirOnTestFS = null;
509 return ret;
510 }
511
512
513
514
515
516
517 public boolean cleanupDataTestDirOnTestFS(String subdirName) throws IOException {
518 Path cpath = getDataTestDirOnTestFS(subdirName);
519 return getTestFileSystem().delete(cpath, true);
520 }
521
522
523
524
525
526
527
528
529 public MiniDFSCluster startMiniDFSCluster(int servers) throws Exception {
530 return startMiniDFSCluster(servers, null);
531 }
532
533
534
535
536
537
538
539
540
541
542
543
544 public MiniDFSCluster startMiniDFSCluster(final String hosts[])
545 throws Exception {
546 if ( hosts != null && hosts.length != 0) {
547 return startMiniDFSCluster(hosts.length, hosts);
548 } else {
549 return startMiniDFSCluster(1, null);
550 }
551 }
552
553
554
555
556
557
558
559
560
561
562 public MiniDFSCluster startMiniDFSCluster(int servers, final String hosts[])
563 throws Exception {
564 createDirsAndSetProperties();
565 EditLogFileOutputStream.setShouldSkipFsyncForTesting(true);
566
567
568 org.apache.log4j.Logger.getLogger(org.apache.hadoop.metrics2.util.MBeans.class).
569 setLevel(org.apache.log4j.Level.ERROR);
570 org.apache.log4j.Logger.getLogger(org.apache.hadoop.metrics2.impl.MetricsSystemImpl.class).
571 setLevel(org.apache.log4j.Level.ERROR);
572
573
574 this.dfsCluster = new MiniDFSCluster(0, this.conf, servers, true, true,
575 true, null, null, hosts, null);
576
577
578 setFs();
579
580
581 this.dfsCluster.waitClusterUp();
582
583
584 dataTestDirOnTestFS = null;
585
586 return this.dfsCluster;
587 }
588
589 private void setFs() throws IOException {
590 if(this.dfsCluster == null){
591 LOG.info("Skipping setting fs because dfsCluster is null");
592 return;
593 }
594 FileSystem fs = this.dfsCluster.getFileSystem();
595 FSUtils.setFsDefault(this.conf, new Path(fs.getUri()));
596 }
597
598 public MiniDFSCluster startMiniDFSCluster(int servers, final String racks[], String hosts[])
599 throws Exception {
600 createDirsAndSetProperties();
601 this.dfsCluster = new MiniDFSCluster(0, this.conf, servers, true, true,
602 true, null, racks, hosts, null);
603
604
605 FileSystem fs = this.dfsCluster.getFileSystem();
606 FSUtils.setFsDefault(this.conf, new Path(fs.getUri()));
607
608
609 this.dfsCluster.waitClusterUp();
610
611
612 dataTestDirOnTestFS = null;
613
614 return this.dfsCluster;
615 }
616
617 public MiniDFSCluster startMiniDFSClusterForTestWAL(int namenodePort) throws IOException {
618 createDirsAndSetProperties();
619 dfsCluster = new MiniDFSCluster(namenodePort, conf, 5, false, true, true, null,
620 null, null, null);
621 return dfsCluster;
622 }
623
624
625 private void createDirsAndSetProperties() throws IOException {
626 setupClusterTestDir();
627 System.setProperty(TEST_DIRECTORY_KEY, clusterTestDir.getPath());
628 createDirAndSetProperty("cache_data", "test.cache.data");
629 createDirAndSetProperty("hadoop_tmp", "hadoop.tmp.dir");
630 hadoopLogDir = createDirAndSetProperty("hadoop_logs", "hadoop.log.dir");
631 createDirAndSetProperty("mapred_local", "mapreduce.cluster.local.dir");
632 createDirAndSetProperty("mapred_temp", "mapreduce.cluster.temp.dir");
633 enableShortCircuit();
634
635 Path root = getDataTestDirOnTestFS("hadoop");
636 conf.set(MapreduceTestingShim.getMROutputDirProp(),
637 new Path(root, "mapred-output-dir").toString());
638 conf.set("mapreduce.jobtracker.system.dir", new Path(root, "mapred-system-dir").toString());
639 conf.set("mapreduce.jobtracker.staging.root.dir",
640 new Path(root, "mapreduce-jobtracker-staging-root-dir").toString());
641 conf.set("mapreduce.job.working.dir", new Path(root, "mapred-working-dir").toString());
642 }
643
644
645
646
647
648
649
650 public boolean isReadShortCircuitOn(){
651 final String propName = "hbase.tests.use.shortcircuit.reads";
652 String readOnProp = System.getProperty(propName);
653 if (readOnProp != null){
654 return Boolean.parseBoolean(readOnProp);
655 } else {
656 return conf.getBoolean(propName, false);
657 }
658 }
659
660
661
662
663 private void enableShortCircuit() {
664 if (isReadShortCircuitOn()) {
665 String curUser = System.getProperty("user.name");
666 LOG.info("read short circuit is ON for user " + curUser);
667
668 conf.set("dfs.block.local-path-access.user", curUser);
669
670 conf.setBoolean("dfs.client.read.shortcircuit", true);
671
672 conf.setBoolean("dfs.client.read.shortcircuit.skip.checksum", true);
673 } else {
674 LOG.info("read short circuit is OFF");
675 }
676 }
677
678 private String createDirAndSetProperty(final String relPath, String property) {
679 String path = getDataTestDir(relPath).toString();
680 System.setProperty(property, path);
681 conf.set(property, path);
682 new File(path).mkdirs();
683 LOG.info("Setting " + property + " to " + path + " in system properties and HBase conf");
684 return path;
685 }
686
687
688
689
690
691
692 public void shutdownMiniDFSCluster() throws IOException {
693 if (this.dfsCluster != null) {
694
695 this.dfsCluster.shutdown();
696 dfsCluster = null;
697 dataTestDirOnTestFS = null;
698 FSUtils.setFsDefault(this.conf, new Path("file:///"));
699 }
700 }
701
702
703
704
705
706
707
708
709 public MiniZooKeeperCluster startMiniZKCluster() throws Exception {
710 return startMiniZKCluster(1);
711 }
712
713
714
715
716
717
718
719
720
721 public MiniZooKeeperCluster startMiniZKCluster(
722 final int zooKeeperServerNum,
723 final int ... clientPortList)
724 throws Exception {
725 setupClusterTestDir();
726 return startMiniZKCluster(clusterTestDir, zooKeeperServerNum, clientPortList);
727 }
728
729 private MiniZooKeeperCluster startMiniZKCluster(final File dir)
730 throws Exception {
731 return startMiniZKCluster(dir, 1, null);
732 }
733
734
735
736
737
738 private MiniZooKeeperCluster startMiniZKCluster(final File dir,
739 final int zooKeeperServerNum,
740 final int [] clientPortList)
741 throws Exception {
742 if (this.zkCluster != null) {
743 throw new IOException("Cluster already running at " + dir);
744 }
745 this.passedZkCluster = false;
746 this.zkCluster = new MiniZooKeeperCluster(this.getConfiguration());
747 final int defPort = this.conf.getInt("test.hbase.zookeeper.property.clientPort", 0);
748 if (defPort > 0){
749
750 this.zkCluster.setDefaultClientPort(defPort);
751 }
752
753 if (clientPortList != null) {
754
755 int clientPortListSize = (clientPortList.length <= zooKeeperServerNum) ?
756 clientPortList.length : zooKeeperServerNum;
757 for (int i=0; i < clientPortListSize; i++) {
758 this.zkCluster.addClientPort(clientPortList[i]);
759 }
760 }
761 int clientPort = this.zkCluster.startup(dir,zooKeeperServerNum);
762 this.conf.set(HConstants.ZOOKEEPER_CLIENT_PORT,
763 Integer.toString(clientPort));
764 return this.zkCluster;
765 }
766
767
768
769
770
771
772
773 public void shutdownMiniZKCluster() throws IOException {
774 if (this.zkCluster != null) {
775 this.zkCluster.shutdown();
776 this.zkCluster = null;
777 }
778 }
779
780
781
782
783
784
785
786 public MiniHBaseCluster startMiniCluster() throws Exception {
787 return startMiniCluster(1, 1);
788 }
789
790
791
792
793
794
795
796
797
798 public MiniHBaseCluster startMiniCluster(final int numSlaves, boolean create)
799 throws Exception {
800 return startMiniCluster(1, numSlaves, create);
801 }
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816 public MiniHBaseCluster startMiniCluster(final int numSlaves)
817 throws Exception {
818 return startMiniCluster(1, numSlaves, false);
819 }
820
821
822
823
824
825
826
827
828 public MiniHBaseCluster startMiniCluster(final int numMasters,
829 final int numSlaves, boolean create)
830 throws Exception {
831 return startMiniCluster(numMasters, numSlaves, null, create);
832 }
833
834
835
836
837
838
839
840 public MiniHBaseCluster startMiniCluster(final int numMasters,
841 final int numSlaves)
842 throws Exception {
843 return startMiniCluster(numMasters, numSlaves, null, false);
844 }
845
846 public MiniHBaseCluster startMiniCluster(final int numMasters,
847 final int numSlaves, final String[] dataNodeHosts, boolean create)
848 throws Exception {
849 return startMiniCluster(numMasters, numSlaves, numSlaves, dataNodeHosts,
850 null, null, create);
851 }
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877 public MiniHBaseCluster startMiniCluster(final int numMasters,
878 final int numSlaves, final String[] dataNodeHosts) throws Exception {
879 return startMiniCluster(numMasters, numSlaves, numSlaves, dataNodeHosts,
880 null, null);
881 }
882
883
884
885
886
887 public MiniHBaseCluster startMiniCluster(final int numMasters,
888 final int numSlaves, final int numDataNodes) throws Exception {
889 return startMiniCluster(numMasters, numSlaves, numDataNodes, null, null, null);
890 }
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919 public MiniHBaseCluster startMiniCluster(final int numMasters,
920 final int numSlaves, final String[] dataNodeHosts, Class<? extends HMaster> masterClass,
921 Class<? extends MiniHBaseCluster.MiniHBaseClusterRegionServer> regionserverClass)
922 throws Exception {
923 return startMiniCluster(
924 numMasters, numSlaves, numSlaves, dataNodeHosts, masterClass, regionserverClass);
925 }
926
927 public MiniHBaseCluster startMiniCluster(final int numMasters,
928 final int numSlaves, int numDataNodes, final String[] dataNodeHosts,
929 Class<? extends HMaster> masterClass,
930 Class<? extends MiniHBaseCluster.MiniHBaseClusterRegionServer> regionserverClass)
931 throws Exception {
932 return startMiniCluster(numMasters, numSlaves, numDataNodes, dataNodeHosts,
933 masterClass, regionserverClass, false);
934 }
935
936
937
938
939
940
941
942
943 public MiniHBaseCluster startMiniCluster(final int numMasters,
944 final int numSlaves, int numDataNodes, final String[] dataNodeHosts,
945 Class<? extends HMaster> masterClass,
946 Class<? extends MiniHBaseCluster.MiniHBaseClusterRegionServer> regionserverClass,
947 boolean create)
948 throws Exception {
949 if (dataNodeHosts != null && dataNodeHosts.length != 0) {
950 numDataNodes = dataNodeHosts.length;
951 }
952
953 LOG.info("Starting up minicluster with " + numMasters + " master(s) and " +
954 numSlaves + " regionserver(s) and " + numDataNodes + " datanode(s)");
955
956
957 if (miniClusterRunning) {
958 throw new IllegalStateException("A mini-cluster is already running");
959 }
960 miniClusterRunning = true;
961
962 setupClusterTestDir();
963 System.setProperty(TEST_DIRECTORY_KEY, this.clusterTestDir.getPath());
964
965
966
967 if(this.dfsCluster == null) {
968 dfsCluster = startMiniDFSCluster(numDataNodes, dataNodeHosts);
969 }
970
971
972 if (this.zkCluster == null) {
973 startMiniZKCluster(clusterTestDir);
974 }
975
976
977 return startMiniHBaseCluster(numMasters, numSlaves, masterClass,
978 regionserverClass, create);
979 }
980
981 public MiniHBaseCluster startMiniHBaseCluster(final int numMasters, final int numSlaves)
982 throws IOException, InterruptedException{
983 return startMiniHBaseCluster(numMasters, numSlaves, null, null, false);
984 }
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999 public MiniHBaseCluster startMiniHBaseCluster(final int numMasters,
1000 final int numSlaves, Class<? extends HMaster> masterClass,
1001 Class<? extends MiniHBaseCluster.MiniHBaseClusterRegionServer> regionserverClass,
1002 boolean create)
1003 throws IOException, InterruptedException {
1004
1005 createRootDir(create);
1006
1007
1008
1009 if (conf.getInt(ServerManager.WAIT_ON_REGIONSERVERS_MINTOSTART, -1) == -1) {
1010 conf.setInt(ServerManager.WAIT_ON_REGIONSERVERS_MINTOSTART, numSlaves);
1011 }
1012 if (conf.getInt(ServerManager.WAIT_ON_REGIONSERVERS_MAXTOSTART, -1) == -1) {
1013 conf.setInt(ServerManager.WAIT_ON_REGIONSERVERS_MAXTOSTART, numSlaves);
1014 }
1015
1016 Configuration c = new Configuration(this.conf);
1017 this.hbaseCluster =
1018 new MiniHBaseCluster(c, numMasters, numSlaves, masterClass, regionserverClass);
1019
1020 Table t = new HTable(c, TableName.META_TABLE_NAME);
1021 ResultScanner s = t.getScanner(new Scan());
1022 while (s.next() != null) {
1023 continue;
1024 }
1025 s.close();
1026 t.close();
1027
1028 getHBaseAdmin();
1029 LOG.info("Minicluster is up");
1030
1031
1032
1033 setHBaseFsTmpDir();
1034
1035 return (MiniHBaseCluster)this.hbaseCluster;
1036 }
1037
1038
1039
1040
1041
1042
1043
1044 public void restartHBaseCluster(int servers) throws IOException, InterruptedException {
1045 this.hbaseCluster = new MiniHBaseCluster(this.conf, servers);
1046
1047 Table t = new HTable(new Configuration(this.conf), TableName.META_TABLE_NAME);
1048 ResultScanner s = t.getScanner(new Scan());
1049 while (s.next() != null) {
1050
1051 }
1052 LOG.info("HBase has been restarted");
1053 s.close();
1054 t.close();
1055 }
1056
1057
1058
1059
1060
1061
1062 public MiniHBaseCluster getMiniHBaseCluster() {
1063 if (this.hbaseCluster == null || this.hbaseCluster instanceof MiniHBaseCluster) {
1064 return (MiniHBaseCluster)this.hbaseCluster;
1065 }
1066 throw new RuntimeException(hbaseCluster + " not an instance of " +
1067 MiniHBaseCluster.class.getName());
1068 }
1069
1070
1071
1072
1073
1074
1075 public void shutdownMiniCluster() throws Exception {
1076 LOG.info("Shutting down minicluster");
1077 if (this.connection != null && !this.connection.isClosed()) {
1078 this.connection.close();
1079 this.connection = null;
1080 }
1081 shutdownMiniHBaseCluster();
1082 if (!this.passedZkCluster){
1083 shutdownMiniZKCluster();
1084 }
1085 shutdownMiniDFSCluster();
1086
1087 cleanupTestDir();
1088 miniClusterRunning = false;
1089 LOG.info("Minicluster is down");
1090 }
1091
1092
1093
1094
1095
1096 @Override
1097 public boolean cleanupTestDir() throws IOException {
1098 boolean ret = super.cleanupTestDir();
1099 if (deleteDir(this.clusterTestDir)) {
1100 this.clusterTestDir = null;
1101 return ret & true;
1102 }
1103 return false;
1104 }
1105
1106
1107
1108
1109
1110 public void shutdownMiniHBaseCluster() throws IOException {
1111 if (hbaseAdmin != null) {
1112 hbaseAdmin.close0();
1113 hbaseAdmin = null;
1114 }
1115
1116
1117 conf.setInt(ServerManager.WAIT_ON_REGIONSERVERS_MINTOSTART, -1);
1118 conf.setInt(ServerManager.WAIT_ON_REGIONSERVERS_MAXTOSTART, -1);
1119 if (this.hbaseCluster != null) {
1120 this.hbaseCluster.shutdown();
1121
1122 this.hbaseCluster.waitUntilShutDown();
1123 this.hbaseCluster = null;
1124 }
1125
1126 if (zooKeeperWatcher != null) {
1127 zooKeeperWatcher.close();
1128 zooKeeperWatcher = null;
1129 }
1130 }
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140 public Path getDefaultRootDirPath(boolean create) throws IOException {
1141 if (!create) {
1142 return getDataTestDirOnTestFS();
1143 } else {
1144 return getNewDataTestDirOnTestFS();
1145 }
1146 }
1147
1148
1149
1150
1151
1152
1153
1154
1155 public Path getDefaultRootDirPath() throws IOException {
1156 return getDefaultRootDirPath(false);
1157 }
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171 public Path createRootDir(boolean create) throws IOException {
1172 FileSystem fs = FileSystem.get(this.conf);
1173 Path hbaseRootdir = getDefaultRootDirPath(create);
1174 FSUtils.setRootDir(this.conf, hbaseRootdir);
1175 fs.mkdirs(hbaseRootdir);
1176 FSUtils.setVersion(fs, hbaseRootdir);
1177 return hbaseRootdir;
1178 }
1179
1180
1181
1182
1183
1184
1185
1186 public Path createRootDir() throws IOException {
1187 return createRootDir(false);
1188 }
1189
1190
1191 private void setHBaseFsTmpDir() throws IOException {
1192 String hbaseFsTmpDirInString = this.conf.get("hbase.fs.tmp.dir");
1193 if (hbaseFsTmpDirInString == null) {
1194 this.conf.set("hbase.fs.tmp.dir", getDataTestDirOnTestFS("hbase-staging").toString());
1195 LOG.info("Setting hbase.fs.tmp.dir to " + this.conf.get("hbase.fs.tmp.dir"));
1196 } else {
1197 LOG.info("The hbase.fs.tmp.dir is set to " + hbaseFsTmpDirInString);
1198 }
1199 }
1200
1201
1202
1203
1204
1205 public void flush() throws IOException {
1206 getMiniHBaseCluster().flushcache();
1207 }
1208
1209
1210
1211
1212
1213 public void flush(TableName tableName) throws IOException {
1214 getMiniHBaseCluster().flushcache(tableName);
1215 }
1216
1217
1218
1219
1220
1221 public void compact(boolean major) throws IOException {
1222 getMiniHBaseCluster().compact(major);
1223 }
1224
1225
1226
1227
1228
1229 public void compact(TableName tableName, boolean major) throws IOException {
1230 getMiniHBaseCluster().compact(tableName, major);
1231 }
1232
1233
1234
1235
1236
1237
1238
1239
1240 public Table createTable(TableName tableName, String family)
1241 throws IOException{
1242 return createTable(tableName, new String[]{family});
1243 }
1244
1245
1246
1247
1248
1249
1250
1251
1252 public HTable createTable(byte[] tableName, byte[] family)
1253 throws IOException{
1254 return createTable(TableName.valueOf(tableName), new byte[][]{family});
1255 }
1256
1257
1258
1259
1260
1261
1262
1263
1264 public Table createTable(TableName tableName, String[] families)
1265 throws IOException {
1266 List<byte[]> fams = new ArrayList<byte[]>(families.length);
1267 for (String family : families) {
1268 fams.add(Bytes.toBytes(family));
1269 }
1270 return createTable(tableName, fams.toArray(new byte[0][]));
1271 }
1272
1273
1274
1275
1276
1277
1278
1279
1280 public HTable createTable(TableName tableName, byte[] family)
1281 throws IOException{
1282 return createTable(tableName, new byte[][]{family});
1283 }
1284
1285
1286
1287
1288
1289
1290
1291
1292
1293 public HTable createMultiRegionTable(TableName tableName, byte[] family, int numRegions)
1294 throws IOException {
1295 if (numRegions < 3) throw new IOException("Must create at least 3 regions");
1296 byte[] startKey = Bytes.toBytes("aaaaa");
1297 byte[] endKey = Bytes.toBytes("zzzzz");
1298 byte[][] splitKeys = Bytes.split(startKey, endKey, numRegions - 3);
1299
1300 return createTable(tableName, new byte[][] { family }, splitKeys);
1301 }
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311 public HTable createTable(byte[] tableName, byte[][] families)
1312 throws IOException {
1313 return createTable(tableName, families,
1314 new Configuration(getConfiguration()));
1315 }
1316
1317
1318
1319
1320
1321
1322
1323
1324 public HTable createTable(TableName tableName, byte[][] families)
1325 throws IOException {
1326 return createTable(tableName, families, (byte[][]) null);
1327 }
1328
1329
1330
1331
1332
1333
1334
1335
1336 public HTable createMultiRegionTable(TableName tableName, byte[][] families) throws IOException {
1337 return createTable(tableName, families, KEYS_FOR_HBA_CREATE_TABLE);
1338 }
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348 public HTable createTable(TableName tableName, byte[][] families, byte[][] splitKeys)
1349 throws IOException {
1350 return createTable(tableName, families, splitKeys, new Configuration(getConfiguration()));
1351 }
1352
1353 public HTable createTable(byte[] tableName, byte[][] families,
1354 int numVersions, byte[] startKey, byte[] endKey, int numRegions) throws IOException {
1355 return createTable(TableName.valueOf(tableName), families, numVersions,
1356 startKey, endKey, numRegions);
1357 }
1358
1359 public HTable createTable(String tableName, byte[][] families,
1360 int numVersions, byte[] startKey, byte[] endKey, int numRegions) throws IOException {
1361 return createTable(TableName.valueOf(tableName), families, numVersions,
1362 startKey, endKey, numRegions);
1363 }
1364
1365 public HTable createTable(TableName tableName, byte[][] families,
1366 int numVersions, byte[] startKey, byte[] endKey, int numRegions)
1367 throws IOException{
1368 HTableDescriptor desc = new HTableDescriptor(tableName);
1369 for (byte[] family : families) {
1370 HColumnDescriptor hcd = new HColumnDescriptor(family)
1371 .setMaxVersions(numVersions);
1372 desc.addFamily(hcd);
1373 }
1374 getHBaseAdmin().createTable(desc, startKey, endKey, numRegions);
1375
1376 waitUntilAllRegionsAssigned(tableName);
1377 return new HTable(getConfiguration(), tableName);
1378 }
1379
1380
1381
1382
1383
1384
1385
1386
1387
1388 public HTable createTable(HTableDescriptor htd, byte[][] families, Configuration c)
1389 throws IOException {
1390 return createTable(htd, families, (byte[][]) null, c);
1391 }
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402 public HTable createTable(HTableDescriptor htd, byte[][] families, byte[][] splitKeys,
1403 Configuration c) throws IOException {
1404 for (byte[] family : families) {
1405 HColumnDescriptor hcd = new HColumnDescriptor(family);
1406
1407
1408
1409 hcd.setBloomFilterType(BloomType.NONE);
1410 htd.addFamily(hcd);
1411 }
1412 getHBaseAdmin().createTable(htd, splitKeys);
1413
1414
1415 waitUntilAllRegionsAssigned(htd.getTableName());
1416 return (HTable) getConnection().getTable(htd.getTableName());
1417 }
1418
1419
1420
1421
1422
1423
1424
1425
1426 public HTable createTable(HTableDescriptor htd, byte[][] splitRows)
1427 throws IOException {
1428 getHBaseAdmin().createTable(htd, splitRows);
1429
1430 waitUntilAllRegionsAssigned(htd.getTableName());
1431 return new HTable(getConfiguration(), htd.getTableName());
1432 }
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442 public HTable createTable(TableName tableName, byte[][] families,
1443 final Configuration c)
1444 throws IOException {
1445 return createTable(tableName, families, (byte[][]) null, c);
1446 }
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457 public HTable createTable(TableName tableName, byte[][] families, byte[][] splitKeys,
1458 final Configuration c) throws IOException {
1459 return createTable(new HTableDescriptor(tableName), families, splitKeys, c);
1460 }
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470 public HTable createTable(byte[] tableName, byte[][] families,
1471 final Configuration c)
1472 throws IOException {
1473 HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
1474 for(byte[] family : families) {
1475 HColumnDescriptor hcd = new HColumnDescriptor(family);
1476
1477
1478
1479 hcd.setBloomFilterType(BloomType.NONE);
1480 desc.addFamily(hcd);
1481 }
1482 getHBaseAdmin().createTable(desc);
1483 return new HTable(c, desc.getTableName());
1484 }
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495 public HTable createTable(TableName tableName, byte[][] families,
1496 final Configuration c, int numVersions)
1497 throws IOException {
1498 HTableDescriptor desc = new HTableDescriptor(tableName);
1499 for(byte[] family : families) {
1500 HColumnDescriptor hcd = new HColumnDescriptor(family)
1501 .setMaxVersions(numVersions);
1502 desc.addFamily(hcd);
1503 }
1504 getHBaseAdmin().createTable(desc);
1505
1506 waitUntilAllRegionsAssigned(tableName);
1507 return new HTable(c, tableName);
1508 }
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519 public HTable createTable(byte[] tableName, byte[][] families,
1520 final Configuration c, int numVersions)
1521 throws IOException {
1522 HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
1523 for(byte[] family : families) {
1524 HColumnDescriptor hcd = new HColumnDescriptor(family)
1525 .setMaxVersions(numVersions);
1526 desc.addFamily(hcd);
1527 }
1528 getHBaseAdmin().createTable(desc);
1529 return new HTable(c, desc.getTableName());
1530 }
1531
1532
1533
1534
1535
1536
1537
1538
1539
1540 public HTable createTable(byte[] tableName, byte[] family, int numVersions)
1541 throws IOException {
1542 return createTable(tableName, new byte[][]{family}, numVersions);
1543 }
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553 public HTable createTable(TableName tableName, byte[] family, int numVersions)
1554 throws IOException {
1555 return createTable(tableName, new byte[][]{family}, numVersions);
1556 }
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566 public HTable createTable(byte[] tableName, byte[][] families,
1567 int numVersions)
1568 throws IOException {
1569 return createTable(TableName.valueOf(tableName), families, numVersions);
1570 }
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580 public HTable createTable(TableName tableName, byte[][] families,
1581 int numVersions)
1582 throws IOException {
1583 return createTable(tableName, families, numVersions, (byte[][]) null);
1584 }
1585
1586
1587
1588
1589
1590
1591
1592
1593
1594
1595 public HTable createTable(TableName tableName, byte[][] families, int numVersions,
1596 byte[][] splitKeys) throws IOException {
1597 HTableDescriptor desc = new HTableDescriptor(tableName);
1598 for (byte[] family : families) {
1599 HColumnDescriptor hcd = new HColumnDescriptor(family).setMaxVersions(numVersions);
1600 desc.addFamily(hcd);
1601 }
1602 getHBaseAdmin().createTable(desc, splitKeys);
1603
1604 waitUntilAllRegionsAssigned(tableName);
1605 return new HTable(new Configuration(getConfiguration()), tableName);
1606 }
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616 public HTable createMultiRegionTable(TableName tableName, byte[][] families, int numVersions)
1617 throws IOException {
1618 return createTable(tableName, families, numVersions, KEYS_FOR_HBA_CREATE_TABLE);
1619 }
1620
1621
1622
1623
1624
1625
1626
1627
1628
1629
1630 public HTable createTable(byte[] tableName, byte[][] families,
1631 int numVersions, int blockSize) throws IOException {
1632 return createTable(TableName.valueOf(tableName),
1633 families, numVersions, blockSize);
1634 }
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645 public HTable createTable(TableName tableName, byte[][] families,
1646 int numVersions, int blockSize) throws IOException {
1647 HTableDescriptor desc = new HTableDescriptor(tableName);
1648 for (byte[] family : families) {
1649 HColumnDescriptor hcd = new HColumnDescriptor(family)
1650 .setMaxVersions(numVersions)
1651 .setBlocksize(blockSize);
1652 desc.addFamily(hcd);
1653 }
1654 getHBaseAdmin().createTable(desc);
1655
1656 waitUntilAllRegionsAssigned(tableName);
1657 return new HTable(new Configuration(getConfiguration()), tableName);
1658 }
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668 public HTable createTable(byte[] tableName, byte[][] families,
1669 int[] numVersions)
1670 throws IOException {
1671 return createTable(TableName.valueOf(tableName), families, numVersions);
1672 }
1673
1674
1675
1676
1677
1678
1679
1680
1681
1682 public HTable createTable(TableName tableName, byte[][] families,
1683 int[] numVersions)
1684 throws IOException {
1685 HTableDescriptor desc = new HTableDescriptor(tableName);
1686 int i = 0;
1687 for (byte[] family : families) {
1688 HColumnDescriptor hcd = new HColumnDescriptor(family)
1689 .setMaxVersions(numVersions[i]);
1690 desc.addFamily(hcd);
1691 i++;
1692 }
1693 getHBaseAdmin().createTable(desc);
1694
1695 waitUntilAllRegionsAssigned(tableName);
1696 return new HTable(new Configuration(getConfiguration()), tableName);
1697 }
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707 public HTable createTable(byte[] tableName, byte[] family, byte[][] splitRows)
1708 throws IOException{
1709 return createTable(TableName.valueOf(tableName), family, splitRows);
1710 }
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720 public HTable createTable(TableName tableName, byte[] family, byte[][] splitRows)
1721 throws IOException {
1722 HTableDescriptor desc = new HTableDescriptor(tableName);
1723 HColumnDescriptor hcd = new HColumnDescriptor(family);
1724 desc.addFamily(hcd);
1725 getHBaseAdmin().createTable(desc, splitRows);
1726
1727 waitUntilAllRegionsAssigned(tableName);
1728 return new HTable(getConfiguration(), tableName);
1729 }
1730
1731
1732
1733
1734
1735
1736
1737
1738 public HTable createMultiRegionTable(TableName tableName, byte[] family) throws IOException {
1739 return createTable(tableName, family, KEYS_FOR_HBA_CREATE_TABLE);
1740 }
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750 public HTable createTable(byte[] tableName, byte[][] families, byte[][] splitRows)
1751 throws IOException {
1752 HTableDescriptor desc = new HTableDescriptor(TableName.valueOf(tableName));
1753 for(byte[] family:families) {
1754 HColumnDescriptor hcd = new HColumnDescriptor(family);
1755 desc.addFamily(hcd);
1756 }
1757 getHBaseAdmin().createTable(desc, splitRows);
1758
1759 waitUntilAllRegionsAssigned(desc.getTableName());
1760 return new HTable(getConfiguration(), desc.getTableName());
1761 }
1762
1763
1764
1765
1766 public static WAL createWal(final Configuration conf, final Path rootDir, final HRegionInfo hri)
1767 throws IOException {
1768
1769
1770 Configuration confForWAL = new Configuration(conf);
1771 confForWAL.set(HConstants.HBASE_DIR, rootDir.toString());
1772 return (new WALFactory(confForWAL,
1773 Collections.<WALActionsListener>singletonList(new MetricsWAL()),
1774 "hregion-" + RandomStringUtils.randomNumeric(8))).
1775 getWAL(hri.getEncodedNameAsBytes());
1776 }
1777
1778
1779
1780
1781
1782 public static HRegion createRegionAndWAL(final HRegionInfo info, final Path rootDir,
1783 final Configuration conf, final HTableDescriptor htd) throws IOException {
1784 return createRegionAndWAL(info, rootDir, conf, htd, true);
1785 }
1786
1787
1788
1789
1790
1791 public static HRegion createRegionAndWAL(final HRegionInfo info, final Path rootDir,
1792 final Configuration conf, final HTableDescriptor htd, boolean initialize)
1793 throws IOException {
1794 WAL wal = createWal(conf, rootDir, info);
1795 return HRegion.createHRegion(info, rootDir, conf, htd, wal, initialize);
1796 }
1797
1798
1799
1800
1801 public static void closeRegionAndWAL(final Region r) throws IOException {
1802 closeRegionAndWAL((HRegion)r);
1803 }
1804
1805
1806
1807
1808 public static void closeRegionAndWAL(final HRegion r) throws IOException {
1809 if (r == null) return;
1810 r.close();
1811 if (r.getWAL() == null) return;
1812 r.getWAL().close();
1813 }
1814
1815
1816
1817
1818 @SuppressWarnings("serial")
1819 public static void modifyTableSync(Admin admin, HTableDescriptor desc)
1820 throws IOException, InterruptedException {
1821 admin.modifyTable(desc.getTableName(), desc);
1822 Pair<Integer, Integer> status = new Pair<Integer, Integer>() {{
1823 setFirst(0);
1824 setSecond(0);
1825 }};
1826 int i = 0;
1827 do {
1828 status = admin.getAlterStatus(desc.getTableName());
1829 if (status.getSecond() != 0) {
1830 LOG.debug(status.getSecond() - status.getFirst() + "/" + status.getSecond()
1831 + " regions updated.");
1832 Thread.sleep(1 * 1000l);
1833 } else {
1834 LOG.debug("All regions updated.");
1835 break;
1836 }
1837 } while (status.getFirst() != 0 && i++ < 500);
1838 if (status.getFirst() != 0) {
1839 throw new IOException("Failed to update all regions even after 500 seconds.");
1840 }
1841 }
1842
1843
1844
1845
1846 public static void setReplicas(Admin admin, TableName table, int replicaCount)
1847 throws IOException, InterruptedException {
1848 admin.disableTable(table);
1849 HTableDescriptor desc = admin.getTableDescriptor(table);
1850 desc.setRegionReplication(replicaCount);
1851 admin.modifyTable(desc.getTableName(), desc);
1852 admin.enableTable(table);
1853 }
1854
1855
1856
1857
1858
1859 public void deleteTable(String tableName) throws IOException {
1860 deleteTable(TableName.valueOf(tableName));
1861 }
1862
1863
1864
1865
1866
1867 public void deleteTable(byte[] tableName) throws IOException {
1868 deleteTable(TableName.valueOf(tableName));
1869 }
1870
1871
1872
1873
1874
1875 public void deleteTable(TableName tableName) throws IOException {
1876 try {
1877 getHBaseAdmin().disableTable(tableName);
1878 } catch (TableNotEnabledException e) {
1879 LOG.debug("Table: " + tableName + " already disabled, so just deleting it.");
1880 }
1881 getHBaseAdmin().deleteTable(tableName);
1882 }
1883
1884
1885
1886
1887
1888 public void deleteTableIfAny(TableName tableName) throws IOException {
1889 try {
1890 deleteTable(tableName);
1891 } catch (TableNotFoundException e) {
1892
1893 }
1894 }
1895
1896
1897
1898
1899
1900 public final static byte [] fam1 = Bytes.toBytes("colfamily11");
1901 public final static byte [] fam2 = Bytes.toBytes("colfamily21");
1902 public final static byte [] fam3 = Bytes.toBytes("colfamily31");
1903 public static final byte[][] COLUMNS = {fam1, fam2, fam3};
1904 private static final int MAXVERSIONS = 3;
1905
1906 public static final char FIRST_CHAR = 'a';
1907 public static final char LAST_CHAR = 'z';
1908 public static final byte [] START_KEY_BYTES = {FIRST_CHAR, FIRST_CHAR, FIRST_CHAR};
1909 public static final String START_KEY = new String(START_KEY_BYTES, HConstants.UTF8_CHARSET);
1910
1911
1912
1913
1914
1915
1916
1917
1918 public HTableDescriptor createTableDescriptor(final String name,
1919 final int minVersions, final int versions, final int ttl, KeepDeletedCells keepDeleted) {
1920 HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(name));
1921 for (byte[] cfName : new byte[][]{ fam1, fam2, fam3 }) {
1922 htd.addFamily(new HColumnDescriptor(cfName)
1923 .setMinVersions(minVersions)
1924 .setMaxVersions(versions)
1925 .setKeepDeletedCells(keepDeleted)
1926 .setBlockCacheEnabled(false)
1927 .setTimeToLive(ttl)
1928 );
1929 }
1930 return htd;
1931 }
1932
1933
1934
1935
1936
1937
1938
1939 public HTableDescriptor createTableDescriptor(final String name) {
1940 return createTableDescriptor(name, HColumnDescriptor.DEFAULT_MIN_VERSIONS,
1941 MAXVERSIONS, HConstants.FOREVER, HColumnDescriptor.DEFAULT_KEEP_DELETED);
1942 }
1943
1944
1945
1946
1947
1948 public HRegion createHRegion(
1949 final HRegionInfo info,
1950 final Path rootDir,
1951 final Configuration conf,
1952 final HTableDescriptor htd) throws IOException {
1953 return HRegion.createHRegion(info, rootDir, conf, htd);
1954 }
1955
1956
1957
1958
1959
1960
1961
1962
1963
1964 public HRegion createLocalHRegion(HTableDescriptor desc, byte [] startKey,
1965 byte [] endKey)
1966 throws IOException {
1967 HRegionInfo hri = new HRegionInfo(desc.getTableName(), startKey, endKey);
1968 return createLocalHRegion(hri, desc);
1969 }
1970
1971
1972
1973
1974
1975
1976
1977
1978 public HRegion createLocalHRegion(HRegionInfo info, HTableDescriptor desc) throws IOException {
1979 return HRegion.createHRegion(info, getDataTestDir(), getConfiguration(), desc);
1980 }
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990 public HRegion createLocalHRegion(HRegionInfo info, HTableDescriptor desc, WAL wal)
1991 throws IOException {
1992 return HRegion.createHRegion(info, getDataTestDir(), getConfiguration(), desc, wal);
1993 }
1994
1995
1996
1997
1998
1999
2000
2001
2002
2003
2004
2005
2006
2007 public HRegion createLocalHRegion(byte[] tableName, byte[] startKey, byte[] stopKey,
2008 String callingMethod, Configuration conf, boolean isReadOnly, Durability durability,
2009 WAL wal, byte[]... families) throws IOException {
2010 HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(tableName));
2011 htd.setReadOnly(isReadOnly);
2012 for (byte[] family : families) {
2013 HColumnDescriptor hcd = new HColumnDescriptor(family);
2014
2015 hcd.setMaxVersions(Integer.MAX_VALUE);
2016 htd.addFamily(hcd);
2017 }
2018 htd.setDurability(durability);
2019 HRegionInfo info = new HRegionInfo(htd.getTableName(), startKey, stopKey, false);
2020 return createLocalHRegion(info, htd, wal);
2021 }
2022
2023
2024
2025
2026
2027
2028
2029
2030
2031
2032 public HTable deleteTableData(byte[] tableName) throws IOException {
2033 return deleteTableData(TableName.valueOf(tableName));
2034 }
2035
2036
2037
2038
2039
2040
2041
2042
2043 public HTable deleteTableData(TableName tableName) throws IOException {
2044 HTable table = new HTable(getConfiguration(), tableName);
2045 Scan scan = new Scan();
2046 ResultScanner resScan = table.getScanner(scan);
2047 for(Result res : resScan) {
2048 Delete del = new Delete(res.getRow());
2049 table.delete(del);
2050 }
2051 resScan = table.getScanner(scan);
2052 resScan.close();
2053 return table;
2054 }
2055
2056
2057
2058
2059
2060
2061
2062
2063 public HTable truncateTable(final TableName tableName, final boolean preserveRegions)
2064 throws IOException {
2065 Admin admin = getHBaseAdmin();
2066 admin.truncateTable(tableName, preserveRegions);
2067 return new HTable(getConfiguration(), tableName);
2068 }
2069
2070
2071
2072
2073
2074
2075
2076
2077
2078
2079 public HTable truncateTable(final TableName tableName) throws IOException {
2080 return truncateTable(tableName, false);
2081 }
2082
2083
2084
2085
2086
2087
2088
2089
2090 public HTable truncateTable(final byte[] tableName, final boolean preserveRegions)
2091 throws IOException {
2092 return truncateTable(TableName.valueOf(tableName), preserveRegions);
2093 }
2094
2095
2096
2097
2098
2099
2100
2101
2102
2103
2104 public HTable truncateTable(final byte[] tableName) throws IOException {
2105 return truncateTable(tableName, false);
2106 }
2107
2108
2109
2110
2111
2112
2113
2114
2115 public int loadTable(final Table t, final byte[] f) throws IOException {
2116 return loadTable(t, new byte[][] {f});
2117 }
2118
2119
2120
2121
2122
2123
2124
2125
2126 public int loadTable(final Table t, final byte[] f, boolean writeToWAL) throws IOException {
2127 return loadTable(t, new byte[][] {f}, null, writeToWAL);
2128 }
2129
2130
2131
2132
2133
2134
2135
2136
2137 public int loadTable(final Table t, final byte[][] f) throws IOException {
2138 return loadTable(t, f, null);
2139 }
2140
2141
2142
2143
2144
2145
2146
2147
2148
2149 public int loadTable(final Table t, final byte[][] f, byte[] value) throws IOException {
2150 return loadTable(t, f, value, true);
2151 }
2152
2153
2154
2155
2156
2157
2158
2159
2160
2161 public int loadTable(final Table t, final byte[][] f, byte[] value, boolean writeToWAL) throws IOException {
2162 List<Put> puts = new ArrayList<>();
2163 for (byte[] row : HBaseTestingUtility.ROWS) {
2164 Put put = new Put(row);
2165 put.setDurability(writeToWAL ? Durability.USE_DEFAULT : Durability.SKIP_WAL);
2166 for (int i = 0; i < f.length; i++) {
2167 put.add(f[i], null, value != null ? value : row);
2168 }
2169 puts.add(put);
2170 }
2171 t.put(puts);
2172 return puts.size();
2173 }
2174
2175
2176
2177
2178 public static class SeenRowTracker {
2179 int dim = 'z' - 'a' + 1;
2180 int[][][] seenRows = new int[dim][dim][dim];
2181 byte[] startRow;
2182 byte[] stopRow;
2183
2184 public SeenRowTracker(byte[] startRow, byte[] stopRow) {
2185 this.startRow = startRow;
2186 this.stopRow = stopRow;
2187 }
2188
2189 void reset() {
2190 for (byte[] row : ROWS) {
2191 seenRows[i(row[0])][i(row[1])][i(row[2])] = 0;
2192 }
2193 }
2194
2195 int i(byte b) {
2196 return b - 'a';
2197 }
2198
2199 public void addRow(byte[] row) {
2200 seenRows[i(row[0])][i(row[1])][i(row[2])]++;
2201 }
2202
2203
2204
2205
2206 public void validate() {
2207 for (byte b1 = 'a'; b1 <= 'z'; b1++) {
2208 for (byte b2 = 'a'; b2 <= 'z'; b2++) {
2209 for (byte b3 = 'a'; b3 <= 'z'; b3++) {
2210 int count = seenRows[i(b1)][i(b2)][i(b3)];
2211 int expectedCount = 0;
2212 if (Bytes.compareTo(new byte[] {b1,b2,b3}, startRow) >= 0
2213 && Bytes.compareTo(new byte[] {b1,b2,b3}, stopRow) < 0) {
2214 expectedCount = 1;
2215 }
2216 if (count != expectedCount) {
2217 String row = new String(new byte[] {b1,b2,b3});
2218 throw new RuntimeException("Row:" + row + " has a seen count of " + count + " instead of " + expectedCount);
2219 }
2220 }
2221 }
2222 }
2223 }
2224 }
2225
2226 public int loadRegion(final HRegion r, final byte[] f) throws IOException {
2227 return loadRegion(r, f, false);
2228 }
2229
2230 public int loadRegion(final Region r, final byte[] f) throws IOException {
2231 return loadRegion((HRegion)r, f);
2232 }
2233
2234
2235
2236
2237
2238
2239
2240
2241
2242 public int loadRegion(final HRegion r, final byte[] f, final boolean flush)
2243 throws IOException {
2244 byte[] k = new byte[3];
2245 int rowCount = 0;
2246 for (byte b1 = 'a'; b1 <= 'z'; b1++) {
2247 for (byte b2 = 'a'; b2 <= 'z'; b2++) {
2248 for (byte b3 = 'a'; b3 <= 'z'; b3++) {
2249 k[0] = b1;
2250 k[1] = b2;
2251 k[2] = b3;
2252 Put put = new Put(k);
2253 put.setDurability(Durability.SKIP_WAL);
2254 put.add(f, null, k);
2255 if (r.getWAL() == null) {
2256 put.setDurability(Durability.SKIP_WAL);
2257 }
2258 int preRowCount = rowCount;
2259 int pause = 10;
2260 int maxPause = 1000;
2261 while (rowCount == preRowCount) {
2262 try {
2263 r.put(put);
2264 rowCount++;
2265 } catch (RegionTooBusyException e) {
2266 pause = (pause * 2 >= maxPause) ? maxPause : pause * 2;
2267 Threads.sleep(pause);
2268 }
2269 }
2270 }
2271 }
2272 if (flush) {
2273 r.flush(true);
2274 }
2275 }
2276 return rowCount;
2277 }
2278
2279 public void loadNumericRows(final Table t, final byte[] f, int startRow, int endRow)
2280 throws IOException {
2281 for (int i = startRow; i < endRow; i++) {
2282 byte[] data = Bytes.toBytes(String.valueOf(i));
2283 Put put = new Put(data);
2284 put.add(f, null, data);
2285 t.put(put);
2286 }
2287 }
2288
2289 public void verifyNumericRows(Table table, final byte[] f, int startRow, int endRow,
2290 int replicaId)
2291 throws IOException {
2292 for (int i = startRow; i < endRow; i++) {
2293 String failMsg = "Failed verification of row :" + i;
2294 byte[] data = Bytes.toBytes(String.valueOf(i));
2295 Get get = new Get(data);
2296 get.setReplicaId(replicaId);
2297 get.setConsistency(Consistency.TIMELINE);
2298 Result result = table.get(get);
2299 assertTrue(failMsg, result.containsColumn(f, null));
2300 assertEquals(failMsg, result.getColumnCells(f, null).size(), 1);
2301 Cell cell = result.getColumnLatestCell(f, null);
2302 assertTrue(failMsg,
2303 Bytes.equals(data, 0, data.length, cell.getValueArray(), cell.getValueOffset(),
2304 cell.getValueLength()));
2305 }
2306 }
2307
2308 public void verifyNumericRows(Region region, final byte[] f, int startRow, int endRow)
2309 throws IOException {
2310 verifyNumericRows((HRegion)region, f, startRow, endRow);
2311 }
2312
2313 public void verifyNumericRows(HRegion region, final byte[] f, int startRow, int endRow)
2314 throws IOException {
2315 verifyNumericRows(region, f, startRow, endRow, true);
2316 }
2317
2318 public void verifyNumericRows(Region region, final byte[] f, int startRow, int endRow,
2319 final boolean present) throws IOException {
2320 verifyNumericRows((HRegion)region, f, startRow, endRow, present);
2321 }
2322
2323 public void verifyNumericRows(HRegion region, final byte[] f, int startRow, int endRow,
2324 final boolean present) throws IOException {
2325 for (int i = startRow; i < endRow; i++) {
2326 String failMsg = "Failed verification of row :" + i;
2327 byte[] data = Bytes.toBytes(String.valueOf(i));
2328 Result result = region.get(new Get(data));
2329
2330 boolean hasResult = result != null && !result.isEmpty();
2331 assertEquals(failMsg + result, present, hasResult);
2332 if (!present) continue;
2333
2334 assertTrue(failMsg, result.containsColumn(f, null));
2335 assertEquals(failMsg, result.getColumnCells(f, null).size(), 1);
2336 Cell cell = result.getColumnLatestCell(f, null);
2337 assertTrue(failMsg,
2338 Bytes.equals(data, 0, data.length, cell.getValueArray(), cell.getValueOffset(),
2339 cell.getValueLength()));
2340 }
2341 }
2342
2343 public void deleteNumericRows(final HTable t, final byte[] f, int startRow, int endRow)
2344 throws IOException {
2345 for (int i = startRow; i < endRow; i++) {
2346 byte[] data = Bytes.toBytes(String.valueOf(i));
2347 Delete delete = new Delete(data);
2348 delete.deleteFamily(f);
2349 t.delete(delete);
2350 }
2351 }
2352
2353
2354
2355
2356 public int countRows(final Table table) throws IOException {
2357 Scan scan = new Scan();
2358 ResultScanner results = table.getScanner(scan);
2359 int count = 0;
2360 for (@SuppressWarnings("unused") Result res : results) {
2361 count++;
2362 }
2363 results.close();
2364 return count;
2365 }
2366
2367 public int countRows(final Table table, final byte[]... families) throws IOException {
2368 Scan scan = new Scan();
2369 for (byte[] family: families) {
2370 scan.addFamily(family);
2371 }
2372 ResultScanner results = table.getScanner(scan);
2373 int count = 0;
2374 for (@SuppressWarnings("unused") Result res : results) {
2375 count++;
2376 }
2377 results.close();
2378 return count;
2379 }
2380
2381
2382
2383
2384 public int countRows(final TableName tableName) throws IOException {
2385 Table table = getConnection().getTable(tableName);
2386 try {
2387 return countRows(table);
2388 } finally {
2389 table.close();
2390 }
2391 }
2392
2393
2394
2395
2396 public String checksumRows(final Table table) throws Exception {
2397 Scan scan = new Scan();
2398 ResultScanner results = table.getScanner(scan);
2399 MessageDigest digest = MessageDigest.getInstance("MD5");
2400 for (Result res : results) {
2401 digest.update(res.getRow());
2402 }
2403 results.close();
2404 return digest.toString();
2405 }
2406
2407
2408 public static final byte[][] ROWS = new byte[(int) Math.pow('z' - 'a' + 1, 3)][3];
2409 static {
2410 int i = 0;
2411 for (byte b1 = 'a'; b1 <= 'z'; b1++) {
2412 for (byte b2 = 'a'; b2 <= 'z'; b2++) {
2413 for (byte b3 = 'a'; b3 <= 'z'; b3++) {
2414 ROWS[i][0] = b1;
2415 ROWS[i][1] = b2;
2416 ROWS[i][2] = b3;
2417 i++;
2418 }
2419 }
2420 }
2421 }
2422
2423 public static final byte[][] KEYS = {
2424 HConstants.EMPTY_BYTE_ARRAY, Bytes.toBytes("bbb"),
2425 Bytes.toBytes("ccc"), Bytes.toBytes("ddd"), Bytes.toBytes("eee"),
2426 Bytes.toBytes("fff"), Bytes.toBytes("ggg"), Bytes.toBytes("hhh"),
2427 Bytes.toBytes("iii"), Bytes.toBytes("jjj"), Bytes.toBytes("kkk"),
2428 Bytes.toBytes("lll"), Bytes.toBytes("mmm"), Bytes.toBytes("nnn"),
2429 Bytes.toBytes("ooo"), Bytes.toBytes("ppp"), Bytes.toBytes("qqq"),
2430 Bytes.toBytes("rrr"), Bytes.toBytes("sss"), Bytes.toBytes("ttt"),
2431 Bytes.toBytes("uuu"), Bytes.toBytes("vvv"), Bytes.toBytes("www"),
2432 Bytes.toBytes("xxx"), Bytes.toBytes("yyy")
2433 };
2434
2435 public static final byte[][] KEYS_FOR_HBA_CREATE_TABLE = {
2436 Bytes.toBytes("bbb"),
2437 Bytes.toBytes("ccc"), Bytes.toBytes("ddd"), Bytes.toBytes("eee"),
2438 Bytes.toBytes("fff"), Bytes.toBytes("ggg"), Bytes.toBytes("hhh"),
2439 Bytes.toBytes("iii"), Bytes.toBytes("jjj"), Bytes.toBytes("kkk"),
2440 Bytes.toBytes("lll"), Bytes.toBytes("mmm"), Bytes.toBytes("nnn"),
2441 Bytes.toBytes("ooo"), Bytes.toBytes("ppp"), Bytes.toBytes("qqq"),
2442 Bytes.toBytes("rrr"), Bytes.toBytes("sss"), Bytes.toBytes("ttt"),
2443 Bytes.toBytes("uuu"), Bytes.toBytes("vvv"), Bytes.toBytes("www"),
2444 Bytes.toBytes("xxx"), Bytes.toBytes("yyy"), Bytes.toBytes("zzz")
2445 };
2446
2447
2448
2449
2450
2451
2452
2453
2454
2455
2456
2457 public List<HRegionInfo> createMultiRegionsInMeta(final Configuration conf,
2458 final HTableDescriptor htd, byte [][] startKeys)
2459 throws IOException {
2460 Table meta = new HTable(conf, TableName.META_TABLE_NAME);
2461 Arrays.sort(startKeys, Bytes.BYTES_COMPARATOR);
2462 List<HRegionInfo> newRegions = new ArrayList<HRegionInfo>(startKeys.length);
2463
2464 for (int i = 0; i < startKeys.length; i++) {
2465 int j = (i + 1) % startKeys.length;
2466 HRegionInfo hri = new HRegionInfo(htd.getTableName(), startKeys[i],
2467 startKeys[j]);
2468 MetaTableAccessor.addRegionToMeta(meta, hri);
2469 newRegions.add(hri);
2470 }
2471
2472 meta.close();
2473 return newRegions;
2474 }
2475
2476
2477
2478
2479
2480
2481 public List<byte[]> getMetaTableRows() throws IOException {
2482
2483 Table t = new HTable(new Configuration(this.conf), TableName.META_TABLE_NAME);
2484 List<byte[]> rows = new ArrayList<byte[]>();
2485 ResultScanner s = t.getScanner(new Scan());
2486 for (Result result : s) {
2487 LOG.info("getMetaTableRows: row -> " +
2488 Bytes.toStringBinary(result.getRow()));
2489 rows.add(result.getRow());
2490 }
2491 s.close();
2492 t.close();
2493 return rows;
2494 }
2495
2496
2497
2498
2499
2500
2501 public List<byte[]> getMetaTableRows(TableName tableName) throws IOException {
2502
2503 Table t = new HTable(new Configuration(this.conf), TableName.META_TABLE_NAME);
2504 List<byte[]> rows = new ArrayList<byte[]>();
2505 ResultScanner s = t.getScanner(new Scan());
2506 for (Result result : s) {
2507 HRegionInfo info = HRegionInfo.getHRegionInfo(result);
2508 if (info == null) {
2509 LOG.error("No region info for row " + Bytes.toString(result.getRow()));
2510
2511 continue;
2512 }
2513
2514 if (info.getTable().equals(tableName)) {
2515 LOG.info("getMetaTableRows: row -> " +
2516 Bytes.toStringBinary(result.getRow()) + info);
2517 rows.add(result.getRow());
2518 }
2519 }
2520 s.close();
2521 t.close();
2522 return rows;
2523 }
2524
2525
2526
2527
2528
2529
2530
2531
2532
2533
2534
2535
2536 public HRegionServer getRSForFirstRegionInTable(TableName tableName)
2537 throws IOException, InterruptedException {
2538 List<byte[]> metaRows = getMetaTableRows(tableName);
2539 if (metaRows == null || metaRows.isEmpty()) {
2540 return null;
2541 }
2542 LOG.debug("Found " + metaRows.size() + " rows for table " +
2543 tableName);
2544 byte [] firstrow = metaRows.get(0);
2545 LOG.debug("FirstRow=" + Bytes.toString(firstrow));
2546 long pause = getConfiguration().getLong(HConstants.HBASE_CLIENT_PAUSE,
2547 HConstants.DEFAULT_HBASE_CLIENT_PAUSE);
2548 int numRetries = getConfiguration().getInt(HConstants.HBASE_CLIENT_RETRIES_NUMBER,
2549 HConstants.DEFAULT_HBASE_CLIENT_RETRIES_NUMBER);
2550 RetryCounter retrier = new RetryCounter(numRetries+1, (int)pause, TimeUnit.MICROSECONDS);
2551 while(retrier.shouldRetry()) {
2552 int index = getMiniHBaseCluster().getServerWith(firstrow);
2553 if (index != -1) {
2554 return getMiniHBaseCluster().getRegionServerThreads().get(index).getRegionServer();
2555 }
2556
2557 retrier.sleepUntilNextRetry();
2558 }
2559 return null;
2560 }
2561
2562
2563
2564
2565
2566
2567
2568 public MiniMRCluster startMiniMapReduceCluster() throws IOException {
2569 startMiniMapReduceCluster(2);
2570 return mrCluster;
2571 }
2572
2573
2574
2575
2576
2577 private void forceChangeTaskLogDir() {
2578 Field logDirField;
2579 try {
2580 logDirField = TaskLog.class.getDeclaredField("LOG_DIR");
2581 logDirField.setAccessible(true);
2582
2583 Field modifiersField = Field.class.getDeclaredField("modifiers");
2584 modifiersField.setAccessible(true);
2585 modifiersField.setInt(logDirField, logDirField.getModifiers() & ~Modifier.FINAL);
2586
2587 logDirField.set(null, new File(hadoopLogDir, "userlogs"));
2588 } catch (SecurityException e) {
2589 throw new RuntimeException(e);
2590 } catch (NoSuchFieldException e) {
2591
2592 throw new RuntimeException(e);
2593 } catch (IllegalArgumentException e) {
2594 throw new RuntimeException(e);
2595 } catch (IllegalAccessException e) {
2596 throw new RuntimeException(e);
2597 }
2598 }
2599
2600
2601
2602
2603
2604
2605
2606 private void startMiniMapReduceCluster(final int servers) throws IOException {
2607 if (mrCluster != null) {
2608 throw new IllegalStateException("MiniMRCluster is already running");
2609 }
2610 LOG.info("Starting mini mapreduce cluster...");
2611 setupClusterTestDir();
2612 createDirsAndSetProperties();
2613
2614 forceChangeTaskLogDir();
2615
2616
2617
2618
2619 conf.setFloat("yarn.nodemanager.vmem-pmem-ratio", 8.0f);
2620
2621
2622
2623 conf.setBoolean("mapreduce.map.speculative", false);
2624 conf.setBoolean("mapreduce.reduce.speculative", false);
2625
2626
2627
2628 mrCluster = new MiniMRCluster(servers,
2629 FS_URI != null ? FS_URI : FileSystem.get(conf).getUri().toString(), 1,
2630 null, null, new JobConf(this.conf));
2631 JobConf jobConf = MapreduceTestingShim.getJobConf(mrCluster);
2632 if (jobConf == null) {
2633 jobConf = mrCluster.createJobConf();
2634 }
2635
2636 jobConf.set("mapreduce.cluster.local.dir",
2637 conf.get("mapreduce.cluster.local.dir"));
2638 LOG.info("Mini mapreduce cluster started");
2639
2640
2641
2642
2643 conf.set("mapreduce.jobtracker.address", jobConf.get("mapreduce.jobtracker.address"));
2644
2645 conf.set("mapreduce.framework.name", "yarn");
2646 conf.setBoolean("yarn.is.minicluster", true);
2647 String rmAddress = jobConf.get("yarn.resourcemanager.address");
2648 if (rmAddress != null) {
2649 conf.set("yarn.resourcemanager.address", rmAddress);
2650 }
2651 String historyAddress = jobConf.get("mapreduce.jobhistory.address");
2652 if (historyAddress != null) {
2653 conf.set("mapreduce.jobhistory.address", historyAddress);
2654 }
2655 String schedulerAddress =
2656 jobConf.get("yarn.resourcemanager.scheduler.address");
2657 if (schedulerAddress != null) {
2658 conf.set("yarn.resourcemanager.scheduler.address", schedulerAddress);
2659 }
2660 }
2661
2662
2663
2664
2665 public void shutdownMiniMapReduceCluster() {
2666 if (mrCluster != null) {
2667 LOG.info("Stopping mini mapreduce cluster...");
2668 mrCluster.shutdown();
2669 mrCluster = null;
2670 LOG.info("Mini mapreduce cluster stopped");
2671 }
2672
2673 conf.set("mapreduce.jobtracker.address", "local");
2674 }
2675
2676
2677
2678
2679 public RegionServerServices createMockRegionServerService() throws IOException {
2680 return createMockRegionServerService((ServerName)null);
2681 }
2682
2683
2684
2685
2686
2687 public RegionServerServices createMockRegionServerService(RpcServerInterface rpc) throws IOException {
2688 final MockRegionServerServices rss = new MockRegionServerServices(getZooKeeperWatcher());
2689 rss.setFileSystem(getTestFileSystem());
2690 rss.setRpcServer(rpc);
2691 return rss;
2692 }
2693
2694
2695
2696
2697
2698 public RegionServerServices createMockRegionServerService(ServerName name) throws IOException {
2699 final MockRegionServerServices rss = new MockRegionServerServices(getZooKeeperWatcher(), name);
2700 rss.setFileSystem(getTestFileSystem());
2701 return rss;
2702 }
2703
2704
2705
2706
2707
2708
2709 public void enableDebug(Class<?> clazz) {
2710 Log l = LogFactory.getLog(clazz);
2711 if (l instanceof Log4JLogger) {
2712 ((Log4JLogger) l).getLogger().setLevel(org.apache.log4j.Level.DEBUG);
2713 } else if (l instanceof Jdk14Logger) {
2714 ((Jdk14Logger) l).getLogger().setLevel(java.util.logging.Level.ALL);
2715 }
2716 }
2717
2718
2719
2720
2721
2722 public void expireMasterSession() throws Exception {
2723 HMaster master = getMiniHBaseCluster().getMaster();
2724 expireSession(master.getZooKeeper(), false);
2725 }
2726
2727
2728
2729
2730
2731
2732 public void expireRegionServerSession(int index) throws Exception {
2733 HRegionServer rs = getMiniHBaseCluster().getRegionServer(index);
2734 expireSession(rs.getZooKeeper(), false);
2735 decrementMinRegionServerCount();
2736 }
2737
2738 private void decrementMinRegionServerCount() {
2739
2740
2741 decrementMinRegionServerCount(getConfiguration());
2742
2743
2744 for (MasterThread master : getHBaseCluster().getMasterThreads()) {
2745 decrementMinRegionServerCount(master.getMaster().getConfiguration());
2746 }
2747 }
2748
2749 private void decrementMinRegionServerCount(Configuration conf) {
2750 int currentCount = conf.getInt(
2751 ServerManager.WAIT_ON_REGIONSERVERS_MINTOSTART, -1);
2752 if (currentCount != -1) {
2753 conf.setInt(ServerManager.WAIT_ON_REGIONSERVERS_MINTOSTART,
2754 Math.max(currentCount - 1, 1));
2755 }
2756 }
2757
2758 public void expireSession(ZooKeeperWatcher nodeZK) throws Exception {
2759 expireSession(nodeZK, false);
2760 }
2761
2762 @Deprecated
2763 public void expireSession(ZooKeeperWatcher nodeZK, Server server)
2764 throws Exception {
2765 expireSession(nodeZK, false);
2766 }
2767
2768
2769
2770
2771
2772
2773
2774
2775
2776
2777
2778
2779 public void expireSession(ZooKeeperWatcher nodeZK, boolean checkStatus)
2780 throws Exception {
2781 Configuration c = new Configuration(this.conf);
2782 String quorumServers = ZKConfig.getZKQuorumServersString(c);
2783 ZooKeeper zk = nodeZK.getRecoverableZooKeeper().getZooKeeper();
2784 byte[] password = zk.getSessionPasswd();
2785 long sessionID = zk.getSessionId();
2786
2787
2788
2789
2790
2791
2792
2793
2794 ZooKeeper monitor = new ZooKeeper(quorumServers,
2795 1000, new org.apache.zookeeper.Watcher(){
2796 @Override
2797 public void process(WatchedEvent watchedEvent) {
2798 LOG.info("Monitor ZKW received event="+watchedEvent);
2799 }
2800 } , sessionID, password);
2801
2802
2803 ZooKeeper newZK = new ZooKeeper(quorumServers,
2804 1000, EmptyWatcher.instance, sessionID, password);
2805
2806
2807
2808 long start = System.currentTimeMillis();
2809 while (newZK.getState() != States.CONNECTED
2810 && System.currentTimeMillis() - start < 1000) {
2811 Thread.sleep(1);
2812 }
2813 newZK.close();
2814 LOG.info("ZK Closed Session 0x" + Long.toHexString(sessionID));
2815
2816
2817 monitor.close();
2818
2819 if (checkStatus) {
2820 new HTable(new Configuration(conf), TableName.META_TABLE_NAME).close();
2821 }
2822 }
2823
2824
2825
2826
2827
2828
2829
2830 public MiniHBaseCluster getHBaseCluster() {
2831 return getMiniHBaseCluster();
2832 }
2833
2834
2835
2836
2837
2838
2839
2840
2841
2842 public HBaseCluster getHBaseClusterInterface() {
2843
2844
2845 return hbaseCluster;
2846 }
2847
2848
2849
2850
2851
2852
2853
2854 public Connection getConnection() throws IOException {
2855 if (this.connection == null) {
2856 this.connection = ConnectionFactory.createConnection(this.conf);
2857 }
2858 return this.connection;
2859 }
2860
2861
2862
2863
2864
2865
2866
2867
2868
2869
2870 public synchronized HBaseAdmin getHBaseAdmin()
2871 throws IOException {
2872 if (hbaseAdmin == null){
2873 this.hbaseAdmin = new HBaseAdminForTests(getConnection());
2874 }
2875 return hbaseAdmin;
2876 }
2877
2878 private HBaseAdminForTests hbaseAdmin = null;
2879 private static class HBaseAdminForTests extends HBaseAdmin {
2880 public HBaseAdminForTests(Connection connection) throws MasterNotRunningException,
2881 ZooKeeperConnectionException, IOException {
2882 super(connection);
2883 }
2884
2885 @Override
2886 public synchronized void close() throws IOException {
2887 LOG.warn("close() called on HBaseAdmin instance returned from " +
2888 "HBaseTestingUtility.getHBaseAdmin()");
2889 }
2890
2891 private synchronized void close0() throws IOException {
2892 super.close();
2893 }
2894 }
2895
2896
2897
2898
2899
2900
2901
2902
2903
2904
2905 public synchronized ZooKeeperWatcher getZooKeeperWatcher()
2906 throws IOException {
2907 if (zooKeeperWatcher == null) {
2908 zooKeeperWatcher = new ZooKeeperWatcher(conf, "testing utility",
2909 new Abortable() {
2910 @Override public void abort(String why, Throwable e) {
2911 throw new RuntimeException("Unexpected abort in HBaseTestingUtility:"+why, e);
2912 }
2913 @Override public boolean isAborted() {return false;}
2914 });
2915 }
2916 return zooKeeperWatcher;
2917 }
2918 private ZooKeeperWatcher zooKeeperWatcher;
2919
2920
2921
2922
2923
2924
2925
2926
2927
2928 public void closeRegion(String regionName) throws IOException {
2929 closeRegion(Bytes.toBytes(regionName));
2930 }
2931
2932
2933
2934
2935
2936
2937
2938 public void closeRegion(byte[] regionName) throws IOException {
2939 getHBaseAdmin().closeRegion(regionName, null);
2940 }
2941
2942
2943
2944
2945
2946
2947
2948
2949 public void closeRegionByRow(String row, RegionLocator table) throws IOException {
2950 closeRegionByRow(Bytes.toBytes(row), table);
2951 }
2952
2953
2954
2955
2956
2957
2958
2959
2960 public void closeRegionByRow(byte[] row, RegionLocator table) throws IOException {
2961 HRegionLocation hrl = table.getRegionLocation(row);
2962 closeRegion(hrl.getRegionInfo().getRegionName());
2963 }
2964
2965
2966
2967
2968
2969
2970
2971
2972 public HRegion getSplittableRegion(TableName tableName, int maxAttempts) {
2973 List<HRegion> regions = getHBaseCluster().getRegions(tableName);
2974 int regCount = regions.size();
2975 Set<Integer> attempted = new HashSet<Integer>();
2976 int idx;
2977 int attempts = 0;
2978 do {
2979 regions = getHBaseCluster().getRegions(tableName);
2980 if (regCount != regions.size()) {
2981
2982 attempted.clear();
2983 }
2984 regCount = regions.size();
2985
2986
2987 if (regCount > 0) {
2988 idx = random.nextInt(regCount);
2989
2990 if (attempted.contains(idx))
2991 continue;
2992 try {
2993 regions.get(idx).checkSplit();
2994 return regions.get(idx);
2995 } catch (Exception ex) {
2996 LOG.warn("Caught exception", ex);
2997 attempted.add(idx);
2998 }
2999 }
3000 attempts++;
3001 } while (maxAttempts == -1 || attempts < maxAttempts);
3002 return null;
3003 }
3004
3005 public MiniZooKeeperCluster getZkCluster() {
3006 return zkCluster;
3007 }
3008
3009 public void setZkCluster(MiniZooKeeperCluster zkCluster) {
3010 this.passedZkCluster = true;
3011 this.zkCluster = zkCluster;
3012 conf.setInt(HConstants.ZOOKEEPER_CLIENT_PORT, zkCluster.getClientPort());
3013 }
3014
3015 public MiniDFSCluster getDFSCluster() {
3016 return dfsCluster;
3017 }
3018
3019 public void setDFSCluster(MiniDFSCluster cluster) throws IllegalStateException, IOException {
3020 setDFSCluster(cluster, true);
3021 }
3022
3023
3024
3025
3026
3027
3028
3029
3030
3031 public void setDFSCluster(MiniDFSCluster cluster, boolean requireDown)
3032 throws IllegalStateException, IOException {
3033 if (dfsCluster != null && requireDown && dfsCluster.isClusterUp()) {
3034 throw new IllegalStateException("DFSCluster is already running! Shut it down first.");
3035 }
3036 this.dfsCluster = cluster;
3037 this.setFs();
3038 }
3039
3040 public FileSystem getTestFileSystem() throws IOException {
3041 return HFileSystem.get(conf);
3042 }
3043
3044
3045
3046
3047
3048
3049
3050
3051 public void waitTableAvailable(TableName table)
3052 throws InterruptedException, IOException {
3053 waitTableAvailable(table.getName(), 30000);
3054 }
3055
3056 public void waitTableAvailable(TableName table, long timeoutMillis)
3057 throws InterruptedException, IOException {
3058 waitFor(timeoutMillis, predicateTableAvailable(table));
3059 }
3060
3061 public String explainTableAvailability(TableName tableName) throws IOException {
3062 String msg = explainTableState(tableName) + ",";
3063 if (getHBaseCluster().getMaster().isAlive()) {
3064 Map<HRegionInfo, ServerName> assignments =
3065 getHBaseCluster().getMaster().getAssignmentManager().getRegionStates()
3066 .getRegionAssignments();
3067 final List<Pair<HRegionInfo, ServerName>> metaLocations =
3068 MetaTableAccessor
3069 .getTableRegionsAndLocations(getZooKeeperWatcher(), connection, tableName);
3070 for (Pair<HRegionInfo, ServerName> metaLocation : metaLocations) {
3071 HRegionInfo hri = metaLocation.getFirst();
3072 ServerName sn = metaLocation.getSecond();
3073 if (!assignments.containsKey(hri)) {
3074 msg += ", region " + hri
3075 + " not assigned, but found in meta, it expected to be on " + sn;
3076
3077 } else if (sn == null) {
3078 msg += ", region " + hri
3079 + " assigned, but has no server in meta";
3080 } else if (!sn.equals(assignments.get(hri))) {
3081 msg += ", region " + hri
3082 + " assigned, but has different servers in meta and AM ( " +
3083 sn + " <> " + assignments.get(hri);
3084 }
3085 }
3086 }
3087 return msg;
3088 }
3089
3090 public String explainTableState(TableName tableName) throws IOException {
3091 try {
3092 if (getHBaseAdmin().isTableEnabled(tableName))
3093 return "table enabled in zk";
3094 else if (getHBaseAdmin().isTableDisabled(tableName))
3095 return "table disabled in zk";
3096 else
3097 return "table in uknown state";
3098 } catch (TableNotFoundException e) {
3099 return "table not exists";
3100 }
3101 }
3102
3103
3104
3105
3106
3107
3108
3109
3110 public void waitTableAvailable(byte[] table, long timeoutMillis)
3111 throws InterruptedException, IOException {
3112 waitTableAvailable(getHBaseAdmin(), table, timeoutMillis);
3113 }
3114
3115 public void waitTableAvailable(Admin admin, byte[] table, long timeoutMillis)
3116 throws InterruptedException, IOException {
3117 long startWait = System.currentTimeMillis();
3118 while (!admin.isTableAvailable(TableName.valueOf(table))) {
3119 assertTrue("Timed out waiting for table to become available " +
3120 Bytes.toStringBinary(table),
3121 System.currentTimeMillis() - startWait < timeoutMillis);
3122 Thread.sleep(200);
3123 }
3124 }
3125
3126
3127
3128
3129
3130
3131
3132
3133
3134
3135 public void waitTableEnabled(TableName table)
3136 throws InterruptedException, IOException {
3137 waitTableEnabled(table, 30000);
3138 }
3139
3140
3141
3142
3143
3144
3145
3146
3147
3148
3149 public void waitTableEnabled(byte[] table, long timeoutMillis)
3150 throws InterruptedException, IOException {
3151 waitTableEnabled(TableName.valueOf(table), timeoutMillis);
3152 }
3153
3154 public void waitTableEnabled(TableName table, long timeoutMillis)
3155 throws IOException {
3156 waitFor(timeoutMillis, predicateTableEnabled(table));
3157 }
3158
3159
3160
3161
3162
3163
3164
3165
3166 public void waitTableDisabled(byte[] table)
3167 throws InterruptedException, IOException {
3168 waitTableDisabled(getHBaseAdmin(), table, 30000);
3169 }
3170
3171 public void waitTableDisabled(Admin admin, byte[] table)
3172 throws InterruptedException, IOException {
3173 waitTableDisabled(admin, table, 30000);
3174 }
3175
3176
3177
3178
3179
3180
3181
3182
3183 public void waitTableDisabled(byte[] table, long timeoutMillis)
3184 throws InterruptedException, IOException {
3185 waitTableDisabled(getHBaseAdmin(), table, timeoutMillis);
3186 }
3187
3188 public void waitTableDisabled(Admin admin, byte[] table, long timeoutMillis)
3189 throws InterruptedException, IOException {
3190 TableName tableName = TableName.valueOf(table);
3191 long startWait = System.currentTimeMillis();
3192 while (!admin.isTableDisabled(tableName)) {
3193 assertTrue("Timed out waiting for table to become disabled " +
3194 Bytes.toStringBinary(table),
3195 System.currentTimeMillis() - startWait < timeoutMillis);
3196 Thread.sleep(200);
3197 }
3198 }
3199
3200
3201
3202
3203
3204
3205
3206
3207 public boolean ensureSomeRegionServersAvailable(final int num)
3208 throws IOException {
3209 boolean startedServer = false;
3210 MiniHBaseCluster hbaseCluster = getMiniHBaseCluster();
3211 for (int i=hbaseCluster.getLiveRegionServerThreads().size(); i<num; ++i) {
3212 LOG.info("Started new server=" + hbaseCluster.startRegionServer());
3213 startedServer = true;
3214 }
3215
3216 return startedServer;
3217 }
3218
3219
3220
3221
3222
3223
3224
3225
3226
3227
3228 public boolean ensureSomeNonStoppedRegionServersAvailable(final int num)
3229 throws IOException {
3230 boolean startedServer = ensureSomeRegionServersAvailable(num);
3231
3232 int nonStoppedServers = 0;
3233 for (JVMClusterUtil.RegionServerThread rst :
3234 getMiniHBaseCluster().getRegionServerThreads()) {
3235
3236 HRegionServer hrs = rst.getRegionServer();
3237 if (hrs.isStopping() || hrs.isStopped()) {
3238 LOG.info("A region server is stopped or stopping:"+hrs);
3239 } else {
3240 nonStoppedServers++;
3241 }
3242 }
3243 for (int i=nonStoppedServers; i<num; ++i) {
3244 LOG.info("Started new server=" + getMiniHBaseCluster().startRegionServer());
3245 startedServer = true;
3246 }
3247 return startedServer;
3248 }
3249
3250
3251
3252
3253
3254
3255
3256
3257
3258
3259
3260 public static User getDifferentUser(final Configuration c,
3261 final String differentiatingSuffix)
3262 throws IOException {
3263 FileSystem currentfs = FileSystem.get(c);
3264 if (!(currentfs instanceof DistributedFileSystem)) {
3265 return User.getCurrent();
3266 }
3267
3268
3269 String username = User.getCurrent().getName() +
3270 differentiatingSuffix;
3271 User user = User.createUserForTesting(c, username,
3272 new String[]{"supergroup"});
3273 return user;
3274 }
3275
3276 public static NavigableSet<String> getAllOnlineRegions(MiniHBaseCluster cluster)
3277 throws IOException {
3278 NavigableSet<String> online = new TreeSet<String>();
3279 for (RegionServerThread rst : cluster.getLiveRegionServerThreads()) {
3280 try {
3281 for (HRegionInfo region :
3282 ProtobufUtil.getOnlineRegions(rst.getRegionServer().getRSRpcServices())) {
3283 online.add(region.getRegionNameAsString());
3284 }
3285 } catch (RegionServerStoppedException e) {
3286
3287 }
3288 }
3289 for (MasterThread mt : cluster.getLiveMasterThreads()) {
3290 try {
3291 for (HRegionInfo region :
3292 ProtobufUtil.getOnlineRegions(mt.getMaster().getRSRpcServices())) {
3293 online.add(region.getRegionNameAsString());
3294 }
3295 } catch (RegionServerStoppedException e) {
3296
3297 } catch (ServerNotRunningYetException e) {
3298
3299 }
3300 }
3301 return online;
3302 }
3303
3304
3305
3306
3307
3308
3309
3310
3311
3312
3313
3314
3315
3316
3317 public static void setMaxRecoveryErrorCount(final OutputStream stream,
3318 final int max) {
3319 try {
3320 Class<?> [] clazzes = DFSClient.class.getDeclaredClasses();
3321 for (Class<?> clazz: clazzes) {
3322 String className = clazz.getSimpleName();
3323 if (className.equals("DFSOutputStream")) {
3324 if (clazz.isInstance(stream)) {
3325 Field maxRecoveryErrorCountField =
3326 stream.getClass().getDeclaredField("maxRecoveryErrorCount");
3327 maxRecoveryErrorCountField.setAccessible(true);
3328 maxRecoveryErrorCountField.setInt(stream, max);
3329 break;
3330 }
3331 }
3332 }
3333 } catch (Exception e) {
3334 LOG.info("Could not set max recovery field", e);
3335 }
3336 }
3337
3338
3339
3340
3341
3342
3343
3344
3345
3346 public void waitUntilAllRegionsAssigned(final TableName tableName) throws IOException {
3347 waitUntilAllRegionsAssigned(tableName, 60000);
3348 }
3349
3350
3351
3352
3353
3354
3355
3356
3357
3358
3359 public void waitUntilAllRegionsAssigned(final TableName tableName, final long timeout)
3360 throws IOException {
3361 final Table meta = new HTable(getConfiguration(), TableName.META_TABLE_NAME);
3362 try {
3363 waitFor(timeout, 200, true, new Predicate<IOException>() {
3364 @Override
3365 public boolean evaluate() throws IOException {
3366 boolean allRegionsAssigned = true;
3367 Scan scan = new Scan();
3368 scan.addFamily(HConstants.CATALOG_FAMILY);
3369 ResultScanner s = meta.getScanner(scan);
3370 try {
3371 Result r;
3372 while ((r = s.next()) != null) {
3373 byte[] b = r.getValue(HConstants.CATALOG_FAMILY, HConstants.REGIONINFO_QUALIFIER);
3374 HRegionInfo info = HRegionInfo.parseFromOrNull(b);
3375 if (info != null && info.getTable().equals(tableName)) {
3376 b = r.getValue(HConstants.CATALOG_FAMILY, HConstants.SERVER_QUALIFIER);
3377 allRegionsAssigned &= (b != null);
3378 }
3379 }
3380 } finally {
3381 s.close();
3382 }
3383 return allRegionsAssigned;
3384 }
3385 });
3386 } finally {
3387 meta.close();
3388 }
3389
3390
3391 if (!getHBaseClusterInterface().isDistributedCluster()) {
3392
3393
3394 HMaster master = getHBaseCluster().getMaster();
3395 final RegionStates states = master.getAssignmentManager().getRegionStates();
3396 waitFor(timeout, 200, new ExplainingPredicate<IOException>() {
3397 @Override
3398 public String explainFailure() throws IOException {
3399 return explainTableAvailability(tableName);
3400 }
3401
3402 @Override
3403 public boolean evaluate() throws IOException {
3404 List<HRegionInfo> hris = states.getRegionsOfTable(tableName);
3405 return hris != null && !hris.isEmpty();
3406 }
3407 });
3408 }
3409 }
3410
3411
3412
3413
3414
3415 public static List<Cell> getFromStoreFile(HStore store,
3416 Get get) throws IOException {
3417 Scan scan = new Scan(get);
3418 InternalScanner scanner = (InternalScanner) store.getScanner(scan,
3419 scan.getFamilyMap().get(store.getFamily().getName()),
3420
3421
3422 0);
3423
3424 List<Cell> result = new ArrayList<Cell>();
3425 scanner.next(result);
3426 if (!result.isEmpty()) {
3427
3428 Cell kv = result.get(0);
3429 if (!CellUtil.matchingRow(kv, get.getRow())) {
3430 result.clear();
3431 }
3432 }
3433 scanner.close();
3434 return result;
3435 }
3436
3437
3438
3439
3440
3441
3442
3443
3444
3445 public byte[][] getRegionSplitStartKeys(byte[] startKey, byte[] endKey, int numRegions){
3446 assertTrue(numRegions>3);
3447 byte [][] tmpSplitKeys = Bytes.split(startKey, endKey, numRegions - 3);
3448 byte [][] result = new byte[tmpSplitKeys.length+1][];
3449 System.arraycopy(tmpSplitKeys, 0, result, 1, tmpSplitKeys.length);
3450 result[0] = HConstants.EMPTY_BYTE_ARRAY;
3451 return result;
3452 }
3453
3454
3455
3456
3457
3458 public static List<Cell> getFromStoreFile(HStore store,
3459 byte [] row,
3460 NavigableSet<byte[]> columns
3461 ) throws IOException {
3462 Get get = new Get(row);
3463 Map<byte[], NavigableSet<byte[]>> s = get.getFamilyMap();
3464 s.put(store.getFamily().getName(), columns);
3465
3466 return getFromStoreFile(store,get);
3467 }
3468
3469
3470
3471
3472
3473 public static ZooKeeperWatcher getZooKeeperWatcher(
3474 HBaseTestingUtility TEST_UTIL) throws ZooKeeperConnectionException,
3475 IOException {
3476 ZooKeeperWatcher zkw = new ZooKeeperWatcher(TEST_UTIL.getConfiguration(),
3477 "unittest", new Abortable() {
3478 boolean aborted = false;
3479
3480 @Override
3481 public void abort(String why, Throwable e) {
3482 aborted = true;
3483 throw new RuntimeException("Fatal ZK error, why=" + why, e);
3484 }
3485
3486 @Override
3487 public boolean isAborted() {
3488 return aborted;
3489 }
3490 });
3491 return zkw;
3492 }
3493
3494
3495
3496
3497
3498
3499
3500
3501
3502
3503
3504
3505 public static ZooKeeperWatcher createAndForceNodeToOpenedState(
3506 HBaseTestingUtility TEST_UTIL, Region region,
3507 ServerName serverName) throws ZooKeeperConnectionException,
3508 IOException, KeeperException, NodeExistsException {
3509 return createAndForceNodeToOpenedState(TEST_UTIL, (HRegion)region, serverName);
3510 }
3511
3512
3513
3514
3515
3516
3517
3518
3519
3520
3521
3522
3523 public static ZooKeeperWatcher createAndForceNodeToOpenedState(
3524 HBaseTestingUtility TEST_UTIL, HRegion region,
3525 ServerName serverName) throws ZooKeeperConnectionException,
3526 IOException, KeeperException, NodeExistsException {
3527 ZooKeeperWatcher zkw = getZooKeeperWatcher(TEST_UTIL);
3528 ZKAssign.createNodeOffline(zkw, region.getRegionInfo(), serverName);
3529 int version = ZKAssign.transitionNodeOpening(zkw, region
3530 .getRegionInfo(), serverName);
3531 ZKAssign.transitionNodeOpened(zkw, region.getRegionInfo(), serverName,
3532 version);
3533 return zkw;
3534 }
3535
3536 public static void assertKVListsEqual(String additionalMsg,
3537 final List<? extends Cell> expected,
3538 final List<? extends Cell> actual) {
3539 final int eLen = expected.size();
3540 final int aLen = actual.size();
3541 final int minLen = Math.min(eLen, aLen);
3542
3543 int i;
3544 for (i = 0; i < minLen
3545 && KeyValue.COMPARATOR.compare(expected.get(i), actual.get(i)) == 0;
3546 ++i) {}
3547
3548 if (additionalMsg == null) {
3549 additionalMsg = "";
3550 }
3551 if (!additionalMsg.isEmpty()) {
3552 additionalMsg = ". " + additionalMsg;
3553 }
3554
3555 if (eLen != aLen || i != minLen) {
3556 throw new AssertionError(
3557 "Expected and actual KV arrays differ at position " + i + ": " +
3558 safeGetAsStr(expected, i) + " (length " + eLen +") vs. " +
3559 safeGetAsStr(actual, i) + " (length " + aLen + ")" + additionalMsg);
3560 }
3561 }
3562
3563 public static <T> String safeGetAsStr(List<T> lst, int i) {
3564 if (0 <= i && i < lst.size()) {
3565 return lst.get(i).toString();
3566 } else {
3567 return "<out_of_range>";
3568 }
3569 }
3570
3571 public String getClusterKey() {
3572 return conf.get(HConstants.ZOOKEEPER_QUORUM) + ":"
3573 + conf.get(HConstants.ZOOKEEPER_CLIENT_PORT) + ":"
3574 + conf.get(HConstants.ZOOKEEPER_ZNODE_PARENT,
3575 HConstants.DEFAULT_ZOOKEEPER_ZNODE_PARENT);
3576 }
3577
3578
3579 public HTable createRandomTable(String tableName,
3580 final Collection<String> families,
3581 final int maxVersions,
3582 final int numColsPerRow,
3583 final int numFlushes,
3584 final int numRegions,
3585 final int numRowsPerFlush)
3586 throws IOException, InterruptedException {
3587
3588 LOG.info("\n\nCreating random table " + tableName + " with " + numRegions +
3589 " regions, " + numFlushes + " storefiles per region, " +
3590 numRowsPerFlush + " rows per flush, maxVersions=" + maxVersions +
3591 "\n");
3592
3593 final Random rand = new Random(tableName.hashCode() * 17L + 12938197137L);
3594 final int numCF = families.size();
3595 final byte[][] cfBytes = new byte[numCF][];
3596 {
3597 int cfIndex = 0;
3598 for (String cf : families) {
3599 cfBytes[cfIndex++] = Bytes.toBytes(cf);
3600 }
3601 }
3602
3603 final int actualStartKey = 0;
3604 final int actualEndKey = Integer.MAX_VALUE;
3605 final int keysPerRegion = (actualEndKey - actualStartKey) / numRegions;
3606 final int splitStartKey = actualStartKey + keysPerRegion;
3607 final int splitEndKey = actualEndKey - keysPerRegion;
3608 final String keyFormat = "%08x";
3609 final HTable table = createTable(tableName, cfBytes,
3610 maxVersions,
3611 Bytes.toBytes(String.format(keyFormat, splitStartKey)),
3612 Bytes.toBytes(String.format(keyFormat, splitEndKey)),
3613 numRegions);
3614
3615 if (hbaseCluster != null) {
3616 getMiniHBaseCluster().flushcache(TableName.META_TABLE_NAME);
3617 }
3618
3619 for (int iFlush = 0; iFlush < numFlushes; ++iFlush) {
3620 for (int iRow = 0; iRow < numRowsPerFlush; ++iRow) {
3621 final byte[] row = Bytes.toBytes(String.format(keyFormat,
3622 actualStartKey + rand.nextInt(actualEndKey - actualStartKey)));
3623
3624 Put put = new Put(row);
3625 Delete del = new Delete(row);
3626 for (int iCol = 0; iCol < numColsPerRow; ++iCol) {
3627 final byte[] cf = cfBytes[rand.nextInt(numCF)];
3628 final long ts = rand.nextInt();
3629 final byte[] qual = Bytes.toBytes("col" + iCol);
3630 if (rand.nextBoolean()) {
3631 final byte[] value = Bytes.toBytes("value_for_row_" + iRow +
3632 "_cf_" + Bytes.toStringBinary(cf) + "_col_" + iCol + "_ts_" +
3633 ts + "_random_" + rand.nextLong());
3634 put.add(cf, qual, ts, value);
3635 } else if (rand.nextDouble() < 0.8) {
3636 del.deleteColumn(cf, qual, ts);
3637 } else {
3638 del.deleteColumns(cf, qual, ts);
3639 }
3640 }
3641
3642 if (!put.isEmpty()) {
3643 table.put(put);
3644 }
3645
3646 if (!del.isEmpty()) {
3647 table.delete(del);
3648 }
3649 }
3650 LOG.info("Initiating flush #" + iFlush + " for table " + tableName);
3651 table.flushCommits();
3652 if (hbaseCluster != null) {
3653 getMiniHBaseCluster().flushcache(table.getName());
3654 }
3655 }
3656
3657 return table;
3658 }
3659
3660 private static final int MIN_RANDOM_PORT = 0xc000;
3661 private static final int MAX_RANDOM_PORT = 0xfffe;
3662 private static Random random = new Random();
3663
3664
3665
3666
3667
3668 public static int randomPort() {
3669 return MIN_RANDOM_PORT
3670 + random.nextInt(MAX_RANDOM_PORT - MIN_RANDOM_PORT);
3671 }
3672
3673
3674
3675
3676
3677 public static int randomFreePort() {
3678 int port = 0;
3679 do {
3680 port = randomPort();
3681 if (takenRandomPorts.contains(port)) {
3682 continue;
3683 }
3684 takenRandomPorts.add(port);
3685
3686 try {
3687 ServerSocket sock = new ServerSocket(port);
3688 sock.close();
3689 } catch (IOException ex) {
3690 port = 0;
3691 }
3692 } while (port == 0);
3693 return port;
3694 }
3695
3696
3697 public static String randomMultiCastAddress() {
3698 return "226.1.1." + random.nextInt(254);
3699 }
3700
3701
3702
3703 public static void waitForHostPort(String host, int port)
3704 throws IOException {
3705 final int maxTimeMs = 10000;
3706 final int maxNumAttempts = maxTimeMs / HConstants.SOCKET_RETRY_WAIT_MS;
3707 IOException savedException = null;
3708 LOG.info("Waiting for server at " + host + ":" + port);
3709 for (int attempt = 0; attempt < maxNumAttempts; ++attempt) {
3710 try {
3711 Socket sock = new Socket(InetAddress.getByName(host), port);
3712 sock.close();
3713 savedException = null;
3714 LOG.info("Server at " + host + ":" + port + " is available");
3715 break;
3716 } catch (UnknownHostException e) {
3717 throw new IOException("Failed to look up " + host, e);
3718 } catch (IOException e) {
3719 savedException = e;
3720 }
3721 Threads.sleepWithoutInterrupt(HConstants.SOCKET_RETRY_WAIT_MS);
3722 }
3723
3724 if (savedException != null) {
3725 throw savedException;
3726 }
3727 }
3728
3729
3730
3731
3732
3733
3734 public static int createPreSplitLoadTestTable(Configuration conf,
3735 TableName tableName, byte[] columnFamily, Algorithm compression,
3736 DataBlockEncoding dataBlockEncoding) throws IOException {
3737 return createPreSplitLoadTestTable(conf, tableName,
3738 columnFamily, compression, dataBlockEncoding, DEFAULT_REGIONS_PER_SERVER, 1,
3739 Durability.USE_DEFAULT);
3740 }
3741
3742
3743
3744
3745
3746 public static int createPreSplitLoadTestTable(Configuration conf,
3747 TableName tableName, byte[] columnFamily, Algorithm compression,
3748 DataBlockEncoding dataBlockEncoding, int numRegionsPerServer, int regionReplication,
3749 Durability durability)
3750 throws IOException {
3751 HTableDescriptor desc = new HTableDescriptor(tableName);
3752 desc.setDurability(durability);
3753 desc.setRegionReplication(regionReplication);
3754 HColumnDescriptor hcd = new HColumnDescriptor(columnFamily);
3755 hcd.setDataBlockEncoding(dataBlockEncoding);
3756 hcd.setCompressionType(compression);
3757 return createPreSplitLoadTestTable(conf, desc, hcd, numRegionsPerServer);
3758 }
3759
3760
3761
3762
3763
3764
3765 public static int createPreSplitLoadTestTable(Configuration conf,
3766 TableName tableName, byte[][] columnFamilies, Algorithm compression,
3767 DataBlockEncoding dataBlockEncoding, int numRegionsPerServer, int regionReplication,
3768 Durability durability)
3769 throws IOException {
3770 HTableDescriptor desc = new HTableDescriptor(tableName);
3771 desc.setDurability(durability);
3772 desc.setRegionReplication(regionReplication);
3773 HColumnDescriptor[] hcds = new HColumnDescriptor[columnFamilies.length];
3774 for (int i = 0; i < columnFamilies.length; i++) {
3775 HColumnDescriptor hcd = new HColumnDescriptor(columnFamilies[i]);
3776 hcd.setDataBlockEncoding(dataBlockEncoding);
3777 hcd.setCompressionType(compression);
3778 hcds[i] = hcd;
3779 }
3780 return createPreSplitLoadTestTable(conf, desc, hcds, numRegionsPerServer);
3781 }
3782
3783
3784
3785
3786
3787
3788 public static int createPreSplitLoadTestTable(Configuration conf,
3789 HTableDescriptor desc, HColumnDescriptor hcd) throws IOException {
3790 return createPreSplitLoadTestTable(conf, desc, hcd, DEFAULT_REGIONS_PER_SERVER);
3791 }
3792
3793
3794
3795
3796
3797
3798 public static int createPreSplitLoadTestTable(Configuration conf,
3799 HTableDescriptor desc, HColumnDescriptor hcd, int numRegionsPerServer) throws IOException {
3800 return createPreSplitLoadTestTable(conf, desc, new HColumnDescriptor[] {hcd},
3801 numRegionsPerServer);
3802 }
3803
3804
3805
3806
3807
3808
3809 public static int createPreSplitLoadTestTable(Configuration conf,
3810 HTableDescriptor desc, HColumnDescriptor[] hcds, int numRegionsPerServer) throws IOException {
3811 for (HColumnDescriptor hcd : hcds) {
3812 if (!desc.hasFamily(hcd.getName())) {
3813 desc.addFamily(hcd);
3814 }
3815 }
3816
3817 int totalNumberOfRegions = 0;
3818 Connection unmanagedConnection = ConnectionFactory.createConnection(conf);
3819 Admin admin = unmanagedConnection.getAdmin();
3820
3821 try {
3822
3823
3824
3825 int numberOfServers = admin.getClusterStatus().getServers().size();
3826 if (numberOfServers == 0) {
3827 throw new IllegalStateException("No live regionservers");
3828 }
3829
3830 totalNumberOfRegions = numberOfServers * numRegionsPerServer;
3831 LOG.info("Number of live regionservers: " + numberOfServers + ", " +
3832 "pre-splitting table into " + totalNumberOfRegions + " regions " +
3833 "(regions per server: " + numRegionsPerServer + ")");
3834
3835 byte[][] splits = new RegionSplitter.HexStringSplit().split(
3836 totalNumberOfRegions);
3837
3838 admin.createTable(desc, splits);
3839 } catch (MasterNotRunningException e) {
3840 LOG.error("Master not running", e);
3841 throw new IOException(e);
3842 } catch (TableExistsException e) {
3843 LOG.warn("Table " + desc.getTableName() +
3844 " already exists, continuing");
3845 } finally {
3846 admin.close();
3847 unmanagedConnection.close();
3848 }
3849 return totalNumberOfRegions;
3850 }
3851
3852 public static int getMetaRSPort(Configuration conf) throws IOException {
3853 try (Connection c = ConnectionFactory.createConnection();
3854 RegionLocator locator = c.getRegionLocator(TableName.META_TABLE_NAME)) {
3855 return locator.getRegionLocation(Bytes.toBytes("")).getPort();
3856 }
3857 }
3858
3859
3860
3861
3862
3863
3864
3865 public void assertRegionOnServer(
3866 final HRegionInfo hri, final ServerName server,
3867 final long timeout) throws IOException, InterruptedException {
3868 long timeoutTime = System.currentTimeMillis() + timeout;
3869 while (true) {
3870 List<HRegionInfo> regions = getHBaseAdmin().getOnlineRegions(server);
3871 if (regions.contains(hri)) return;
3872 long now = System.currentTimeMillis();
3873 if (now > timeoutTime) break;
3874 Thread.sleep(10);
3875 }
3876 fail("Could not find region " + hri.getRegionNameAsString()
3877 + " on server " + server);
3878 }
3879
3880
3881
3882
3883
3884 public void assertRegionOnlyOnServer(
3885 final HRegionInfo hri, final ServerName server,
3886 final long timeout) throws IOException, InterruptedException {
3887 long timeoutTime = System.currentTimeMillis() + timeout;
3888 while (true) {
3889 List<HRegionInfo> regions = getHBaseAdmin().getOnlineRegions(server);
3890 if (regions.contains(hri)) {
3891 List<JVMClusterUtil.RegionServerThread> rsThreads =
3892 getHBaseCluster().getLiveRegionServerThreads();
3893 for (JVMClusterUtil.RegionServerThread rsThread: rsThreads) {
3894 HRegionServer rs = rsThread.getRegionServer();
3895 if (server.equals(rs.getServerName())) {
3896 continue;
3897 }
3898 Collection<Region> hrs = rs.getOnlineRegionsLocalContext();
3899 for (Region r: hrs) {
3900 assertTrue("Region should not be double assigned",
3901 r.getRegionInfo().getRegionId() != hri.getRegionId());
3902 }
3903 }
3904 return;
3905 }
3906 long now = System.currentTimeMillis();
3907 if (now > timeoutTime) break;
3908 Thread.sleep(10);
3909 }
3910 fail("Could not find region " + hri.getRegionNameAsString()
3911 + " on server " + server);
3912 }
3913
3914 public HRegion createTestRegion(String tableName, HColumnDescriptor hcd)
3915 throws IOException {
3916 HTableDescriptor htd = new HTableDescriptor(TableName.valueOf(tableName));
3917 htd.addFamily(hcd);
3918 HRegionInfo info =
3919 new HRegionInfo(TableName.valueOf(tableName), null, null, false);
3920 HRegion region =
3921 HRegion.createHRegion(info, getDataTestDir(), getConfiguration(), htd);
3922 return region;
3923 }
3924
3925 public void setFileSystemURI(String fsURI) {
3926 FS_URI = fsURI;
3927 }
3928
3929
3930
3931
3932 public <E extends Exception> long waitFor(long timeout, Predicate<E> predicate)
3933 throws E {
3934 return Waiter.waitFor(this.conf, timeout, predicate);
3935 }
3936
3937
3938
3939
3940 public <E extends Exception> long waitFor(long timeout, long interval, Predicate<E> predicate)
3941 throws E {
3942 return Waiter.waitFor(this.conf, timeout, interval, predicate);
3943 }
3944
3945
3946
3947
3948 public <E extends Exception> long waitFor(long timeout, long interval,
3949 boolean failIfTimeout, Predicate<E> predicate) throws E {
3950 return Waiter.waitFor(this.conf, timeout, interval, failIfTimeout, predicate);
3951 }
3952
3953
3954
3955
3956 public ExplainingPredicate<IOException> predicateNoRegionsInTransition() {
3957 return new ExplainingPredicate<IOException>() {
3958 @Override
3959 public String explainFailure() throws IOException {
3960 final RegionStates regionStates = getMiniHBaseCluster().getMaster()
3961 .getAssignmentManager().getRegionStates();
3962 return "found in transition: " + regionStates.getRegionsInTransition().toString();
3963 }
3964
3965 @Override
3966 public boolean evaluate() throws IOException {
3967 final RegionStates regionStates = getMiniHBaseCluster().getMaster()
3968 .getAssignmentManager().getRegionStates();
3969 return !regionStates.isRegionsInTransition();
3970 }
3971 };
3972 }
3973
3974
3975
3976
3977 public Waiter.Predicate<IOException> predicateTableEnabled(final TableName tableName) {
3978 return new ExplainingPredicate<IOException>() {
3979 @Override
3980 public String explainFailure() throws IOException {
3981 return explainTableState(tableName);
3982 }
3983
3984 @Override
3985 public boolean evaluate() throws IOException {
3986 return getHBaseAdmin().tableExists(tableName) && getHBaseAdmin().isTableEnabled(tableName);
3987 }
3988 };
3989 }
3990
3991
3992
3993
3994 public Waiter.Predicate<IOException> predicateTableDisabled(final TableName tableName) {
3995 return new ExplainingPredicate<IOException>() {
3996 @Override
3997 public String explainFailure() throws IOException {
3998 return explainTableState(tableName);
3999 }
4000
4001 @Override
4002 public boolean evaluate() throws IOException {
4003 return getHBaseAdmin().isTableDisabled(tableName);
4004 }
4005 };
4006 }
4007
4008
4009
4010
4011 public Waiter.Predicate<IOException> predicateTableAvailable(final TableName tableName) {
4012 return new ExplainingPredicate<IOException>() {
4013 @Override
4014 public String explainFailure() throws IOException {
4015 return explainTableAvailability(tableName);
4016 }
4017
4018 @Override
4019 public boolean evaluate() throws IOException {
4020 boolean tableAvailable = getHBaseAdmin().isTableAvailable(tableName);
4021 if (tableAvailable) {
4022 try {
4023 Canary.sniff(getHBaseAdmin(), tableName);
4024 } catch (Exception e) {
4025 throw new IOException("Canary sniff failed for table " + tableName, e);
4026 }
4027 }
4028 return tableAvailable;
4029 }
4030 };
4031 }
4032
4033
4034
4035
4036
4037
4038 public void waitUntilNoRegionsInTransition(
4039 final long timeout) throws Exception {
4040 waitFor(timeout, predicateNoRegionsInTransition());
4041 }
4042
4043
4044
4045
4046
4047
4048 public void waitLabelAvailable(long timeoutMillis, final String... labels) {
4049 final VisibilityLabelsCache labelsCache = VisibilityLabelsCache.get();
4050 waitFor(timeoutMillis, new Waiter.ExplainingPredicate<RuntimeException>() {
4051
4052 @Override
4053 public boolean evaluate() {
4054 for (String label : labels) {
4055 if (labelsCache.getLabelOrdinal(label) == 0) {
4056 return false;
4057 }
4058 }
4059 return true;
4060 }
4061
4062 @Override
4063 public String explainFailure() {
4064 for (String label : labels) {
4065 if (labelsCache.getLabelOrdinal(label) == 0) {
4066 return label + " is not available yet";
4067 }
4068 }
4069 return "";
4070 }
4071 });
4072 }
4073
4074
4075
4076
4077
4078
4079 public static List<HColumnDescriptor> generateColumnDescriptors() {
4080 return generateColumnDescriptors("");
4081 }
4082
4083
4084
4085
4086
4087
4088
4089 public static List<HColumnDescriptor> generateColumnDescriptors(final String prefix) {
4090 List<HColumnDescriptor> htds = new ArrayList<HColumnDescriptor>();
4091 long familyId = 0;
4092 for (Compression.Algorithm compressionType: getSupportedCompressionAlgorithms()) {
4093 for (DataBlockEncoding encodingType: DataBlockEncoding.values()) {
4094 for (BloomType bloomType: BloomType.values()) {
4095 String name = String.format("%s-cf-!@#&-%d!@#", prefix, familyId);
4096 HColumnDescriptor htd = new HColumnDescriptor(name);
4097 htd.setCompressionType(compressionType);
4098 htd.setDataBlockEncoding(encodingType);
4099 htd.setBloomFilterType(bloomType);
4100 htds.add(htd);
4101 familyId++;
4102 }
4103 }
4104 }
4105 return htds;
4106 }
4107
4108
4109
4110
4111
4112 public static Compression.Algorithm[] getSupportedCompressionAlgorithms() {
4113 String[] allAlgos = HFile.getSupportedCompressionAlgorithms();
4114 List<Compression.Algorithm> supportedAlgos = new ArrayList<Compression.Algorithm>();
4115 for (String algoName : allAlgos) {
4116 try {
4117 Compression.Algorithm algo = Compression.getCompressionAlgorithmByName(algoName);
4118 algo.getCompressor();
4119 supportedAlgos.add(algo);
4120 } catch (Throwable t) {
4121
4122 }
4123 }
4124 return supportedAlgos.toArray(new Algorithm[supportedAlgos.size()]);
4125 }
4126 }