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