View Javadoc

1   /**
2    *
3    * Licensed to the Apache Software Foundation (ASF) under one
4    * or more contributor license agreements.  See the NOTICE file
5    * distributed with this work for additional information
6    * regarding copyright ownership.  The ASF licenses this file
7    * to you under the Apache License, Version 2.0 (the
8    * "License"); you may not use this file except in compliance
9    * with the License.  You may obtain a copy of the License at
10   *
11   *     http://www.apache.org/licenses/LICENSE-2.0
12   *
13   * Unless required by applicable law or agreed to in writing, software
14   * distributed under the License is distributed on an "AS IS" BASIS,
15   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
16   * See the License for the specific language governing permissions and
17   * limitations under the License.
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   * This class creates a single process HBase cluster. One thread is created for
45   * a master and one per region server.
46   *
47   * Call {@link #startup()} to start the cluster running and {@link #shutdown()}
48   * to close it all down. {@link #join} the cluster is you want to wait on
49   * shutdown completion.
50   *
51   * <p>Runs master on port 16000 by default.  Because we can't just kill the
52   * process -- not till HADOOP-1700 gets fixed and even then.... -- we need to
53   * be able to find the master with a remote client to run shutdown.  To use a
54   * port other than 16000, set the hbase.master to a value of 'local:PORT':
55   * that is 'local', not 'localhost', and the port number the master should use
56   * instead of 16000.
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    /** local mode */
69    public static final String LOCAL = "local";
70    /** 'local:' */
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     * Constructor.
80     * @param conf
81     * @throws IOException
82     */
83    public LocalHBaseCluster(final Configuration conf)
84    throws IOException {
85      this(conf, DEFAULT_NO);
86    }
87  
88    /**
89     * Constructor.
90     * @param conf Configuration to use.  Post construction has the master's
91     * address.
92     * @param noRegionServers Count of regionservers to start.
93     * @throws IOException
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    * Constructor.
103    * @param conf Configuration to use.  Post construction has the active master
104    * address.
105    * @param noMasters Count of masters to start.
106    * @param noRegionServers Count of regionservers to start.
107    * @throws IOException
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    * Constructor.
130    * @param conf Configuration to use.  Post construction has the master's
131    * address.
132    * @param noMasters Count of masters to start.
133    * @param noRegionServers Count of regionservers to start.
134    * @param masterClass
135    * @param regionServerClass
136    * @throws IOException
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     // Always have masters and regionservers come up on port '0' so we don't
146     // clash over default ports.
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     // Start the HMasters.
161     for (int i = 0; i < noMasters; i++) {
162       addMaster(new Configuration(conf), i);
163     }
164     // Start the HRegionServers.
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     // Create each regionserver with its own Configuration instance so each has
184     // its HConnection instance rather than share (see HBASE_INSTANCES down in
185     // the guts of HConnectionManager.
186 
187     // Also, create separate CoordinatedStateManager instance per Server.
188     // This is special case when we have to have more than 1 CoordinatedStateManager
189     // within 1 process.
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     // Create each master with its own Configuration instance so each has
219     // its HConnection instance rather than share (see HBASE_INSTANCES down in
220     // the guts of HConnectionManager.
221 
222     // Also, create separate CoordinatedStateManager instance per Server.
223     // This is special case when we have to have more than 1 CoordinatedStateManager
224     // within 1 process.
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    * @param serverNumber
247    * @return region server
248    */
249   public HRegionServer getRegionServer(int serverNumber) {
250     return regionThreads.get(serverNumber).getRegionServer();
251   }
252 
253   /**
254    * @return Read-only list of region server threads.
255    */
256   public List<JVMClusterUtil.RegionServerThread> getRegionServers() {
257     return Collections.unmodifiableList(this.regionThreads);
258   }
259 
260   /**
261    * @return List of running servers (Some servers may have been killed or
262    * aborted during lifetime of cluster; these servers are not included in this
263    * list).
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    * @return the Configuration used by this LocalHBaseCluster
278    */
279   public Configuration getConfiguration() {
280     return this.conf;
281   }
282 
283   /**
284    * Wait for the specified region server to stop. Removes this thread from list of running threads.
285    * @return Name of region server that just went down.
286    */
287   public String waitOnRegionServer(int serverNumber) {
288     JVMClusterUtil.RegionServerThread regionServerThread = this.regionThreads.get(serverNumber);
289     return waitOnRegionServer(regionServerThread);
290   }
291 
292   /**
293    * Wait for the specified region server to stop. Removes this thread from list of running threads.
294    * @return Name of region server that just went down.
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    * @return the HMaster thread
311    */
312   public HMaster getMaster(int serverNumber) {
313     return masterThreads.get(serverNumber).getMaster();
314   }
315 
316   /**
317    * Gets the current active master, if available.  If no active master, returns
318    * null.
319    * @return the HMaster for the active master
320    */
321   public HMaster getActiveMaster() {
322     for (JVMClusterUtil.MasterThread mt : masterThreads) {
323       // Ensure that the current active master is not stopped.
324       // We don't want to return a stopping master as an active master.
325       if (mt.getMaster().isActiveMaster() && !mt.getMaster().isStopped()) {
326         return mt.getMaster();
327       }
328     }
329     return null;
330   }
331 
332   /**
333    * @return Read-only list of master threads.
334    */
335   public List<JVMClusterUtil.MasterThread> getMasters() {
336     return Collections.unmodifiableList(this.masterThreads);
337   }
338 
339   /**
340    * @return List of running master servers (Some servers may have been killed
341    * or aborted during lifetime of cluster; these servers are not included in
342    * this list).
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    * Wait for the specified master to stop. Removes this thread from list of running threads.
357    * @return Name of master that just went down.
358    */
359   public String waitOnMaster(int serverNumber) {
360     JVMClusterUtil.MasterThread masterThread = this.masterThreads.get(serverNumber);
361     return waitOnMaster(masterThread);
362   }
363 
364   /**
365    * Wait for the specified master to stop. Removes this thread from list of running threads.
366    * @return Name of master that just went down.
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    * Wait for Mini HBase Cluster to shut down.
383    * Presumes you've already called {@link #shutdown()}.
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    * Start the cluster.
412    */
413   public void startup() throws IOException {
414     JVMClusterUtil.startup(this.masterThreads, this.regionThreads);
415   }
416 
417   /**
418    * Shut down the mini HBase cluster
419    */
420   public void shutdown() {
421     JVMClusterUtil.shutdown(this.masterThreads, this.regionThreads);
422   }
423 
424   /**
425    * @param c Configuration to check.
426    * @return True if a 'local' address in hbase.master value.
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    * Test things basically work.
435    * @param args
436    * @throws IOException
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 }