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.PrivilegedExceptionAction;
23 import java.util.ArrayList;
24 import java.util.Collections;
25 import java.util.List;
26
27 import org.apache.commons.logging.Log;
28 import org.apache.commons.logging.LogFactory;
29 import org.apache.hadoop.hbase.classification.InterfaceAudience;
30 import org.apache.hadoop.hbase.classification.InterfaceStability;
31 import org.apache.hadoop.conf.Configuration;
32 import org.apache.hadoop.hbase.client.Admin;
33 import org.apache.hadoop.hbase.client.HBaseAdmin;
34 import org.apache.hadoop.hbase.regionserver.HRegionServer;
35 import org.apache.hadoop.hbase.security.User;
36 import org.apache.hadoop.hbase.util.JVMClusterUtil.RegionServerThread;
37 import org.apache.hadoop.hbase.util.Threads;
38
39 import java.util.concurrent.CopyOnWriteArrayList;
40 import org.apache.hadoop.hbase.master.HMaster;
41 import org.apache.hadoop.hbase.util.JVMClusterUtil;
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59 @InterfaceAudience.Public
60 @InterfaceStability.Evolving
61 public class LocalHBaseCluster {
62 private static final Log LOG = LogFactory.getLog(LocalHBaseCluster.class);
63 private final List<JVMClusterUtil.MasterThread> masterThreads =
64 new CopyOnWriteArrayList<JVMClusterUtil.MasterThread>();
65 private final List<JVMClusterUtil.RegionServerThread> regionThreads =
66 new CopyOnWriteArrayList<JVMClusterUtil.RegionServerThread>();
67 private final static int DEFAULT_NO = 1;
68
69 public static final String LOCAL = "local";
70
71 public static final String LOCAL_COLON = LOCAL + ":";
72 public static final String ASSIGN_RANDOM_PORTS = "hbase.localcluster.assign.random.ports";
73
74 private final Configuration conf;
75 private final Class<? extends HMaster> masterClass;
76 private final Class<? extends HRegionServer> regionServerClass;
77
78
79
80
81
82
83 public LocalHBaseCluster(final Configuration conf)
84 throws IOException {
85 this(conf, DEFAULT_NO);
86 }
87
88
89
90
91
92
93
94
95 public LocalHBaseCluster(final Configuration conf, final int noRegionServers)
96 throws IOException {
97 this(conf, 1, noRegionServers, getMasterImplementation(conf),
98 getRegionServerImplementation(conf));
99 }
100
101
102
103
104
105
106
107
108
109 public LocalHBaseCluster(final Configuration conf, final int noMasters,
110 final int noRegionServers)
111 throws IOException {
112 this(conf, noMasters, noRegionServers, getMasterImplementation(conf),
113 getRegionServerImplementation(conf));
114 }
115
116 @SuppressWarnings("unchecked")
117 private static Class<? extends HRegionServer> getRegionServerImplementation(final Configuration conf) {
118 return (Class<? extends HRegionServer>)conf.getClass(HConstants.REGION_SERVER_IMPL,
119 HRegionServer.class);
120 }
121
122 @SuppressWarnings("unchecked")
123 private static Class<? extends HMaster> getMasterImplementation(final Configuration conf) {
124 return (Class<? extends HMaster>)conf.getClass(HConstants.MASTER_IMPL,
125 HMaster.class);
126 }
127
128
129
130
131
132
133
134
135
136
137
138 @SuppressWarnings("unchecked")
139 public LocalHBaseCluster(final Configuration conf, final int noMasters,
140 final int noRegionServers, final Class<? extends HMaster> masterClass,
141 final Class<? extends HRegionServer> regionServerClass)
142 throws IOException {
143 this.conf = conf;
144
145
146
147 if (conf.getBoolean(ASSIGN_RANDOM_PORTS, true)) {
148 conf.set(HConstants.MASTER_PORT, "0");
149 conf.set(HConstants.REGIONSERVER_PORT, "0");
150 if (conf.getInt(HConstants.REGIONSERVER_INFO_PORT, 0) != -1) {
151 conf.set(HConstants.REGIONSERVER_INFO_PORT, "0");
152 }
153 if (conf.getInt(HConstants.MASTER_INFO_PORT, 0) != -1) {
154 conf.set(HConstants.MASTER_INFO_PORT, "0");
155 }
156 }
157
158 this.masterClass = (Class<? extends HMaster>)
159 conf.getClass(HConstants.MASTER_IMPL, masterClass);
160
161 for (int i = 0; i < noMasters; i++) {
162 addMaster(new Configuration(conf), i);
163 }
164
165 this.regionServerClass =
166 (Class<? extends HRegionServer>)conf.getClass(HConstants.REGION_SERVER_IMPL,
167 regionServerClass);
168
169 for (int i = 0; i < noRegionServers; i++) {
170 addRegionServer(new Configuration(conf), i);
171 }
172 }
173
174 public JVMClusterUtil.RegionServerThread addRegionServer()
175 throws IOException {
176 return addRegionServer(new Configuration(conf), this.regionThreads.size());
177 }
178
179 @SuppressWarnings("unchecked")
180 public JVMClusterUtil.RegionServerThread addRegionServer(
181 Configuration config, final int index)
182 throws IOException {
183
184
185
186
187
188
189
190 CoordinatedStateManager cp = CoordinatedStateManagerFactory.getCoordinatedStateManager(conf);
191
192 JVMClusterUtil.RegionServerThread rst =
193 JVMClusterUtil.createRegionServerThread(config, cp, (Class<? extends HRegionServer>) conf
194 .getClass(HConstants.REGION_SERVER_IMPL, this.regionServerClass), index);
195
196 this.regionThreads.add(rst);
197 return rst;
198 }
199
200 public JVMClusterUtil.RegionServerThread addRegionServer(
201 final Configuration config, final int index, User user)
202 throws IOException, InterruptedException {
203 return user.runAs(
204 new PrivilegedExceptionAction<JVMClusterUtil.RegionServerThread>() {
205 @Override
206 public JVMClusterUtil.RegionServerThread run() throws Exception {
207 return addRegionServer(config, index);
208 }
209 });
210 }
211
212 public JVMClusterUtil.MasterThread addMaster() throws IOException {
213 return addMaster(new Configuration(conf), this.masterThreads.size());
214 }
215
216 public JVMClusterUtil.MasterThread addMaster(Configuration c, final int index)
217 throws IOException {
218
219
220
221
222
223
224
225 CoordinatedStateManager cp = CoordinatedStateManagerFactory.getCoordinatedStateManager(conf);
226
227 JVMClusterUtil.MasterThread mt = JVMClusterUtil.createMasterThread(c, cp,
228 (Class<? extends HMaster>) conf.getClass(HConstants.MASTER_IMPL, this.masterClass), index);
229 this.masterThreads.add(mt);
230 return mt;
231 }
232
233 public JVMClusterUtil.MasterThread addMaster(
234 final Configuration c, final int index, User user)
235 throws IOException, InterruptedException {
236 return user.runAs(
237 new PrivilegedExceptionAction<JVMClusterUtil.MasterThread>() {
238 @Override
239 public JVMClusterUtil.MasterThread run() throws Exception {
240 return addMaster(c, index);
241 }
242 });
243 }
244
245
246
247
248
249 public HRegionServer getRegionServer(int serverNumber) {
250 return regionThreads.get(serverNumber).getRegionServer();
251 }
252
253
254
255
256 public List<JVMClusterUtil.RegionServerThread> getRegionServers() {
257 return Collections.unmodifiableList(this.regionThreads);
258 }
259
260
261
262
263
264
265 public List<JVMClusterUtil.RegionServerThread> getLiveRegionServers() {
266 List<JVMClusterUtil.RegionServerThread> liveServers =
267 new ArrayList<JVMClusterUtil.RegionServerThread>();
268 List<RegionServerThread> list = getRegionServers();
269 for (JVMClusterUtil.RegionServerThread rst: list) {
270 if (rst.isAlive()) liveServers.add(rst);
271 else LOG.info("Not alive " + rst.getName());
272 }
273 return liveServers;
274 }
275
276
277
278
279 public Configuration getConfiguration() {
280 return this.conf;
281 }
282
283
284
285
286
287 public String waitOnRegionServer(int serverNumber) {
288 JVMClusterUtil.RegionServerThread regionServerThread = this.regionThreads.get(serverNumber);
289 return waitOnRegionServer(regionServerThread);
290 }
291
292
293
294
295
296 public String waitOnRegionServer(JVMClusterUtil.RegionServerThread rst) {
297 while (rst.isAlive()) {
298 try {
299 LOG.info("Waiting on " + rst.getRegionServer().toString());
300 rst.join();
301 } catch (InterruptedException e) {
302 e.printStackTrace();
303 }
304 }
305 regionThreads.remove(rst);
306 return rst.getName();
307 }
308
309
310
311
312 public HMaster getMaster(int serverNumber) {
313 return masterThreads.get(serverNumber).getMaster();
314 }
315
316
317
318
319
320
321 public HMaster getActiveMaster() {
322 for (JVMClusterUtil.MasterThread mt : masterThreads) {
323
324
325 if (mt.getMaster().isActiveMaster() && !mt.getMaster().isStopped()) {
326 return mt.getMaster();
327 }
328 }
329 return null;
330 }
331
332
333
334
335 public List<JVMClusterUtil.MasterThread> getMasters() {
336 return Collections.unmodifiableList(this.masterThreads);
337 }
338
339
340
341
342
343
344 public List<JVMClusterUtil.MasterThread> getLiveMasters() {
345 List<JVMClusterUtil.MasterThread> liveServers = new ArrayList<>();
346 List<JVMClusterUtil.MasterThread> list = getMasters();
347 for (JVMClusterUtil.MasterThread mt: list) {
348 if (mt.isAlive()) {
349 liveServers.add(mt);
350 }
351 }
352 return liveServers;
353 }
354
355
356
357
358
359 public String waitOnMaster(int serverNumber) {
360 JVMClusterUtil.MasterThread masterThread = this.masterThreads.get(serverNumber);
361 return waitOnMaster(masterThread);
362 }
363
364
365
366
367
368 public String waitOnMaster(JVMClusterUtil.MasterThread masterThread) {
369 while (masterThread.isAlive()) {
370 try {
371 LOG.info("Waiting on " + masterThread.getMaster().getServerName().toString());
372 masterThread.join();
373 } catch (InterruptedException e) {
374 e.printStackTrace();
375 }
376 }
377 masterThreads.remove(masterThread);
378 return masterThread.getName();
379 }
380
381
382
383
384
385 public void join() {
386 if (this.regionThreads != null) {
387 for(Thread t: this.regionThreads) {
388 if (t.isAlive()) {
389 try {
390 Threads.threadDumpingIsAlive(t);
391 } catch (InterruptedException e) {
392 LOG.debug("Interrupted", e);
393 }
394 }
395 }
396 }
397 if (this.masterThreads != null) {
398 for (Thread t : this.masterThreads) {
399 if (t.isAlive()) {
400 try {
401 Threads.threadDumpingIsAlive(t);
402 } catch (InterruptedException e) {
403 LOG.debug("Interrupted", e);
404 }
405 }
406 }
407 }
408 }
409
410
411
412
413 public void startup() throws IOException {
414 JVMClusterUtil.startup(this.masterThreads, this.regionThreads);
415 }
416
417
418
419
420 public void shutdown() {
421 JVMClusterUtil.shutdown(this.masterThreads, this.regionThreads);
422 }
423
424
425
426
427
428 public static boolean isLocal(final Configuration c) {
429 boolean mode = c.getBoolean(HConstants.CLUSTER_DISTRIBUTED, HConstants.DEFAULT_CLUSTER_DISTRIBUTED);
430 return(mode == HConstants.CLUSTER_IS_LOCAL);
431 }
432
433
434
435
436
437
438 public static void main(String[] args) throws IOException {
439 Configuration conf = HBaseConfiguration.create();
440 LocalHBaseCluster cluster = new LocalHBaseCluster(conf);
441 cluster.startup();
442 Admin admin = new HBaseAdmin(conf);
443 try {
444 HTableDescriptor htd =
445 new HTableDescriptor(TableName.valueOf(cluster.getClass().getName()));
446 admin.createTable(htd);
447 } finally {
448 admin.close();
449 }
450 cluster.shutdown();
451 }
452 }