1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19 package org.apache.hadoop.hbase;
20
21 import java.io.IOException;
22 import java.security.PrivilegedAction;
23 import java.util.ArrayList;
24 import java.util.List;
25
26 import org.apache.commons.logging.Log;
27 import org.apache.commons.logging.LogFactory;
28 import org.apache.hadoop.hbase.classification.InterfaceAudience;
29 import org.apache.hadoop.hbase.classification.InterfaceStability;
30 import org.apache.hadoop.conf.Configuration;
31 import org.apache.hadoop.fs.FileSystem;
32 import org.apache.hadoop.hbase.master.HMaster;
33 import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService;
34 import org.apache.hadoop.hbase.protobuf.generated.ClientProtos.ClientService;
35 import org.apache.hadoop.hbase.protobuf.generated.MasterProtos.MasterService;
36 import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionServerStartupResponse;
37 import org.apache.hadoop.hbase.regionserver.HRegion;
38 import org.apache.hadoop.hbase.regionserver.HRegionServer;
39 import org.apache.hadoop.hbase.regionserver.Region;
40 import org.apache.hadoop.hbase.security.User;
41 import org.apache.hadoop.hbase.test.MetricsAssertHelper;
42 import org.apache.hadoop.hbase.util.JVMClusterUtil;
43 import org.apache.hadoop.hbase.util.JVMClusterUtil.MasterThread;
44 import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread;
45 import org.apache.hadoop.hbase.util.Threads;
46
47
48
49
50
51
52
53 @InterfaceAudience.Public
54 @InterfaceStability.Evolving
55 public class MiniHBaseCluster extends HBaseCluster {
56 private static final Log LOG = LogFactory.getLog(MiniHBaseCluster.class.getName());
57 public LocalHBaseCluster hbaseCluster;
58 private static int index;
59
60
61
62
63
64
65
66 public MiniHBaseCluster(Configuration conf, int numRegionServers)
67 throws IOException, InterruptedException {
68 this(conf, 1, numRegionServers);
69 }
70
71
72
73
74
75
76
77
78 public MiniHBaseCluster(Configuration conf, int numMasters, int numRegionServers)
79 throws IOException, InterruptedException {
80 this(conf, numMasters, numRegionServers, null, null);
81 }
82
83 public MiniHBaseCluster(Configuration conf, int numMasters, int numRegionServers,
84 Class<? extends HMaster> masterClass,
85 Class<? extends MiniHBaseCluster.MiniHBaseClusterRegionServer> regionserverClass)
86 throws IOException, InterruptedException {
87 super(conf);
88 conf.set(HConstants.MASTER_PORT, "0");
89 if (conf.getInt(HConstants.MASTER_INFO_PORT, 0) != -1) {
90 conf.set(HConstants.MASTER_INFO_PORT, "0");
91 }
92
93
94 CompatibilityFactory.getInstance(MetricsAssertHelper.class).init();
95
96 init(numMasters, numRegionServers, masterClass, regionserverClass);
97 this.initialClusterStatus = getClusterStatus();
98 }
99
100 public Configuration getConfiguration() {
101 return this.conf;
102 }
103
104
105
106
107
108
109
110 public static class MiniHBaseClusterRegionServer extends HRegionServer {
111 private Thread shutdownThread = null;
112 private User user = null;
113
114 public MiniHBaseClusterRegionServer(Configuration conf, CoordinatedStateManager cp)
115 throws IOException, InterruptedException {
116 super(conf, cp);
117 this.user = User.getCurrent();
118 }
119
120
121
122
123
124
125
126
127
128 @Override
129 protected void handleReportForDutyResponse(
130 final RegionServerStartupResponse c) throws IOException {
131 super.handleReportForDutyResponse(c);
132
133 this.shutdownThread = new SingleFileSystemShutdownThread(getFileSystem());
134 }
135
136 @Override
137 public void run() {
138 try {
139 this.user.runAs(new PrivilegedAction<Object>(){
140 public Object run() {
141 runRegionServer();
142 return null;
143 }
144 });
145 } catch (Throwable t) {
146 LOG.error("Exception in run", t);
147 } finally {
148
149 if (this.shutdownThread != null) {
150 this.shutdownThread.start();
151 Threads.shutdown(this.shutdownThread, 30000);
152 }
153 }
154 }
155
156 private void runRegionServer() {
157 super.run();
158 }
159
160 @Override
161 public void kill() {
162 super.kill();
163 }
164
165 @Override
166 public void abort(final String reason, final Throwable cause) {
167 this.user.runAs(new PrivilegedAction<Object>() {
168 public Object run() {
169 abortRegionServer(reason, cause);
170 return null;
171 }
172 });
173 }
174
175 private void abortRegionServer(String reason, Throwable cause) {
176 super.abort(reason, cause);
177 }
178 }
179
180
181
182
183
184 static class SingleFileSystemShutdownThread extends Thread {
185 private final FileSystem fs;
186 SingleFileSystemShutdownThread(final FileSystem fs) {
187 super("Shutdown of " + fs);
188 this.fs = fs;
189 }
190 @Override
191 public void run() {
192 try {
193 LOG.info("Hook closing fs=" + this.fs);
194 this.fs.close();
195 } catch (NullPointerException npe) {
196 LOG.debug("Need to fix these: " + npe.toString());
197 } catch (IOException e) {
198 LOG.warn("Running hook", e);
199 }
200 }
201 }
202
203 private void init(final int nMasterNodes, final int nRegionNodes,
204 Class<? extends HMaster> masterClass,
205 Class<? extends MiniHBaseCluster.MiniHBaseClusterRegionServer> regionserverClass)
206 throws IOException, InterruptedException {
207 try {
208 if (masterClass == null){
209 masterClass = HMaster.class;
210 }
211 if (regionserverClass == null){
212 regionserverClass = MiniHBaseCluster.MiniHBaseClusterRegionServer.class;
213 }
214
215
216 hbaseCluster = new LocalHBaseCluster(conf, nMasterNodes, 0,
217 masterClass, regionserverClass);
218
219
220 for (int i=0; i<nRegionNodes; i++) {
221 Configuration rsConf = HBaseConfiguration.create(conf);
222 User user = HBaseTestingUtility.getDifferentUser(rsConf,
223 ".hfs."+index++);
224 hbaseCluster.addRegionServer(rsConf, i, user);
225 }
226
227 hbaseCluster.startup();
228 } catch (IOException e) {
229 shutdown();
230 throw e;
231 } catch (Throwable t) {
232 LOG.error("Error starting cluster", t);
233 shutdown();
234 throw new IOException("Shutting down", t);
235 }
236 }
237
238 @Override
239 public void startRegionServer(String hostname, int port) throws IOException {
240 this.startRegionServer();
241 }
242
243 @Override
244 public void killRegionServer(ServerName serverName) throws IOException {
245 HRegionServer server = getRegionServer(getRegionServerIndex(serverName));
246 if (server instanceof MiniHBaseClusterRegionServer) {
247 LOG.info("Killing " + server.toString());
248 ((MiniHBaseClusterRegionServer) server).kill();
249 } else {
250 abortRegionServer(getRegionServerIndex(serverName));
251 }
252 }
253
254 @Override
255 public void stopRegionServer(ServerName serverName) throws IOException {
256 stopRegionServer(getRegionServerIndex(serverName));
257 }
258
259 @Override
260 public void waitForRegionServerToStop(ServerName serverName, long timeout) throws IOException {
261
262 waitOnRegionServer(getRegionServerIndex(serverName));
263 }
264
265 @Override
266 public void startZkNode(String hostname, int port) throws IOException {
267 LOG.warn("Starting zookeeper nodes on mini cluster is not supported");
268 }
269
270 @Override
271 public void killZkNode(ServerName serverName) throws IOException {
272 LOG.warn("Aborting zookeeper nodes on mini cluster is not supported");
273 }
274
275 @Override
276 public void stopZkNode(ServerName serverName) throws IOException {
277 LOG.warn("Stopping zookeeper nodes on mini cluster is not supported");
278 }
279
280 @Override
281 public void waitForZkNodeToStart(ServerName serverName, long timeout) throws IOException {
282 LOG.warn("Waiting for zookeeper nodes to start on mini cluster is not supported");
283 }
284
285 @Override
286 public void waitForZkNodeToStop(ServerName serverName, long timeout) throws IOException {
287 LOG.warn("Waiting for zookeeper nodes to stop on mini cluster is not supported");
288 }
289
290 @Override
291 public void startDataNode(ServerName serverName) throws IOException {
292 LOG.warn("Starting datanodes on mini cluster is not supported");
293 }
294
295 @Override
296 public void killDataNode(ServerName serverName) throws IOException {
297 LOG.warn("Aborting datanodes on mini cluster is not supported");
298 }
299
300 @Override
301 public void stopDataNode(ServerName serverName) throws IOException {
302 LOG.warn("Stopping datanodes on mini cluster is not supported");
303 }
304
305 @Override
306 public void waitForDataNodeToStart(ServerName serverName, long timeout) throws IOException {
307 LOG.warn("Waiting for datanodes to start on mini cluster is not supported");
308 }
309
310 @Override
311 public void waitForDataNodeToStop(ServerName serverName, long timeout) throws IOException {
312 LOG.warn("Waiting for datanodes to stop on mini cluster is not supported");
313 }
314
315 @Override
316 public void startMaster(String hostname, int port) throws IOException {
317 this.startMaster();
318 }
319
320 @Override
321 public void killMaster(ServerName serverName) throws IOException {
322 abortMaster(getMasterIndex(serverName));
323 }
324
325 @Override
326 public void stopMaster(ServerName serverName) throws IOException {
327 stopMaster(getMasterIndex(serverName));
328 }
329
330 @Override
331 public void waitForMasterToStop(ServerName serverName, long timeout) throws IOException {
332
333 waitOnMaster(getMasterIndex(serverName));
334 }
335
336
337
338
339
340
341
342 public JVMClusterUtil.RegionServerThread startRegionServer()
343 throws IOException {
344 final Configuration newConf = HBaseConfiguration.create(conf);
345 User rsUser =
346 HBaseTestingUtility.getDifferentUser(newConf, ".hfs."+index++);
347 JVMClusterUtil.RegionServerThread t = null;
348 try {
349 t = hbaseCluster.addRegionServer(
350 newConf, hbaseCluster.getRegionServers().size(), rsUser);
351 t.start();
352 t.waitForServerOnline();
353 } catch (InterruptedException ie) {
354 throw new IOException("Interrupted adding regionserver to cluster", ie);
355 }
356 return t;
357 }
358
359
360
361
362
363 public String abortRegionServer(int serverNumber) {
364 HRegionServer server = getRegionServer(serverNumber);
365 LOG.info("Aborting " + server.toString());
366 server.abort("Aborting for tests", new Exception("Trace info"));
367 return server.toString();
368 }
369
370
371
372
373
374
375
376 public JVMClusterUtil.RegionServerThread stopRegionServer(int serverNumber) {
377 return stopRegionServer(serverNumber, true);
378 }
379
380
381
382
383
384
385
386
387
388
389
390 public JVMClusterUtil.RegionServerThread stopRegionServer(int serverNumber,
391 final boolean shutdownFS) {
392 JVMClusterUtil.RegionServerThread server =
393 hbaseCluster.getRegionServers().get(serverNumber);
394 LOG.info("Stopping " + server.toString());
395 server.getRegionServer().stop("Stopping rs " + serverNumber);
396 return server;
397 }
398
399
400
401
402
403
404
405 public String waitOnRegionServer(final int serverNumber) {
406 return this.hbaseCluster.waitOnRegionServer(serverNumber);
407 }
408
409
410
411
412
413
414
415
416 public JVMClusterUtil.MasterThread startMaster() throws IOException {
417 Configuration c = HBaseConfiguration.create(conf);
418 User user =
419 HBaseTestingUtility.getDifferentUser(c, ".hfs."+index++);
420
421 JVMClusterUtil.MasterThread t = null;
422 try {
423 t = hbaseCluster.addMaster(c, hbaseCluster.getMasters().size(), user);
424 t.start();
425 } catch (InterruptedException ie) {
426 throw new IOException("Interrupted adding master to cluster", ie);
427 }
428 return t;
429 }
430
431
432
433
434
435 public MasterService.BlockingInterface getMasterAdminService() {
436 return this.hbaseCluster.getActiveMaster().getMasterRpcServices();
437 }
438
439
440
441
442
443 public HMaster getMaster() {
444 return this.hbaseCluster.getActiveMaster();
445 }
446
447
448
449
450
451 public MasterThread getMasterThread() {
452 for (MasterThread mt: hbaseCluster.getLiveMasters()) {
453 if (mt.getMaster().isActiveMaster()) {
454 return mt;
455 }
456 }
457 return null;
458 }
459
460
461
462
463
464 public HMaster getMaster(final int serverNumber) {
465 return this.hbaseCluster.getMaster(serverNumber);
466 }
467
468
469
470
471
472 public String abortMaster(int serverNumber) {
473 HMaster server = getMaster(serverNumber);
474 LOG.info("Aborting " + server.toString());
475 server.abort("Aborting for tests", new Exception("Trace info"));
476 return server.toString();
477 }
478
479
480
481
482
483
484
485 public JVMClusterUtil.MasterThread stopMaster(int serverNumber) {
486 return stopMaster(serverNumber, true);
487 }
488
489
490
491
492
493
494
495
496
497
498
499 public JVMClusterUtil.MasterThread stopMaster(int serverNumber,
500 final boolean shutdownFS) {
501 JVMClusterUtil.MasterThread server =
502 hbaseCluster.getMasters().get(serverNumber);
503 LOG.info("Stopping " + server.toString());
504 server.getMaster().stop("Stopping master " + serverNumber);
505 return server;
506 }
507
508
509
510
511
512
513
514 public String waitOnMaster(final int serverNumber) {
515 return this.hbaseCluster.waitOnMaster(serverNumber);
516 }
517
518
519
520
521
522
523
524
525
526 public boolean waitForActiveAndReadyMaster(long timeout) throws IOException {
527 List<JVMClusterUtil.MasterThread> mts;
528 long start = System.currentTimeMillis();
529 while (!(mts = getMasterThreads()).isEmpty()
530 && (System.currentTimeMillis() - start) < timeout) {
531 for (JVMClusterUtil.MasterThread mt : mts) {
532 if (mt.getMaster().isActiveMaster() && mt.getMaster().isInitialized()) {
533 return true;
534 }
535 }
536
537 Threads.sleep(100);
538 }
539 return false;
540 }
541
542
543
544
545 public List<JVMClusterUtil.MasterThread> getMasterThreads() {
546 return this.hbaseCluster.getMasters();
547 }
548
549
550
551
552 public List<JVMClusterUtil.MasterThread> getLiveMasterThreads() {
553 return this.hbaseCluster.getLiveMasters();
554 }
555
556
557
558
559 public void join() {
560 this.hbaseCluster.join();
561 }
562
563
564
565
566
567 @SuppressWarnings("deprecation")
568 public void shutdown() throws IOException {
569 if (this.hbaseCluster != null) {
570 this.hbaseCluster.shutdown();
571 }
572 }
573
574 @Override
575 public void close() throws IOException {
576 }
577
578 @Override
579 public ClusterStatus getClusterStatus() throws IOException {
580 HMaster master = getMaster();
581 return master == null ? null : master.getClusterStatus();
582 }
583
584
585
586
587
588 public void flushcache() throws IOException {
589 for (JVMClusterUtil.RegionServerThread t:
590 this.hbaseCluster.getRegionServers()) {
591 for(Region r: t.getRegionServer().getOnlineRegionsLocalContext()) {
592 r.flush(true);
593 }
594 }
595 }
596
597
598
599
600
601 public void flushcache(TableName tableName) throws IOException {
602 for (JVMClusterUtil.RegionServerThread t:
603 this.hbaseCluster.getRegionServers()) {
604 for(Region r: t.getRegionServer().getOnlineRegionsLocalContext()) {
605 if(r.getTableDesc().getTableName().equals(tableName)) {
606 r.flush(true);
607 }
608 }
609 }
610 }
611
612
613
614
615
616 public void compact(boolean major) throws IOException {
617 for (JVMClusterUtil.RegionServerThread t:
618 this.hbaseCluster.getRegionServers()) {
619 for(Region r: t.getRegionServer().getOnlineRegionsLocalContext()) {
620 r.compact(major);
621 }
622 }
623 }
624
625
626
627
628
629 public void compact(TableName tableName, boolean major) throws IOException {
630 for (JVMClusterUtil.RegionServerThread t:
631 this.hbaseCluster.getRegionServers()) {
632 for(Region r: t.getRegionServer().getOnlineRegionsLocalContext()) {
633 if(r.getTableDesc().getTableName().equals(tableName)) {
634 r.compact(major);
635 }
636 }
637 }
638 }
639
640
641
642
643 public List<JVMClusterUtil.RegionServerThread> getRegionServerThreads() {
644 return this.hbaseCluster.getRegionServers();
645 }
646
647
648
649
650 public List<JVMClusterUtil.RegionServerThread> getLiveRegionServerThreads() {
651 return this.hbaseCluster.getLiveRegionServers();
652 }
653
654
655
656
657
658
659 public HRegionServer getRegionServer(int serverNumber) {
660 return hbaseCluster.getRegionServer(serverNumber);
661 }
662
663 public List<HRegion> getRegions(byte[] tableName) {
664 return getRegions(TableName.valueOf(tableName));
665 }
666
667 public List<HRegion> getRegions(TableName tableName) {
668 List<HRegion> ret = new ArrayList<HRegion>();
669 for (JVMClusterUtil.RegionServerThread rst : getRegionServerThreads()) {
670 HRegionServer hrs = rst.getRegionServer();
671 for (Region region : hrs.getOnlineRegionsLocalContext()) {
672 if (region.getTableDesc().getTableName().equals(tableName)) {
673 ret.add((HRegion)region);
674 }
675 }
676 }
677 return ret;
678 }
679
680
681
682
683
684 public int getServerWithMeta() {
685 return getServerWith(HRegionInfo.FIRST_META_REGIONINFO.getRegionName());
686 }
687
688
689
690
691
692
693
694 public int getServerWith(byte[] regionName) {
695 int index = -1;
696 int count = 0;
697 for (JVMClusterUtil.RegionServerThread rst: getRegionServerThreads()) {
698 HRegionServer hrs = rst.getRegionServer();
699 if (!hrs.isStopped()) {
700 Region region = hrs.getOnlineRegion(regionName);
701 if (region != null) {
702 index = count;
703 break;
704 }
705 }
706 count++;
707 }
708 return index;
709 }
710
711 @Override
712 public ServerName getServerHoldingRegion(final TableName tn, byte[] regionName)
713 throws IOException {
714
715
716
717
718 HMaster master = getMaster();
719 Region region = master.getOnlineRegion(regionName);
720 if (region != null) {
721 return master.getServerName();
722 }
723 int index = getServerWith(regionName);
724 if (index < 0) {
725 return null;
726 }
727 return getRegionServer(index).getServerName();
728 }
729
730
731
732
733
734
735
736 public long countServedRegions() {
737 long count = 0;
738 for (JVMClusterUtil.RegionServerThread rst : getLiveRegionServerThreads()) {
739 count += rst.getRegionServer().getNumberOfOnlineRegions();
740 }
741 for (JVMClusterUtil.MasterThread mt : getLiveMasterThreads()) {
742 count += mt.getMaster().getNumberOfOnlineRegions();
743 }
744 return count;
745 }
746
747
748
749
750
751 public void killAll() {
752 for (RegionServerThread rst : getRegionServerThreads()) {
753 rst.getRegionServer().abort("killAll");
754 }
755 for (MasterThread masterThread : getMasterThreads()) {
756 masterThread.getMaster().abort("killAll", new Throwable());
757 }
758 }
759
760 @Override
761 public void waitUntilShutDown() {
762 this.hbaseCluster.join();
763 }
764
765 public List<HRegion> findRegionsForTable(TableName tableName) {
766 ArrayList<HRegion> ret = new ArrayList<HRegion>();
767 for (JVMClusterUtil.RegionServerThread rst : getRegionServerThreads()) {
768 HRegionServer hrs = rst.getRegionServer();
769 for (Region region : hrs.getOnlineRegions(tableName)) {
770 if (region.getTableDesc().getTableName().equals(tableName)) {
771 ret.add((HRegion)region);
772 }
773 }
774 }
775 return ret;
776 }
777
778
779 protected int getRegionServerIndex(ServerName serverName) {
780
781 List<RegionServerThread> servers = getRegionServerThreads();
782 for (int i=0; i < servers.size(); i++) {
783 if (servers.get(i).getRegionServer().getServerName().equals(serverName)) {
784 return i;
785 }
786 }
787 return -1;
788 }
789
790 protected int getMasterIndex(ServerName serverName) {
791 List<MasterThread> masters = getMasterThreads();
792 for (int i = 0; i < masters.size(); i++) {
793 if (masters.get(i).getMaster().getServerName().equals(serverName)) {
794 return i;
795 }
796 }
797 return -1;
798 }
799
800 @Override
801 public AdminService.BlockingInterface getAdminProtocol(ServerName serverName) throws IOException {
802 return getRegionServer(getRegionServerIndex(serverName)).getRSRpcServices();
803 }
804
805 @Override
806 public ClientService.BlockingInterface getClientProtocol(ServerName serverName)
807 throws IOException {
808 return getRegionServer(getRegionServerIndex(serverName)).getRSRpcServices();
809 }
810 }