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.master;
20  
21  import java.io.IOException;
22  import java.io.InterruptedIOException;
23  import java.lang.reflect.Constructor;
24  import java.lang.reflect.InvocationTargetException;
25  import java.net.InetAddress;
26  import java.net.InetSocketAddress;
27  import java.net.UnknownHostException;
28  import java.util.ArrayList;
29  import java.util.Collection;
30  import java.util.Collections;
31  import java.util.Comparator;
32  import java.util.HashSet;
33  import java.util.Iterator;
34  import java.util.List;
35  import java.util.Map;
36  import java.util.Set;
37  import java.util.concurrent.TimeUnit;
38  import java.util.concurrent.atomic.AtomicReference;
39  import java.util.regex.Pattern;
40  
41  import javax.servlet.ServletException;
42  import javax.servlet.http.HttpServlet;
43  import javax.servlet.http.HttpServletRequest;
44  import javax.servlet.http.HttpServletResponse;
45  
46  import org.apache.commons.logging.Log;
47  import org.apache.commons.logging.LogFactory;
48  import org.apache.hadoop.conf.Configuration;
49  import org.apache.hadoop.fs.Path;
50  import org.apache.hadoop.hbase.ClusterStatus;
51  import org.apache.hadoop.hbase.CoordinatedStateException;
52  import org.apache.hadoop.hbase.CoordinatedStateManager;
53  import org.apache.hadoop.hbase.DoNotRetryIOException;
54  import org.apache.hadoop.hbase.HBaseIOException;
55  import org.apache.hadoop.hbase.HBaseInterfaceAudience;
56  import org.apache.hadoop.hbase.HColumnDescriptor;
57  import org.apache.hadoop.hbase.HConstants;
58  import org.apache.hadoop.hbase.HRegionInfo;
59  import org.apache.hadoop.hbase.HTableDescriptor;
60  import org.apache.hadoop.hbase.MasterNotRunningException;
61  import org.apache.hadoop.hbase.MetaMigrationConvertingToPB;
62  import org.apache.hadoop.hbase.MetaTableAccessor;
63  import org.apache.hadoop.hbase.NamespaceDescriptor;
64  import org.apache.hadoop.hbase.NamespaceNotFoundException;
65  import org.apache.hadoop.hbase.PleaseHoldException;
66  import org.apache.hadoop.hbase.ProcedureInfo;
67  import org.apache.hadoop.hbase.RegionStateListener;
68  import org.apache.hadoop.hbase.ScheduledChore;
69  import org.apache.hadoop.hbase.Server;
70  import org.apache.hadoop.hbase.ServerLoad;
71  import org.apache.hadoop.hbase.ServerName;
72  import org.apache.hadoop.hbase.TableDescriptors;
73  import org.apache.hadoop.hbase.TableName;
74  import org.apache.hadoop.hbase.TableNotDisabledException;
75  import org.apache.hadoop.hbase.TableNotFoundException;
76  import org.apache.hadoop.hbase.UnknownRegionException;
77  import org.apache.hadoop.hbase.classification.InterfaceAudience;
78  import org.apache.hadoop.hbase.client.MetaScanner;
79  import org.apache.hadoop.hbase.client.MetaScanner.MetaScannerVisitor;
80  import org.apache.hadoop.hbase.client.MetaScanner.MetaScannerVisitorBase;
81  import org.apache.hadoop.hbase.client.RegionReplicaUtil;
82  import org.apache.hadoop.hbase.client.Result;
83  import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
84  import org.apache.hadoop.hbase.exceptions.DeserializationException;
85  import org.apache.hadoop.hbase.executor.ExecutorType;
86  import org.apache.hadoop.hbase.ipc.RpcServer;
87  import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
88  import org.apache.hadoop.hbase.master.MasterRpcServices.BalanceSwitchMode;
89  import org.apache.hadoop.hbase.master.RegionState.State;
90  import org.apache.hadoop.hbase.master.balancer.BalancerChore;
91  import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer;
92  import org.apache.hadoop.hbase.master.balancer.ClusterStatusChore;
93  import org.apache.hadoop.hbase.master.balancer.LoadBalancerFactory;
94  import org.apache.hadoop.hbase.master.cleaner.HFileCleaner;
95  import org.apache.hadoop.hbase.master.cleaner.LogCleaner;
96  import org.apache.hadoop.hbase.master.cleaner.ReplicationZKLockCleanerChore;
97  import org.apache.hadoop.hbase.master.handler.DispatchMergingRegionHandler;
98  import org.apache.hadoop.hbase.master.normalizer.RegionNormalizer;
99  import org.apache.hadoop.hbase.master.normalizer.RegionNormalizerChore;
100 import org.apache.hadoop.hbase.master.normalizer.RegionNormalizerFactory;
101 import org.apache.hadoop.hbase.master.procedure.AddColumnFamilyProcedure;
102 import org.apache.hadoop.hbase.master.procedure.CreateTableProcedure;
103 import org.apache.hadoop.hbase.master.procedure.DeleteColumnFamilyProcedure;
104 import org.apache.hadoop.hbase.master.procedure.DeleteTableProcedure;
105 import org.apache.hadoop.hbase.master.procedure.DisableTableProcedure;
106 import org.apache.hadoop.hbase.master.procedure.EnableTableProcedure;
107 import org.apache.hadoop.hbase.master.procedure.MasterProcedureConstants;
108 import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
109 import org.apache.hadoop.hbase.master.procedure.MasterProcedureScheduler.ProcedureEvent;
110 import org.apache.hadoop.hbase.master.procedure.ModifyColumnFamilyProcedure;
111 import org.apache.hadoop.hbase.master.procedure.ModifyTableProcedure;
112 import org.apache.hadoop.hbase.master.procedure.ProcedurePrepareLatch;
113 import org.apache.hadoop.hbase.master.procedure.ProcedureSyncWait;
114 import org.apache.hadoop.hbase.master.procedure.TruncateTableProcedure;
115 import org.apache.hadoop.hbase.master.snapshot.SnapshotManager;
116 import org.apache.hadoop.hbase.monitoring.MemoryBoundedLogMessageBuffer;
117 import org.apache.hadoop.hbase.monitoring.MonitoredTask;
118 import org.apache.hadoop.hbase.monitoring.TaskMonitor;
119 import org.apache.hadoop.hbase.master.normalizer.NormalizationPlan;
120 import org.apache.hadoop.hbase.procedure.MasterProcedureManagerHost;
121 import org.apache.hadoop.hbase.procedure.flush.MasterFlushTableProcedureManager;
122 import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
123 import org.apache.hadoop.hbase.procedure2.store.wal.WALProcedureStore;
124 import org.apache.hadoop.hbase.protobuf.generated.HBaseProtos.RegionServerInfo;
125 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos;
126 import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
127 import org.apache.hadoop.hbase.quotas.MasterQuotaManager;
128 import org.apache.hadoop.hbase.regionserver.DefaultStoreEngine;
129 import org.apache.hadoop.hbase.regionserver.HRegionServer;
130 import org.apache.hadoop.hbase.regionserver.HStore;
131 import org.apache.hadoop.hbase.regionserver.RSRpcServices;
132 import org.apache.hadoop.hbase.regionserver.RegionCoprocessorHost;
133 import org.apache.hadoop.hbase.regionserver.RegionSplitPolicy;
134 import org.apache.hadoop.hbase.regionserver.compactions.ExploringCompactionPolicy;
135 import org.apache.hadoop.hbase.regionserver.compactions.FIFOCompactionPolicy;
136 import org.apache.hadoop.hbase.replication.regionserver.Replication;
137 import org.apache.hadoop.hbase.security.UserProvider;
138 import org.apache.hadoop.hbase.util.Addressing;
139 import org.apache.hadoop.hbase.util.Bytes;
140 import org.apache.hadoop.hbase.util.CompressionTest;
141 import org.apache.hadoop.hbase.util.ConfigUtil;
142 import org.apache.hadoop.hbase.util.EncryptionTest;
143 import org.apache.hadoop.hbase.util.FSUtils;
144 import org.apache.hadoop.hbase.util.HFileArchiveUtil;
145 import org.apache.hadoop.hbase.util.HasThread;
146 import org.apache.hadoop.hbase.util.ModifyRegionUtils;
147 import org.apache.hadoop.hbase.util.Pair;
148 import org.apache.hadoop.hbase.util.Threads;
149 import org.apache.hadoop.hbase.util.VersionInfo;
150 import org.apache.hadoop.hbase.zookeeper.DrainingServerTracker;
151 import org.apache.hadoop.hbase.zookeeper.LoadBalancerTracker;
152 import org.apache.hadoop.hbase.zookeeper.MasterAddressTracker;
153 import org.apache.hadoop.hbase.zookeeper.MetaTableLocator;
154 import org.apache.hadoop.hbase.zookeeper.RegionNormalizerTracker;
155 import org.apache.hadoop.hbase.zookeeper.RegionServerTracker;
156 import org.apache.hadoop.hbase.zookeeper.ZKClusterId;
157 import org.apache.hadoop.hbase.zookeeper.ZKUtil;
158 import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
159 import org.apache.zookeeper.KeeperException;
160 import org.mortbay.jetty.Connector;
161 import org.mortbay.jetty.nio.SelectChannelConnector;
162 import org.mortbay.jetty.servlet.Context;
163 
164 import com.google.common.annotations.VisibleForTesting;
165 import com.google.common.collect.Maps;
166 import com.google.protobuf.Descriptors;
167 import com.google.protobuf.Service;
168 
169 /**
170  * HMaster is the "master server" for HBase. An HBase cluster has one active
171  * master.  If many masters are started, all compete.  Whichever wins goes on to
172  * run the cluster.  All others park themselves in their constructor until
173  * master or cluster shutdown or until the active master loses its lease in
174  * zookeeper.  Thereafter, all running master jostle to take over master role.
175  *
176  * <p>The Master can be asked shutdown the cluster. See {@link #shutdown()}.  In
177  * this case it will tell all regionservers to go down and then wait on them
178  * all reporting in that they are down.  This master will then shut itself down.
179  *
180  * <p>You can also shutdown just this master.  Call {@link #stopMaster()}.
181  *
182  * @see org.apache.zookeeper.Watcher
183  */
184 @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS)
185 @SuppressWarnings("deprecation")
186 public class HMaster extends HRegionServer implements MasterServices, Server {
187   private static final Log LOG = LogFactory.getLog(HMaster.class.getName());
188 
189   /**
190    * Protection against zombie master. Started once Master accepts active responsibility and
191    * starts taking over responsibilities. Allows a finite time window before giving up ownership.
192    */
193   private static class InitializationMonitor extends HasThread {
194     /** The amount of time in milliseconds to sleep before checking initialization status. */
195     public static final String TIMEOUT_KEY = "hbase.master.initializationmonitor.timeout";
196     public static final long TIMEOUT_DEFAULT = TimeUnit.MILLISECONDS.convert(15, TimeUnit.MINUTES);
197 
198     /**
199      * When timeout expired and initialization has not complete, call {@link System#exit(int)} when
200      * true, do nothing otherwise.
201      */
202     public static final String HALT_KEY = "hbase.master.initializationmonitor.haltontimeout";
203     public static final boolean HALT_DEFAULT = false;
204 
205     private final HMaster master;
206     private final long timeout;
207     private final boolean haltOnTimeout;
208 
209     /** Creates a Thread that monitors the {@link #isInitialized()} state. */
210     InitializationMonitor(HMaster master) {
211       super("MasterInitializationMonitor");
212       this.master = master;
213       this.timeout = master.getConfiguration().getLong(TIMEOUT_KEY, TIMEOUT_DEFAULT);
214       this.haltOnTimeout = master.getConfiguration().getBoolean(HALT_KEY, HALT_DEFAULT);
215       this.setDaemon(true);
216     }
217 
218     @Override
219     public void run() {
220       try {
221         while (!master.isStopped() && master.isActiveMaster()) {
222           Thread.sleep(timeout);
223           if (master.isInitialized()) {
224             LOG.debug("Initialization completed within allotted tolerance. Monitor exiting.");
225           } else {
226             LOG.error("Master failed to complete initialization after " + timeout + "ms. Please"
227                 + " consider submitting a bug report including a thread dump of this process.");
228             if (haltOnTimeout) {
229               LOG.error("Zombie Master exiting. Thread dump to stdout");
230               Threads.printThreadInfo(System.out, "Zombie HMaster");
231               System.exit(-1);
232             }
233           }
234         }
235       } catch (InterruptedException ie) {
236         LOG.trace("InitMonitor thread interrupted. Existing.");
237       }
238     }
239   }
240 
241   // MASTER is name of the webapp and the attribute name used stuffing this
242   //instance into web context.
243   public static final String MASTER = "master";
244 
245   // Manager and zk listener for master election
246   private final ActiveMasterManager activeMasterManager;
247   // Region server tracker
248   RegionServerTracker regionServerTracker;
249   // Draining region server tracker
250   private DrainingServerTracker drainingServerTracker;
251   // Tracker for load balancer state
252   LoadBalancerTracker loadBalancerTracker;
253 
254   // Tracker for region normalizer state
255   private RegionNormalizerTracker regionNormalizerTracker;
256 
257   /** Namespace stuff */
258   private TableNamespaceManager tableNamespaceManager;
259 
260   // Metrics for the HMaster
261   final MetricsMaster metricsMaster;
262   // file system manager for the master FS operations
263   private MasterFileSystem fileSystemManager;
264 
265   // server manager to deal with region server info
266   volatile ServerManager serverManager;
267 
268   // manager of assignment nodes in zookeeper
269   AssignmentManager assignmentManager;
270 
271   // buffer for "fatal error" notices from region servers
272   // in the cluster. This is only used for assisting
273   // operations/debugging.
274   MemoryBoundedLogMessageBuffer rsFatals;
275 
276   // flag set after we become the active master (used for testing)
277   private volatile boolean isActiveMaster = false;
278 
279   // flag set after we complete initialization once active,
280   // it is not private since it's used in unit tests
281   private final ProcedureEvent initialized = new ProcedureEvent("master initialized");
282 
283   // flag set after master services are started,
284   // initialization may have not completed yet.
285   volatile boolean serviceStarted = false;
286 
287   // flag set after we complete assignMeta.
288   private final ProcedureEvent serverCrashProcessingEnabled =
289     new ProcedureEvent("server crash processing");
290 
291   LoadBalancer balancer;
292   private RegionNormalizer normalizer;
293   private BalancerChore balancerChore;
294   private RegionNormalizerChore normalizerChore;
295   private ClusterStatusChore clusterStatusChore;
296   private ClusterStatusPublisher clusterStatusPublisherChore = null;
297   private PeriodicDoMetrics periodicDoMetricsChore = null;
298 
299   CatalogJanitor catalogJanitorChore;
300   private ReplicationZKLockCleanerChore replicationZKLockCleanerChore;
301   private LogCleaner logCleaner;
302   private HFileCleaner hfileCleaner;
303 
304   MasterCoprocessorHost cpHost;
305 
306   private final boolean preLoadTableDescriptors;
307 
308   // Time stamps for when a hmaster became active
309   private long masterActiveTime;
310 
311   //should we check the compression codec type at master side, default true, HBASE-6370
312   private final boolean masterCheckCompression;
313 
314   //should we check encryption settings at master side, default true
315   private final boolean masterCheckEncryption;
316 
317   Map<String, Service> coprocessorServiceHandlers = Maps.newHashMap();
318 
319   // monitor for snapshot of hbase tables
320   SnapshotManager snapshotManager;
321   // monitor for distributed procedures
322   MasterProcedureManagerHost mpmHost;
323 
324   // it is assigned after 'initialized' guard set to true, so should be volatile
325   private volatile MasterQuotaManager quotaManager;
326 
327   private ProcedureExecutor<MasterProcedureEnv> procedureExecutor;
328   private WALProcedureStore procedureStore;
329 
330   /** flag used in test cases in order to simulate RS failures during master initialization */
331   private volatile boolean initializationBeforeMetaAssignment = false;
332 
333   /** jetty server for master to redirect requests to regionserver infoServer */
334   private org.mortbay.jetty.Server masterJettyServer;
335 
336   public static class RedirectServlet extends HttpServlet {
337     private static final long serialVersionUID = 2894774810058302472L;
338     private static int regionServerInfoPort;
339 
340     @Override
341     public void doGet(HttpServletRequest request,
342         HttpServletResponse response) throws ServletException, IOException {
343       String redirectUrl = request.getScheme() + "://"
344         + request.getServerName() + ":" + regionServerInfoPort
345         + request.getRequestURI();
346       response.sendRedirect(redirectUrl);
347     }
348   }
349 
350   private static class PeriodicDoMetrics extends ScheduledChore {
351     private final HMaster server;
352     public PeriodicDoMetrics(int doMetricsInterval, final HMaster server) {
353       super(server.getServerName() + "-DoMetricsChore", server, doMetricsInterval);
354       this.server = server;
355     }
356 
357     @Override
358     protected void chore() {
359       server.doMetrics();
360     }
361   }
362 
363   /**
364    * Initializes the HMaster. The steps are as follows:
365    * <p>
366    * <ol>
367    * <li>Initialize the local HRegionServer
368    * <li>Start the ActiveMasterManager.
369    * </ol>
370    * <p>
371    * Remaining steps of initialization occur in
372    * #finishActiveMasterInitialization(MonitoredTask) after
373    * the master becomes the active one.
374    *
375    * @throws InterruptedException
376    * @throws KeeperException
377    * @throws IOException
378    */
379   public HMaster(final Configuration conf, CoordinatedStateManager csm)
380       throws IOException, KeeperException, InterruptedException {
381     super(conf, csm);
382     this.rsFatals = new MemoryBoundedLogMessageBuffer(
383       conf.getLong("hbase.master.buffer.for.rs.fatals", 1*1024*1024));
384 
385     LOG.info("hbase.rootdir=" + FSUtils.getRootDir(this.conf) +
386       ", hbase.cluster.distributed=" + this.conf.getBoolean(HConstants.CLUSTER_DISTRIBUTED, false));
387 
388     // Disable usage of meta replicas in the master
389     this.conf.setBoolean(HConstants.USE_META_REPLICAS, false);
390 
391     Replication.decorateMasterConfiguration(this.conf);
392 
393     // Hack! Maps DFSClient => Master for logs.  HDFS made this
394     // config param for task trackers, but we can piggyback off of it.
395     if (this.conf.get("mapreduce.task.attempt.id") == null) {
396       this.conf.set("mapreduce.task.attempt.id", "hb_m_" + this.serverName.toString());
397     }
398 
399     // should we check the compression codec type at master side, default true, HBASE-6370
400     this.masterCheckCompression = conf.getBoolean("hbase.master.check.compression", true);
401 
402     // should we check encryption settings at master side, default true
403     this.masterCheckEncryption = conf.getBoolean("hbase.master.check.encryption", true);
404 
405     this.metricsMaster = new MetricsMaster(new MetricsMasterWrapperImpl(this));
406 
407     // preload table descriptor at startup
408     this.preLoadTableDescriptors = conf.getBoolean("hbase.master.preload.tabledescriptors", true);
409 
410     // Do we publish the status?
411 
412     boolean shouldPublish = conf.getBoolean(HConstants.STATUS_PUBLISHED,
413         HConstants.STATUS_PUBLISHED_DEFAULT);
414     Class<? extends ClusterStatusPublisher.Publisher> publisherClass =
415         conf.getClass(ClusterStatusPublisher.STATUS_PUBLISHER_CLASS,
416             ClusterStatusPublisher.DEFAULT_STATUS_PUBLISHER_CLASS,
417             ClusterStatusPublisher.Publisher.class);
418 
419     if (shouldPublish) {
420       if (publisherClass == null) {
421         LOG.warn(HConstants.STATUS_PUBLISHED + " is true, but " +
422             ClusterStatusPublisher.DEFAULT_STATUS_PUBLISHER_CLASS +
423             " is not set - not publishing status");
424       } else {
425         clusterStatusPublisherChore = new ClusterStatusPublisher(this, conf, publisherClass);
426         getChoreService().scheduleChore(clusterStatusPublisherChore);
427       }
428     }
429 
430     // Some unit tests don't need a cluster, so no zookeeper at all
431     if (!conf.getBoolean("hbase.testing.nocluster", false)) {
432       activeMasterManager = new ActiveMasterManager(zooKeeper, this.serverName, this);
433       int infoPort = putUpJettyServer();
434       startActiveMasterManager(infoPort);
435     } else {
436       activeMasterManager = null;
437     }
438   }
439 
440   // return the actual infoPort, -1 means disable info server.
441   private int putUpJettyServer() throws IOException {
442     if (!conf.getBoolean("hbase.master.infoserver.redirect", true)) {
443       return -1;
444     }
445     int infoPort = conf.getInt("hbase.master.info.port.orig",
446       HConstants.DEFAULT_MASTER_INFOPORT);
447     // -1 is for disabling info server, so no redirecting
448     if (infoPort < 0 || infoServer == null) {
449       return -1;
450     }
451     String addr = conf.get("hbase.master.info.bindAddress", "0.0.0.0");
452     if (!Addressing.isLocalAddress(InetAddress.getByName(addr))) {
453       String msg =
454           "Failed to start redirecting jetty server. Address " + addr
455               + " does not belong to this host. Correct configuration parameter: "
456               + "hbase.master.info.bindAddress";
457       LOG.error(msg);
458       throw new IOException(msg);
459     }
460 
461     RedirectServlet.regionServerInfoPort = infoServer.getPort();
462     if(RedirectServlet.regionServerInfoPort == infoPort) {
463       return infoPort;
464     }
465     masterJettyServer = new org.mortbay.jetty.Server();
466     Connector connector = new SelectChannelConnector();
467     connector.setHost(addr);
468     connector.setPort(infoPort);
469     masterJettyServer.addConnector(connector);
470     masterJettyServer.setStopAtShutdown(true);
471     Context context = new Context(masterJettyServer, "/", Context.NO_SESSIONS);
472     context.addServlet(RedirectServlet.class, "/*");
473     try {
474       masterJettyServer.start();
475     } catch (Exception e) {
476       throw new IOException("Failed to start redirecting jetty server", e);
477     }
478     return connector.getLocalPort();
479   }
480 
481   /**
482    * For compatibility, if failed with regionserver credentials, try the master one
483    */
484   @Override
485   protected void login(UserProvider user, String host) throws IOException {
486     try {
487       super.login(user, host);
488     } catch (IOException ie) {
489       user.login("hbase.master.keytab.file",
490         "hbase.master.kerberos.principal", host);
491     }
492   }
493 
494   /**
495    * If configured to put regions on active master,
496    * wait till a backup master becomes active.
497    * Otherwise, loop till the server is stopped or aborted.
498    */
499   @Override
500   protected void waitForMasterActive(){
501     boolean tablesOnMaster = BaseLoadBalancer.tablesOnMaster(conf);
502     while (!(tablesOnMaster && isActiveMaster)
503         && !isStopped() && !isAborted()) {
504       sleeper.sleep();
505     }
506   }
507 
508   @VisibleForTesting
509   public MasterRpcServices getMasterRpcServices() {
510     return (MasterRpcServices)rpcServices;
511   }
512 
513   public boolean balanceSwitch(final boolean b) throws IOException {
514     return getMasterRpcServices().switchBalancer(b, BalanceSwitchMode.ASYNC);
515   }
516 
517   @Override
518   protected String getProcessName() {
519     return MASTER;
520   }
521 
522   @Override
523   protected boolean canCreateBaseZNode() {
524     return true;
525   }
526 
527   @Override
528   protected boolean canUpdateTableDescriptor() {
529     return true;
530   }
531 
532   @Override
533   protected RSRpcServices createRpcServices() throws IOException {
534     return new MasterRpcServices(this);
535   }
536 
537   @Override
538   protected void configureInfoServer() {
539     infoServer.addServlet("master-status", "/master-status", MasterStatusServlet.class);
540     infoServer.setAttribute(MASTER, this);
541     if (BaseLoadBalancer.tablesOnMaster(conf)) {
542       super.configureInfoServer();
543     }
544   }
545 
546   @Override
547   protected Class<? extends HttpServlet> getDumpServlet() {
548     return MasterDumpServlet.class;
549   }
550 
551   /**
552    * Emit the HMaster metrics, such as region in transition metrics.
553    * Surrounding in a try block just to be sure metrics doesn't abort HMaster.
554    */
555   private void doMetrics() {
556     try {
557       if (assignmentManager != null) {
558         assignmentManager.updateRegionsInTransitionMetrics();
559       }
560     } catch (Throwable e) {
561       LOG.error("Couldn't update metrics: " + e.getMessage());
562     }
563   }
564 
565   MetricsMaster getMasterMetrics() {
566     return metricsMaster;
567   }
568 
569   /**
570    * Initialize all ZK based system trackers.
571    * @throws IOException
572    * @throws InterruptedException
573    * @throws KeeperException
574    * @throws CoordinatedStateException
575    */
576   void initializeZKBasedSystemTrackers() throws IOException,
577       InterruptedException, KeeperException, CoordinatedStateException {
578     this.balancer = LoadBalancerFactory.getLoadBalancer(conf);
579     this.normalizer = RegionNormalizerFactory.getRegionNormalizer(conf);
580     this.normalizer.setMasterServices(this);
581     this.loadBalancerTracker = new LoadBalancerTracker(zooKeeper, this);
582     this.loadBalancerTracker.start();
583     this.regionNormalizerTracker = new RegionNormalizerTracker(zooKeeper, this);
584     this.regionNormalizerTracker.start();
585     this.assignmentManager = new AssignmentManager(this, serverManager,
586       this.balancer, this.service, this.metricsMaster,
587       this.tableLockManager);
588     zooKeeper.registerListenerFirst(assignmentManager);
589 
590     this.regionServerTracker = new RegionServerTracker(zooKeeper, this,
591         this.serverManager);
592     this.regionServerTracker.start();
593 
594     this.drainingServerTracker = new DrainingServerTracker(zooKeeper, this,
595       this.serverManager);
596     this.drainingServerTracker.start();
597 
598     // Set the cluster as up.  If new RSs, they'll be waiting on this before
599     // going ahead with their startup.
600     boolean wasUp = this.clusterStatusTracker.isClusterUp();
601     if (!wasUp) this.clusterStatusTracker.setClusterUp();
602 
603     LOG.info("Server active/primary master=" + this.serverName +
604         ", sessionid=0x" +
605         Long.toHexString(this.zooKeeper.getRecoverableZooKeeper().getSessionId()) +
606         ", setting cluster-up flag (Was=" + wasUp + ")");
607 
608     // create/initialize the snapshot manager and other procedure managers
609     this.snapshotManager = new SnapshotManager();
610     this.mpmHost = new MasterProcedureManagerHost();
611     this.mpmHost.register(this.snapshotManager);
612     this.mpmHost.register(new MasterFlushTableProcedureManager());
613     this.mpmHost.loadProcedures(conf);
614     this.mpmHost.initialize(this, this.metricsMaster);
615   }
616 
617   /**
618    * Finish initialization of HMaster after becoming the primary master.
619    *
620    * <ol>
621    * <li>Initialize master components - file system manager, server manager,
622    *     assignment manager, region server tracker, etc</li>
623    * <li>Start necessary service threads - balancer, catalog janior,
624    *     executor services, etc</li>
625    * <li>Set cluster as UP in ZooKeeper</li>
626    * <li>Wait for RegionServers to check-in</li>
627    * <li>Split logs and perform data recovery, if necessary</li>
628    * <li>Ensure assignment of meta/namespace regions<li>
629    * <li>Handle either fresh cluster start or master failover</li>
630    * </ol>
631    *
632    * @throws IOException
633    * @throws InterruptedException
634    * @throws KeeperException
635    * @throws CoordinatedStateException
636    */
637   private void finishActiveMasterInitialization(MonitoredTask status)
638       throws IOException, InterruptedException, KeeperException, CoordinatedStateException {
639 
640     isActiveMaster = true;
641     Thread zombieDetector = new Thread(new InitializationMonitor(this));
642     zombieDetector.start();
643 
644     /*
645      * We are active master now... go initialize components we need to run.
646      * Note, there may be dross in zk from previous runs; it'll get addressed
647      * below after we determine if cluster startup or failover.
648      */
649 
650     status.setStatus("Initializing Master file system");
651 
652     this.masterActiveTime = System.currentTimeMillis();
653     // TODO: Do this using Dependency Injection, using PicoContainer, Guice or Spring.
654     this.fileSystemManager = new MasterFileSystem(this, this);
655 
656     // enable table descriptors cache
657     this.tableDescriptors.setCacheOn();
658     // set the META's descriptor to the correct replication
659     this.tableDescriptors.get(TableName.META_TABLE_NAME).setRegionReplication(
660         conf.getInt(HConstants.META_REPLICAS_NUM, HConstants.DEFAULT_META_REPLICA_NUM));
661     // warm-up HTDs cache on master initialization
662     if (preLoadTableDescriptors) {
663       status.setStatus("Pre-loading table descriptors");
664       this.tableDescriptors.getAll();
665     }
666 
667     // publish cluster ID
668     status.setStatus("Publishing Cluster ID in ZooKeeper");
669     ZKClusterId.setClusterId(this.zooKeeper, fileSystemManager.getClusterId());
670     this.serverManager = createServerManager(this, this);
671 
672     setupClusterConnection();
673 
674     // Invalidate all write locks held previously
675     this.tableLockManager.reapWriteLocks();
676 
677     status.setStatus("Initializing ZK system trackers");
678     initializeZKBasedSystemTrackers();
679 
680     // initialize master side coprocessors before we start handling requests
681     status.setStatus("Initializing master coprocessors");
682     this.cpHost = new MasterCoprocessorHost(this, this.conf);
683 
684     // start up all service threads.
685     status.setStatus("Initializing master service threads");
686     startServiceThreads();
687 
688     // Wake up this server to check in
689     sleeper.skipSleepCycle();
690 
691     // Wait for region servers to report in
692     this.serverManager.waitForRegionServers(status);
693     // Check zk for region servers that are up but didn't register
694     for (ServerName sn: this.regionServerTracker.getOnlineServers()) {
695       // The isServerOnline check is opportunistic, correctness is handled inside
696       if (!this.serverManager.isServerOnline(sn)
697           && serverManager.checkAndRecordNewServer(sn, ServerLoad.EMPTY_SERVERLOAD)) {
698         LOG.info("Registered server found up in zk but who has not yet reported in: " + sn);
699       }
700     }
701 
702     // get a list for previously failed RS which need log splitting work
703     // we recover hbase:meta region servers inside master initialization and
704     // handle other failed servers in SSH in order to start up master node ASAP
705     Set<ServerName> previouslyFailedServers =
706       this.fileSystemManager.getFailedServersFromLogFolders();
707 
708     // log splitting for hbase:meta server
709     ServerName oldMetaServerLocation = metaTableLocator.getMetaRegionLocation(this.getZooKeeper());
710     if (oldMetaServerLocation != null && previouslyFailedServers.contains(oldMetaServerLocation)) {
711       splitMetaLogBeforeAssignment(oldMetaServerLocation);
712       // Note: we can't remove oldMetaServerLocation from previousFailedServers list because it
713       // may also host user regions
714     }
715     Set<ServerName> previouslyFailedMetaRSs = getPreviouselyFailedMetaServersFromZK();
716     // need to use union of previouslyFailedMetaRSs recorded in ZK and previouslyFailedServers
717     // instead of previouslyFailedMetaRSs alone to address the following two situations:
718     // 1) the chained failure situation(recovery failed multiple times in a row).
719     // 2) master get killed right before it could delete the recovering hbase:meta from ZK while the
720     // same server still has non-meta wals to be replayed so that
721     // removeStaleRecoveringRegionsFromZK can't delete the stale hbase:meta region
722     // Passing more servers into splitMetaLog is all right. If a server doesn't have hbase:meta wal,
723     // there is no op for the server.
724     previouslyFailedMetaRSs.addAll(previouslyFailedServers);
725 
726     this.initializationBeforeMetaAssignment = true;
727 
728     // Wait for regionserver to finish initialization.
729     if (BaseLoadBalancer.tablesOnMaster(conf)) {
730       waitForServerOnline();
731     }
732 
733     //initialize load balancer
734     this.balancer.setClusterStatus(getClusterStatus());
735     this.balancer.setMasterServices(this);
736     this.balancer.initialize();
737 
738     // Check if master is shutting down because of some issue
739     // in initializing the regionserver or the balancer.
740     if (isStopped()) return;
741 
742     // Make sure meta assigned before proceeding.
743     status.setStatus("Assigning Meta Region");
744     assignMeta(status, previouslyFailedMetaRSs, HRegionInfo.DEFAULT_REPLICA_ID);
745     // check if master is shutting down because above assignMeta could return even hbase:meta isn't
746     // assigned when master is shutting down
747     if (isStopped()) return;
748 
749     status.setStatus("Submitting log splitting work for previously failed region servers");
750     // Master has recovered hbase:meta region server and we put
751     // other failed region servers in a queue to be handled later by SSH
752     for (ServerName tmpServer : previouslyFailedServers) {
753       this.serverManager.processDeadServer(tmpServer, true);
754     }
755 
756     // Update meta with new PB serialization if required. i.e migrate all HRI to PB serialization
757     // in meta. This must happen before we assign all user regions or else the assignment will fail.
758     if (this.conf.getBoolean("hbase.MetaMigrationConvertingToPB", true)) {
759       MetaMigrationConvertingToPB.updateMetaIfNecessary(this);
760     }
761 
762     // Fix up assignment manager status
763     status.setStatus("Starting assignment manager");
764     this.assignmentManager.joinCluster();
765 
766     // set cluster status again after user regions are assigned
767     this.balancer.setClusterStatus(getClusterStatus());
768 
769     // Start balancer and meta catalog janitor after meta and regions have been assigned.
770     status.setStatus("Starting balancer and catalog janitor");
771     this.clusterStatusChore = new ClusterStatusChore(this, balancer);
772     getChoreService().scheduleChore(clusterStatusChore);
773     this.balancerChore = new BalancerChore(this);
774     getChoreService().scheduleChore(balancerChore);
775     this.normalizerChore = new RegionNormalizerChore(this);
776     getChoreService().scheduleChore(normalizerChore);
777     this.catalogJanitorChore = new CatalogJanitor(this, this);
778     getChoreService().scheduleChore(catalogJanitorChore);
779 
780     // Do Metrics periodically
781     periodicDoMetricsChore = new PeriodicDoMetrics(msgInterval, this);
782     getChoreService().scheduleChore(periodicDoMetricsChore);
783 
784     status.setStatus("Starting namespace manager");
785     initNamespace();
786 
787     if (this.cpHost != null) {
788       try {
789         this.cpHost.preMasterInitialization();
790       } catch (IOException e) {
791         LOG.error("Coprocessor preMasterInitialization() hook failed", e);
792       }
793     }
794 
795     status.markComplete("Initialization successful");
796     LOG.info("Master has completed initialization");
797     configurationManager.registerObserver(this.balancer);
798 
799     // Set master as 'initialized'.
800     setInitialized(true);
801 
802     status.setStatus("Starting quota manager");
803     initQuotaManager();
804 
805     // assign the meta replicas
806     Set<ServerName> EMPTY_SET = new HashSet<ServerName>();
807     int numReplicas = conf.getInt(HConstants.META_REPLICAS_NUM,
808            HConstants.DEFAULT_META_REPLICA_NUM);
809     for (int i = 1; i < numReplicas; i++) {
810       assignMeta(status, EMPTY_SET, i);
811     }
812     unassignExcessMetaReplica(zooKeeper, numReplicas);
813 
814     // clear the dead servers with same host name and port of online server because we are not
815     // removing dead server with same hostname and port of rs which is trying to check in before
816     // master initialization. See HBASE-5916.
817     this.serverManager.clearDeadServersWithSameHostNameAndPortOfOnlineServer();
818 
819     // Check and set the znode ACLs if needed in case we are overtaking a non-secure configuration
820     status.setStatus("Checking ZNode ACLs");
821     zooKeeper.checkAndSetZNodeAcls();
822 
823     status.setStatus("Calling postStartMaster coprocessors");
824     if (this.cpHost != null) {
825       // don't let cp initialization errors kill the master
826       try {
827         this.cpHost.postStartMaster();
828       } catch (IOException ioe) {
829         LOG.error("Coprocessor postStartMaster() hook failed", ioe);
830       }
831     }
832 
833     zombieDetector.interrupt();
834   }
835 
836   private void initQuotaManager() throws IOException {
837     quotaManager = new MasterQuotaManager(this);
838     this.assignmentManager.setRegionStateListener((RegionStateListener) quotaManager);
839     quotaManager.start();
840   }
841 
842   /**
843    * Create a {@link ServerManager} instance.
844    * @param master
845    * @param services
846    * @return An instance of {@link ServerManager}
847    * @throws org.apache.hadoop.hbase.ZooKeeperConnectionException
848    * @throws IOException
849    */
850   ServerManager createServerManager(final Server master,
851       final MasterServices services)
852   throws IOException {
853     // We put this out here in a method so can do a Mockito.spy and stub it out
854     // w/ a mocked up ServerManager.
855     return new ServerManager(master, services);
856   }
857 
858   private void unassignExcessMetaReplica(ZooKeeperWatcher zkw, int numMetaReplicasConfigured) {
859     // unassign the unneeded replicas (for e.g., if the previous master was configured
860     // with a replication of 3 and now it is 2, we need to unassign the 1 unneeded replica)
861     try {
862       List<String> metaReplicaZnodes = zooKeeper.getMetaReplicaNodes();
863       for (String metaReplicaZnode : metaReplicaZnodes) {
864         int replicaId = zooKeeper.getMetaReplicaIdFromZnode(metaReplicaZnode);
865         if (replicaId >= numMetaReplicasConfigured) {
866           RegionState r = MetaTableLocator.getMetaRegionState(zkw, replicaId);
867           LOG.info("Closing excess replica of meta region " + r.getRegion());
868           // send a close and wait for a max of 30 seconds
869           ServerManager.closeRegionSilentlyAndWait(getConnection(), r.getServerName(),
870               r.getRegion(), 30000);
871           ZKUtil.deleteNode(zkw, zkw.getZNodeForReplica(replicaId));
872         }
873       }
874     } catch (Exception ex) {
875       // ignore the exception since we don't want the master to be wedged due to potential
876       // issues in the cleanup of the extra regions. We can do that cleanup via hbck or manually
877       LOG.warn("Ignoring exception " + ex);
878     }
879   }
880 
881   /**
882    * Check <code>hbase:meta</code> is assigned. If not, assign it.
883    * @param status MonitoredTask
884    * @param previouslyFailedMetaRSs
885    * @param replicaId
886    * @throws InterruptedException
887    * @throws IOException
888    * @throws KeeperException
889    */
890   void assignMeta(MonitoredTask status, Set<ServerName> previouslyFailedMetaRSs, int replicaId)
891       throws InterruptedException, IOException, KeeperException {
892     // Work on meta region
893     int assigned = 0;
894     long timeout = this.conf.getLong("hbase.catalog.verification.timeout", 1000);
895     if (replicaId == HRegionInfo.DEFAULT_REPLICA_ID) {
896       status.setStatus("Assigning hbase:meta region");
897     } else {
898       status.setStatus("Assigning hbase:meta region, replicaId " + replicaId);
899     }
900     // Get current meta state from zk.
901     RegionStates regionStates = assignmentManager.getRegionStates();
902     RegionState metaState = MetaTableLocator.getMetaRegionState(getZooKeeper(), replicaId);
903     HRegionInfo hri = RegionReplicaUtil.getRegionInfoForReplica(HRegionInfo.FIRST_META_REGIONINFO,
904         replicaId);
905     ServerName currentMetaServer = metaState.getServerName();
906     if (!ConfigUtil.useZKForAssignment(conf)) {
907       regionStates.createRegionState(hri, metaState.getState(),
908         currentMetaServer, null);
909     } else {
910       regionStates.createRegionState(hri);
911     }
912     boolean rit = this.assignmentManager.
913       processRegionInTransitionAndBlockUntilAssigned(hri);
914     boolean metaRegionLocation = metaTableLocator.verifyMetaRegionLocation(
915       this.getConnection(), this.getZooKeeper(), timeout, replicaId);
916     if (!metaRegionLocation || !metaState.isOpened()) {
917       // Meta location is not verified. It should be in transition, or offline.
918       // We will wait for it to be assigned in enableSSHandWaitForMeta below.
919       assigned++;
920       if (!ConfigUtil.useZKForAssignment(conf)) {
921         assignMetaZkLess(regionStates, metaState, timeout, previouslyFailedMetaRSs);
922       } else if (!rit) {
923         // Assign meta since not already in transition
924         if (currentMetaServer != null) {
925           // If the meta server is not known to be dead or online,
926           // just split the meta log, and don't expire it since this
927           // could be a full cluster restart. Otherwise, we will think
928           // this is a failover and lose previous region locations.
929           // If it is really a failover case, AM will find out in rebuilding
930           // user regions. Otherwise, we are good since all logs are split
931           // or known to be replayed before user regions are assigned.
932           if (serverManager.isServerOnline(currentMetaServer)) {
933             LOG.info("Forcing expire of " + currentMetaServer);
934             serverManager.expireServer(currentMetaServer);
935           }
936           if (replicaId == HRegionInfo.DEFAULT_REPLICA_ID) {
937             splitMetaLogBeforeAssignment(currentMetaServer);
938             previouslyFailedMetaRSs.add(currentMetaServer);
939           }
940         }
941         assignmentManager.assignMeta(hri);
942       }
943     } else {
944       // Region already assigned. We didn't assign it. Add to in-memory state.
945       regionStates.updateRegionState(
946         HRegionInfo.FIRST_META_REGIONINFO, State.OPEN, currentMetaServer);
947       this.assignmentManager.regionOnline(
948         HRegionInfo.FIRST_META_REGIONINFO, currentMetaServer);
949     }
950 
951     if (replicaId == HRegionInfo.DEFAULT_REPLICA_ID) enableMeta(TableName.META_TABLE_NAME);
952 
953     if ((RecoveryMode.LOG_REPLAY == this.getMasterFileSystem().getLogRecoveryMode())
954         && (!previouslyFailedMetaRSs.isEmpty())) {
955       // replay WAL edits mode need new hbase:meta RS is assigned firstly
956       status.setStatus("replaying log for Meta Region");
957       this.fileSystemManager.splitMetaLog(previouslyFailedMetaRSs);
958     }
959 
960     // Make sure a hbase:meta location is set. We need to enable SSH here since
961     // if the meta region server is died at this time, we need it to be re-assigned
962     // by SSH so that system tables can be assigned.
963     // No need to wait for meta is assigned = 0 when meta is just verified.
964     if (replicaId == HRegionInfo.DEFAULT_REPLICA_ID) enableCrashedServerProcessing(assigned != 0);
965     LOG.info("hbase:meta with replicaId " + replicaId + " assigned=" + assigned + ", rit=" + rit +
966       ", location=" + metaTableLocator.getMetaRegionLocation(this.getZooKeeper(), replicaId));
967     status.setStatus("META assigned.");
968   }
969 
970   private void assignMetaZkLess(RegionStates regionStates, RegionState regionState, long timeout,
971       Set<ServerName> previouslyFailedRs) throws IOException, KeeperException {
972     ServerName currentServer = regionState.getServerName();
973     if (serverManager.isServerOnline(currentServer)) {
974       LOG.info("Meta was in transition on " + currentServer);
975       assignmentManager.processRegionInTransitionZkLess();
976     } else {
977       if (currentServer != null) {
978         if (regionState.getRegion().getReplicaId() == HRegionInfo.DEFAULT_REPLICA_ID) {
979           splitMetaLogBeforeAssignment(currentServer);
980           regionStates.logSplit(HRegionInfo.FIRST_META_REGIONINFO);
981           previouslyFailedRs.add(currentServer);
982         }
983       }
984       LOG.info("Re-assigning hbase:meta, it was on " + currentServer);
985       regionStates.updateRegionState(regionState.getRegion(), State.OFFLINE);
986       assignmentManager.assignMeta(regionState.getRegion());
987     }
988   }
989 
990   void initNamespace() throws IOException {
991     //create namespace manager
992     tableNamespaceManager = new TableNamespaceManager(this);
993     tableNamespaceManager.start();
994   }
995 
996   boolean isCatalogJanitorEnabled() {
997     return catalogJanitorChore != null ?
998       catalogJanitorChore.getEnabled() : false;
999   }
1000 
1001   private void splitMetaLogBeforeAssignment(ServerName currentMetaServer) throws IOException {
1002     if (RecoveryMode.LOG_REPLAY == this.getMasterFileSystem().getLogRecoveryMode()) {
1003       // In log replay mode, we mark hbase:meta region as recovering in ZK
1004       Set<HRegionInfo> regions = new HashSet<HRegionInfo>();
1005       regions.add(HRegionInfo.FIRST_META_REGIONINFO);
1006       this.fileSystemManager.prepareLogReplay(currentMetaServer, regions);
1007     } else {
1008       // In recovered.edits mode: create recovered edits file for hbase:meta server
1009       this.fileSystemManager.splitMetaLog(currentMetaServer);
1010     }
1011   }
1012 
1013   private void enableCrashedServerProcessing(final boolean waitForMeta)
1014   throws IOException, InterruptedException {
1015     // If crashed server processing is disabled, we enable it and expire those dead but not expired
1016     // servers. This is required so that if meta is assigning to a server which dies after
1017     // assignMeta starts assignment, ServerCrashProcedure can re-assign it. Otherwise, we will be
1018     // stuck here waiting forever if waitForMeta is specified.
1019     if (!isServerCrashProcessingEnabled()) {
1020       setServerCrashProcessingEnabled(true);
1021       this.serverManager.processQueuedDeadServers();
1022     }
1023 
1024     if (waitForMeta) {
1025       metaTableLocator.waitMetaRegionLocation(this.getZooKeeper());
1026       // Above check waits for general meta availability but this does not
1027       // guarantee that the transition has completed
1028       this.assignmentManager.waitForAssignment(HRegionInfo.FIRST_META_REGIONINFO);
1029     }
1030   }
1031 
1032   private void enableMeta(TableName metaTableName) {
1033     if (!this.assignmentManager.getTableStateManager().isTableState(metaTableName,
1034         ZooKeeperProtos.Table.State.ENABLED)) {
1035       this.assignmentManager.setEnabledTable(metaTableName);
1036     }
1037   }
1038 
1039   /**
1040    * This function returns a set of region server names under hbase:meta recovering region ZK node
1041    * @return Set of meta server names which were recorded in ZK
1042    * @throws KeeperException
1043    */
1044   private Set<ServerName> getPreviouselyFailedMetaServersFromZK() throws KeeperException {
1045     Set<ServerName> result = new HashSet<ServerName>();
1046     String metaRecoveringZNode = ZKUtil.joinZNode(zooKeeper.recoveringRegionsZNode,
1047       HRegionInfo.FIRST_META_REGIONINFO.getEncodedName());
1048     List<String> regionFailedServers = ZKUtil.listChildrenNoWatch(zooKeeper, metaRecoveringZNode);
1049     if (regionFailedServers == null) return result;
1050 
1051     for(String failedServer : regionFailedServers) {
1052       ServerName server = ServerName.parseServerName(failedServer);
1053       result.add(server);
1054     }
1055     return result;
1056   }
1057 
1058   @Override
1059   public TableDescriptors getTableDescriptors() {
1060     return this.tableDescriptors;
1061   }
1062 
1063   @Override
1064   public ServerManager getServerManager() {
1065     return this.serverManager;
1066   }
1067 
1068   @Override
1069   public MasterFileSystem getMasterFileSystem() {
1070     return this.fileSystemManager;
1071   }
1072 
1073   /*
1074    * Start up all services. If any of these threads gets an unhandled exception
1075    * then they just die with a logged message.  This should be fine because
1076    * in general, we do not expect the master to get such unhandled exceptions
1077    *  as OOMEs; it should be lightly loaded. See what HRegionServer does if
1078    *  need to install an unexpected exception handler.
1079    */
1080   private void startServiceThreads() throws IOException{
1081    // Start the executor service pools
1082    this.service.startExecutorService(ExecutorType.MASTER_OPEN_REGION,
1083       conf.getInt("hbase.master.executor.openregion.threads", 5));
1084    this.service.startExecutorService(ExecutorType.MASTER_CLOSE_REGION,
1085       conf.getInt("hbase.master.executor.closeregion.threads", 5));
1086    this.service.startExecutorService(ExecutorType.MASTER_SERVER_OPERATIONS,
1087       conf.getInt("hbase.master.executor.serverops.threads", 5));
1088    this.service.startExecutorService(ExecutorType.MASTER_META_SERVER_OPERATIONS,
1089       conf.getInt("hbase.master.executor.serverops.threads", 5));
1090    this.service.startExecutorService(ExecutorType.M_LOG_REPLAY_OPS,
1091       conf.getInt("hbase.master.executor.logreplayops.threads", 10));
1092 
1093    // We depend on there being only one instance of this executor running
1094    // at a time.  To do concurrency, would need fencing of enable/disable of
1095    // tables.
1096    // Any time changing this maxThreads to > 1, pls see the comment at
1097    // AccessController#postCreateTableHandler
1098    this.service.startExecutorService(ExecutorType.MASTER_TABLE_OPERATIONS, 1);
1099    startProcedureExecutor();
1100 
1101    // Start log cleaner thread
1102    int cleanerInterval = conf.getInt("hbase.master.cleaner.interval", 60 * 1000);
1103    this.logCleaner =
1104       new LogCleaner(cleanerInterval,
1105          this, conf, getMasterFileSystem().getFileSystem(),
1106          getMasterFileSystem().getOldLogDir());
1107     getChoreService().scheduleChore(logCleaner);
1108 
1109    //start the hfile archive cleaner thread
1110     Path archiveDir = HFileArchiveUtil.getArchivePath(conf);
1111     this.hfileCleaner = new HFileCleaner(cleanerInterval, this, conf, getMasterFileSystem()
1112         .getFileSystem(), archiveDir);
1113     getChoreService().scheduleChore(hfileCleaner);
1114     serviceStarted = true;
1115     if (LOG.isTraceEnabled()) {
1116       LOG.trace("Started service threads");
1117     }
1118     if (!conf.getBoolean(HConstants.ZOOKEEPER_USEMULTI, true)) {
1119       try {
1120         replicationZKLockCleanerChore = new ReplicationZKLockCleanerChore(this, this,
1121             cleanerInterval, this.getZooKeeper(), this.conf);
1122         getChoreService().scheduleChore(replicationZKLockCleanerChore);
1123       } catch (Exception e) {
1124         LOG.error("start replicationZKLockCleanerChore failed", e);
1125       }
1126     }
1127   }
1128 
1129   @Override
1130   protected void sendShutdownInterrupt() {
1131     super.sendShutdownInterrupt();
1132     stopProcedureExecutor();
1133   }
1134 
1135   @Override
1136   protected void stopServiceThreads() {
1137     if (masterJettyServer != null) {
1138       LOG.info("Stopping master jetty server");
1139       try {
1140         masterJettyServer.stop();
1141       } catch (Exception e) {
1142         LOG.error("Failed to stop master jetty server", e);
1143       }
1144     }
1145     super.stopServiceThreads();
1146     stopChores();
1147 
1148     // Wait for all the remaining region servers to report in IFF we were
1149     // running a cluster shutdown AND we were NOT aborting.
1150     if (!isAborted() && this.serverManager != null &&
1151         this.serverManager.isClusterShutdown()) {
1152       this.serverManager.letRegionServersShutdown();
1153     }
1154     if (LOG.isDebugEnabled()) {
1155       LOG.debug("Stopping service threads");
1156     }
1157     // Clean up and close up shop
1158     if (this.logCleaner != null) this.logCleaner.cancel(true);
1159     if (this.hfileCleaner != null) this.hfileCleaner.cancel(true);
1160     if (this.replicationZKLockCleanerChore != null) this.replicationZKLockCleanerChore.cancel(true);
1161     if (this.quotaManager != null) this.quotaManager.stop();
1162     if (this.activeMasterManager != null) this.activeMasterManager.stop();
1163     if (this.serverManager != null) this.serverManager.stop();
1164     if (this.assignmentManager != null) this.assignmentManager.stop();
1165     if (this.fileSystemManager != null) this.fileSystemManager.stop();
1166     if (this.mpmHost != null) this.mpmHost.stop("server shutting down.");
1167   }
1168 
1169   private void startProcedureExecutor() throws IOException {
1170     final MasterProcedureEnv procEnv = new MasterProcedureEnv(this);
1171     final Path logDir = new Path(fileSystemManager.getRootDir(),
1172         MasterProcedureConstants.MASTER_PROCEDURE_LOGDIR);
1173 
1174     procedureStore = new WALProcedureStore(conf, fileSystemManager.getFileSystem(), logDir,
1175         new MasterProcedureEnv.WALStoreLeaseRecovery(this));
1176     procedureStore.registerListener(new MasterProcedureEnv.MasterProcedureStoreListener(this));
1177     procedureExecutor = new ProcedureExecutor(conf, procEnv, procedureStore,
1178         procEnv.getProcedureQueue());
1179 
1180     final int numThreads = conf.getInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS,
1181         Math.max(Runtime.getRuntime().availableProcessors(),
1182           MasterProcedureConstants.DEFAULT_MIN_MASTER_PROCEDURE_THREADS));
1183     final boolean abortOnCorruption = conf.getBoolean(
1184         MasterProcedureConstants.EXECUTOR_ABORT_ON_CORRUPTION,
1185         MasterProcedureConstants.DEFAULT_EXECUTOR_ABORT_ON_CORRUPTION);
1186     procedureStore.start(numThreads);
1187     procedureExecutor.start(numThreads, abortOnCorruption);
1188   }
1189 
1190   private void stopProcedureExecutor() {
1191     if (procedureExecutor != null) {
1192       procedureExecutor.stop();
1193     }
1194 
1195     if (procedureStore != null) {
1196       procedureStore.stop(isAborted());
1197     }
1198   }
1199 
1200   private void stopChores() {
1201     if (this.balancerChore != null) {
1202       this.balancerChore.cancel(true);
1203     }
1204     if (this.normalizerChore != null) {
1205       this.normalizerChore.cancel(true);
1206     }
1207     if (this.clusterStatusChore != null) {
1208       this.clusterStatusChore.cancel(true);
1209     }
1210     if (this.catalogJanitorChore != null) {
1211       this.catalogJanitorChore.cancel(true);
1212     }
1213     if (this.clusterStatusPublisherChore != null){
1214       clusterStatusPublisherChore.cancel(true);
1215     }
1216     if (this.periodicDoMetricsChore != null) {
1217       periodicDoMetricsChore.cancel();
1218     }
1219   }
1220 
1221   /**
1222    * @return Get remote side's InetAddress
1223    * @throws UnknownHostException
1224    */
1225   InetAddress getRemoteInetAddress(final int port,
1226       final long serverStartCode) throws UnknownHostException {
1227     // Do it out here in its own little method so can fake an address when
1228     // mocking up in tests.
1229     InetAddress ia = RpcServer.getRemoteIp();
1230 
1231     // The call could be from the local regionserver,
1232     // in which case, there is no remote address.
1233     if (ia == null && serverStartCode == startcode) {
1234       InetSocketAddress isa = rpcServices.getSocketAddress();
1235       if (isa != null && isa.getPort() == port) {
1236         ia = isa.getAddress();
1237       }
1238     }
1239     return ia;
1240   }
1241 
1242   /**
1243    * @return Maximum time we should run balancer for
1244    */
1245   private int getBalancerCutoffTime() {
1246     int balancerCutoffTime =
1247       getConfiguration().getInt("hbase.balancer.max.balancing", -1);
1248     if (balancerCutoffTime == -1) {
1249       // No time period set so create one
1250       int balancerPeriod =
1251         getConfiguration().getInt("hbase.balancer.period", 300000);
1252       balancerCutoffTime = balancerPeriod;
1253       // If nonsense period, set it to balancerPeriod
1254       if (balancerCutoffTime <= 0) balancerCutoffTime = balancerPeriod;
1255     }
1256     return balancerCutoffTime;
1257   }
1258 
1259   public boolean balance() throws IOException {
1260     // if master not initialized, don't run balancer.
1261     if (!isInitialized()) {
1262       LOG.debug("Master has not been initialized, don't run balancer.");
1263       return false;
1264     }
1265     // Do this call outside of synchronized block.
1266     int maximumBalanceTime = getBalancerCutoffTime();
1267     synchronized (this.balancer) {
1268       // If balance not true, don't run balancer.
1269       if (!this.loadBalancerTracker.isBalancerOn()) return false;
1270       // Only allow one balance run at at time.
1271       if (this.assignmentManager.getRegionStates().isRegionsInTransition()) {
1272         Map<String, RegionState> regionsInTransition =
1273           this.assignmentManager.getRegionStates().getRegionsInTransition();
1274         LOG.debug("Not running balancer because " + regionsInTransition.size() +
1275           " region(s) in transition: " + org.apache.commons.lang.StringUtils.
1276             abbreviate(regionsInTransition.toString(), 256));
1277         return false;
1278       }
1279       if (this.serverManager.areDeadServersInProgress()) {
1280         LOG.debug("Not running balancer because processing dead regionserver(s): " +
1281           this.serverManager.getDeadServers());
1282         return false;
1283       }
1284 
1285       if (this.cpHost != null) {
1286         try {
1287           if (this.cpHost.preBalance()) {
1288             LOG.debug("Coprocessor bypassing balancer request");
1289             return false;
1290           }
1291         } catch (IOException ioe) {
1292           LOG.error("Error invoking master coprocessor preBalance()", ioe);
1293           return false;
1294         }
1295       }
1296 
1297       Map<TableName, Map<ServerName, List<HRegionInfo>>> assignmentsByTable =
1298         this.assignmentManager.getRegionStates().getAssignmentsByTable();
1299 
1300       List<RegionPlan> plans = new ArrayList<RegionPlan>();
1301       //Give the balancer the current cluster state.
1302       this.balancer.setClusterStatus(getClusterStatus());
1303       for (Map<ServerName, List<HRegionInfo>> assignments : assignmentsByTable.values()) {
1304         List<RegionPlan> partialPlans = this.balancer.balanceCluster(assignments);
1305         if (partialPlans != null) plans.addAll(partialPlans);
1306       }
1307       long cutoffTime = System.currentTimeMillis() + maximumBalanceTime;
1308       int rpCount = 0;  // number of RegionPlans balanced so far
1309       long totalRegPlanExecTime = 0;
1310       if (plans != null && !plans.isEmpty()) {
1311         for (RegionPlan plan: plans) {
1312           LOG.info("balance " + plan);
1313           long balStartTime = System.currentTimeMillis();
1314           //TODO: bulk assign
1315           this.assignmentManager.balance(plan);
1316           totalRegPlanExecTime += System.currentTimeMillis()-balStartTime;
1317           rpCount++;
1318           if (rpCount < plans.size() &&
1319               // if performing next balance exceeds cutoff time, exit the loop
1320               (System.currentTimeMillis() + (totalRegPlanExecTime / rpCount)) > cutoffTime) {
1321             //TODO: After balance, there should not be a cutoff time (keeping it as a security net for now)
1322             LOG.debug("No more balancing till next balance run; maximumBalanceTime=" +
1323               maximumBalanceTime);
1324             break;
1325           }
1326         }
1327       }
1328       if (this.cpHost != null) {
1329         try {
1330           this.cpHost.postBalance(rpCount < plans.size() ? plans.subList(0, rpCount) : plans);
1331         } catch (IOException ioe) {
1332           // balancing already succeeded so don't change the result
1333           LOG.error("Error invoking master coprocessor postBalance()", ioe);
1334         }
1335       }
1336     }
1337     // If LoadBalancer did not generate any plans, it means the cluster is already balanced.
1338     // Return true indicating a success.
1339     return true;
1340   }
1341 
1342   /**
1343    * Perform normalization of cluster (invoked by {@link RegionNormalizerChore}).
1344    *
1345    * @return true if normalization step was performed successfully, false otherwise
1346    *   (specifically, if HMaster hasn't been initialized properly or normalization
1347    *   is globally disabled)
1348    * @throws IOException
1349    * @throws CoordinatedStateException
1350    */
1351   public boolean normalizeRegions() throws IOException, CoordinatedStateException {
1352     if (!isInitialized()) {
1353       LOG.debug("Master has not been initialized, don't run region normalizer.");
1354       return false;
1355     }
1356 
1357     if (!this.regionNormalizerTracker.isNormalizerOn()) {
1358       LOG.debug("Region normalization is disabled, don't run region normalizer.");
1359       return false;
1360     }
1361 
1362     synchronized (this.normalizer) {
1363       // Don't run the normalizer concurrently
1364       List<TableName> allEnabledTables = new ArrayList<>(
1365         this.assignmentManager.getTableStateManager().getTablesInStates(
1366           ZooKeeperProtos.Table.State.ENABLED));
1367 
1368       Collections.shuffle(allEnabledTables);
1369 
1370       for (TableName table : allEnabledTables) {
1371         if (quotaManager.getNamespaceQuotaManager() != null &&
1372             quotaManager.getNamespaceQuotaManager().getState(table.getNamespaceAsString()) != null){
1373           LOG.debug("Skipping normalizing " + table + " since its namespace has quota");
1374           continue;
1375         }
1376         if (table.isSystemTable() || (getTableDescriptors().get(table) != null &&
1377             !getTableDescriptors().get(table).isNormalizationEnabled())) {
1378           LOG.debug("Skipping normalization for table: " + table + ", as it's either system"
1379               + " table or doesn't have auto normalization turned on");
1380           continue;
1381         }
1382         List<NormalizationPlan> plans = this.normalizer.computePlanForTable(table);
1383         if (plans != null) {
1384           for (NormalizationPlan plan : plans) {
1385             plan.execute(clusterConnection.getAdmin());
1386           }
1387         }
1388       }
1389     }
1390     // If Region did not generate any plans, it means the cluster is already balanced.
1391     // Return true indicating a success.
1392     return true;
1393   }
1394 
1395   /**
1396    * @return Client info for use as prefix on an audit log string; who did an action
1397    */
1398   String getClientIdAuditPrefix() {
1399     return "Client=" + RpcServer.getRequestUserName() + "/" + RpcServer.getRemoteAddress();
1400   }
1401 
1402   /**
1403    * Switch for the background CatalogJanitor thread.
1404    * Used for testing.  The thread will continue to run.  It will just be a noop
1405    * if disabled.
1406    * @param b If false, the catalog janitor won't do anything.
1407    */
1408   public void setCatalogJanitorEnabled(final boolean b) {
1409     this.catalogJanitorChore.setEnabled(b);
1410   }
1411 
1412   @Override
1413   public void dispatchMergingRegions(final HRegionInfo region_a,
1414       final HRegionInfo region_b, final boolean forcible) throws IOException {
1415     checkInitialized();
1416     this.service.submit(new DispatchMergingRegionHandler(this,
1417       this.catalogJanitorChore, region_a, region_b, forcible));
1418   }
1419 
1420   void move(final byte[] encodedRegionName,
1421       final byte[] destServerName) throws HBaseIOException {
1422     RegionState regionState = assignmentManager.getRegionStates().
1423       getRegionState(Bytes.toString(encodedRegionName));
1424     if (regionState == null) {
1425       throw new UnknownRegionException(Bytes.toStringBinary(encodedRegionName));
1426     }
1427 
1428     HRegionInfo hri = regionState.getRegion();
1429     ServerName dest;
1430     if (destServerName == null || destServerName.length == 0) {
1431       LOG.info("Passed destination servername is null/empty so " +
1432         "choosing a server at random");
1433       final List<ServerName> destServers = this.serverManager.createDestinationServersList(
1434         regionState.getServerName());
1435       dest = balancer.randomAssignment(hri, destServers);
1436       if (dest == null) {
1437         LOG.debug("Unable to determine a plan to assign " + hri);
1438         return;
1439       }
1440     } else {
1441       dest = ServerName.valueOf(Bytes.toString(destServerName));
1442       if (dest.equals(serverName) && balancer instanceof BaseLoadBalancer
1443           && !((BaseLoadBalancer)balancer).shouldBeOnMaster(hri)) {
1444         // To avoid unnecessary region moving later by balancer. Don't put user
1445         // regions on master. Regions on master could be put on other region
1446         // server intentionally by test however.
1447         LOG.debug("Skipping move of region " + hri.getRegionNameAsString()
1448           + " to avoid unnecessary region moving later by load balancer,"
1449           + " because it should not be on master");
1450         return;
1451       }
1452     }
1453 
1454     if (dest.equals(regionState.getServerName())) {
1455       LOG.debug("Skipping move of region " + hri.getRegionNameAsString()
1456         + " because region already assigned to the same server " + dest + ".");
1457       return;
1458     }
1459 
1460     // Now we can do the move
1461     RegionPlan rp = new RegionPlan(hri, regionState.getServerName(), dest);
1462 
1463     try {
1464       checkInitialized();
1465       if (this.cpHost != null) {
1466         if (this.cpHost.preMove(hri, rp.getSource(), rp.getDestination())) {
1467           return;
1468         }
1469       }
1470       // warmup the region on the destination before initiating the move. this call
1471       // is synchronous and takes some time. doing it before the source region gets
1472       // closed
1473       serverManager.sendRegionWarmup(rp.getDestination(), hri);
1474 
1475       LOG.info(getClientIdAuditPrefix() + " move " + rp + ", running balancer");
1476       this.assignmentManager.balance(rp);
1477       if (this.cpHost != null) {
1478         this.cpHost.postMove(hri, rp.getSource(), rp.getDestination());
1479       }
1480     } catch (IOException ioe) {
1481       if (ioe instanceof HBaseIOException) {
1482         throw (HBaseIOException)ioe;
1483       }
1484       throw new HBaseIOException(ioe);
1485     }
1486   }
1487 
1488   @Override
1489   public long createTable(
1490       final HTableDescriptor hTableDescriptor,
1491       final byte [][] splitKeys,
1492       final long nonceGroup,
1493       final long nonce) throws IOException {
1494     if (isStopped()) {
1495       throw new MasterNotRunningException();
1496     }
1497 
1498     String namespace = hTableDescriptor.getTableName().getNamespaceAsString();
1499     ensureNamespaceExists(namespace);
1500 
1501     HRegionInfo[] newRegions = ModifyRegionUtils.createHRegionInfos(hTableDescriptor, splitKeys);
1502     checkInitialized();
1503     sanityCheckTableDescriptor(hTableDescriptor);
1504 
1505     if (cpHost != null) {
1506       cpHost.preCreateTable(hTableDescriptor, newRegions);
1507     }
1508     LOG.info(getClientIdAuditPrefix() + " create " + hTableDescriptor);
1509 
1510     // TODO: We can handle/merge duplicate requests, and differentiate the case of
1511     //       TableExistsException by saying if the schema is the same or not.
1512     ProcedurePrepareLatch latch = ProcedurePrepareLatch.createLatch();
1513     long procId = this.procedureExecutor.submitProcedure(
1514       new CreateTableProcedure(
1515         procedureExecutor.getEnvironment(), hTableDescriptor, newRegions, latch),
1516       nonceGroup,
1517       nonce);
1518     latch.await();
1519 
1520     if (cpHost != null) {
1521       cpHost.postCreateTable(hTableDescriptor, newRegions);
1522     }
1523 
1524     return procId;
1525   }
1526 
1527   /**
1528    * Checks whether the table conforms to some sane limits, and configured
1529    * values (compression, etc) work. Throws an exception if something is wrong.
1530    * @throws IOException
1531    */
1532   private void sanityCheckTableDescriptor(final HTableDescriptor htd) throws IOException {
1533     final String CONF_KEY = "hbase.table.sanity.checks";
1534     boolean logWarn = false;
1535     if (!conf.getBoolean(CONF_KEY, true)) {
1536       logWarn = true;
1537     }
1538     String tableVal = htd.getConfigurationValue(CONF_KEY);
1539     if (tableVal != null && !Boolean.valueOf(tableVal)) {
1540       logWarn = true;
1541     }
1542 
1543     // check max file size
1544     long maxFileSizeLowerLimit = 2 * 1024 * 1024L; // 2M is the default lower limit
1545     long maxFileSize = htd.getMaxFileSize();
1546     if (maxFileSize < 0) {
1547       maxFileSize = conf.getLong(HConstants.HREGION_MAX_FILESIZE, maxFileSizeLowerLimit);
1548     }
1549     if (maxFileSize < conf.getLong("hbase.hregion.max.filesize.limit", maxFileSizeLowerLimit)) {
1550       String message = "MAX_FILESIZE for table descriptor or "
1551           + "\"hbase.hregion.max.filesize\" (" + maxFileSize
1552           + ") is too small, which might cause over splitting into unmanageable "
1553           + "number of regions.";
1554       warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1555     }
1556 
1557     // check flush size
1558     long flushSizeLowerLimit = 1024 * 1024L; // 1M is the default lower limit
1559     long flushSize = htd.getMemStoreFlushSize();
1560     if (flushSize < 0) {
1561       flushSize = conf.getLong(HConstants.HREGION_MEMSTORE_FLUSH_SIZE, flushSizeLowerLimit);
1562     }
1563     if (flushSize < conf.getLong("hbase.hregion.memstore.flush.size.limit", flushSizeLowerLimit)) {
1564       String message = "MEMSTORE_FLUSHSIZE for table descriptor or "
1565           + "\"hbase.hregion.memstore.flush.size\" ("+flushSize+") is too small, which might cause"
1566           + " very frequent flushing.";
1567       warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1568     }
1569 
1570     // check that coprocessors and other specified plugin classes can be loaded
1571     try {
1572       checkClassLoading(conf, htd);
1573     } catch (Exception ex) {
1574       warnOrThrowExceptionForFailure(logWarn, CONF_KEY, ex.getMessage(), null);
1575     }
1576 
1577     // check compression can be loaded
1578     try {
1579       checkCompression(htd);
1580     } catch (IOException e) {
1581       warnOrThrowExceptionForFailure(logWarn, CONF_KEY, e.getMessage(), e);
1582     }
1583 
1584     // check encryption can be loaded
1585     try {
1586       checkEncryption(conf, htd);
1587     } catch (IOException e) {
1588       warnOrThrowExceptionForFailure(logWarn, CONF_KEY, e.getMessage(), e);
1589     }
1590     // Verify compaction policy
1591     try{
1592       checkCompactionPolicy(conf, htd);
1593     } catch(IOException e){
1594       warnOrThrowExceptionForFailure(false, CONF_KEY, e.getMessage(), e);
1595     }
1596     // check that we have at least 1 CF
1597     if (htd.getColumnFamilies().length == 0) {
1598       String message = "Table should have at least one column family.";
1599       warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1600     }
1601 
1602     for (HColumnDescriptor hcd : htd.getColumnFamilies()) {
1603       if (hcd.getTimeToLive() <= 0) {
1604         String message = "TTL for column family " + hcd.getNameAsString() + " must be positive.";
1605         warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1606       }
1607 
1608       // check blockSize
1609       if (hcd.getBlocksize() < 1024 || hcd.getBlocksize() > 16 * 1024 * 1024) {
1610         String message = "Block size for column family " + hcd.getNameAsString()
1611             + "  must be between 1K and 16MB.";
1612         warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1613       }
1614 
1615       // check versions
1616       if (hcd.getMinVersions() < 0) {
1617         String message = "Min versions for column family " + hcd.getNameAsString()
1618           + "  must be positive.";
1619         warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1620       }
1621       // max versions already being checked
1622 
1623       // HBASE-13776 Setting illegal versions for HColumnDescriptor
1624       //  does not throw IllegalArgumentException
1625       // check minVersions <= maxVerions
1626       if (hcd.getMinVersions() > hcd.getMaxVersions()) {
1627         String message = "Min versions for column family " + hcd.getNameAsString()
1628             + " must be less than the Max versions.";
1629         warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1630       }
1631 
1632       // check replication scope
1633       if (hcd.getScope() < 0) {
1634         String message = "Replication scope for column family "
1635           + hcd.getNameAsString() + "  must be positive.";
1636         warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1637       }
1638 
1639       // check data replication factor, it can be 0(default value) when user has not explicitly
1640       // set the value, in this case we use default replication factor set in the file system.
1641       if (hcd.getDFSReplication() < 0) {
1642         String message = "HFile Replication for column family " + hcd.getNameAsString()
1643             + "  must be greater than zero.";
1644         warnOrThrowExceptionForFailure(logWarn, CONF_KEY, message, null);
1645       }
1646 
1647       // TODO: should we check coprocessors and encryption ?
1648     }
1649   }
1650 
1651   private void checkCompactionPolicy(Configuration conf, HTableDescriptor htd)
1652       throws IOException {
1653     // FIFO compaction has some requirements
1654     // Actually FCP ignores periodic major compactions
1655     String className =
1656         htd.getConfigurationValue(DefaultStoreEngine.DEFAULT_COMPACTION_POLICY_CLASS_KEY);
1657     if (className == null) {
1658       className =
1659           conf.get(DefaultStoreEngine.DEFAULT_COMPACTION_POLICY_CLASS_KEY,
1660             ExploringCompactionPolicy.class.getName());
1661     }
1662 
1663     int blockingFileCount = HStore.DEFAULT_BLOCKING_STOREFILE_COUNT;
1664     String sv = htd.getConfigurationValue(HStore.BLOCKING_STOREFILES_KEY);
1665     if (sv != null) {
1666       blockingFileCount = Integer.parseInt(sv);
1667     } else {
1668       blockingFileCount = conf.getInt(HStore.BLOCKING_STOREFILES_KEY, blockingFileCount);
1669     }
1670 
1671     for (HColumnDescriptor hcd : htd.getColumnFamilies()) {
1672       String compactionPolicy =
1673           hcd.getConfigurationValue(DefaultStoreEngine.DEFAULT_COMPACTION_POLICY_CLASS_KEY);
1674       if (compactionPolicy == null) {
1675         compactionPolicy = className;
1676       }
1677       if (!compactionPolicy.equals(FIFOCompactionPolicy.class.getName())) {
1678         continue;
1679       }
1680       // FIFOCompaction
1681       String message = null;
1682 
1683       // 1. Check TTL
1684       if (hcd.getTimeToLive() == HColumnDescriptor.DEFAULT_TTL) {
1685         message = "Default TTL is not supported for FIFO compaction";
1686         throw new IOException(message);
1687       }
1688 
1689       // 2. Check min versions
1690       if (hcd.getMinVersions() > 0) {
1691         message = "MIN_VERSION > 0 is not supported for FIFO compaction";
1692         throw new IOException(message);
1693       }
1694 
1695       // 3. blocking file count
1696       String sbfc = htd.getConfigurationValue(HStore.BLOCKING_STOREFILES_KEY);
1697       if (sbfc != null) {
1698         blockingFileCount = Integer.parseInt(sbfc);
1699       }
1700       if (blockingFileCount < 1000) {
1701         message =
1702             "blocking file count '" + HStore.BLOCKING_STOREFILES_KEY + "' " + blockingFileCount
1703                 + " is below recommended minimum of 1000";
1704         throw new IOException(message);
1705       }
1706     }
1707   }
1708 
1709   // HBASE-13350 - Helper method to log warning on sanity check failures if checks disabled.
1710   private static void warnOrThrowExceptionForFailure(boolean logWarn, String confKey,
1711       String message, Exception cause) throws IOException {
1712     if (!logWarn) {
1713       throw new DoNotRetryIOException(message + " Set " + confKey +
1714           " to false at conf or table descriptor if you want to bypass sanity checks", cause);
1715     }
1716     LOG.warn(message);
1717   }
1718 
1719   private void startActiveMasterManager(int infoPort) throws KeeperException {
1720     String backupZNode = ZKUtil.joinZNode(
1721       zooKeeper.backupMasterAddressesZNode, serverName.toString());
1722     /*
1723     * Add a ZNode for ourselves in the backup master directory since we
1724     * may not become the active master. If so, we want the actual active
1725     * master to know we are backup masters, so that it won't assign
1726     * regions to us if so configured.
1727     *
1728     * If we become the active master later, ActiveMasterManager will delete
1729     * this node explicitly.  If we crash before then, ZooKeeper will delete
1730     * this node for us since it is ephemeral.
1731     */
1732     LOG.info("Adding backup master ZNode " + backupZNode);
1733     if (!MasterAddressTracker.setMasterAddress(zooKeeper, backupZNode,
1734         serverName, infoPort)) {
1735       LOG.warn("Failed create of " + backupZNode + " by " + serverName);
1736     }
1737 
1738     activeMasterManager.setInfoPort(infoPort);
1739     // Start a thread to try to become the active master, so we won't block here
1740     Threads.setDaemonThreadRunning(new Thread(new Runnable() {
1741       @Override
1742       public void run() {
1743         int timeout = conf.getInt(HConstants.ZK_SESSION_TIMEOUT,
1744           HConstants.DEFAULT_ZK_SESSION_TIMEOUT);
1745         // If we're a backup master, stall until a primary to writes his address
1746         if (conf.getBoolean(HConstants.MASTER_TYPE_BACKUP,
1747           HConstants.DEFAULT_MASTER_TYPE_BACKUP)) {
1748           LOG.debug("HMaster started in backup mode. "
1749             + "Stalling until master znode is written.");
1750           // This will only be a minute or so while the cluster starts up,
1751           // so don't worry about setting watches on the parent znode
1752           while (!activeMasterManager.hasActiveMaster()) {
1753             LOG.debug("Waiting for master address ZNode to be written "
1754               + "(Also watching cluster state node)");
1755             Threads.sleep(timeout);
1756           }
1757         }
1758         MonitoredTask status = TaskMonitor.get().createStatus("Master startup");
1759         status.setDescription("Master startup");
1760         try {
1761           if (activeMasterManager.blockUntilBecomingActiveMaster(timeout, status)) {
1762             finishActiveMasterInitialization(status);
1763           }
1764         } catch (Throwable t) {
1765           status.setStatus("Failed to become active: " + t.getMessage());
1766           LOG.fatal("Failed to become active master", t);
1767           // HBASE-5680: Likely hadoop23 vs hadoop 20.x/1.x incompatibility
1768           if (t instanceof NoClassDefFoundError &&
1769             t.getMessage()
1770               .contains("org/apache/hadoop/hdfs/protocol/HdfsConstants$SafeModeAction")) {
1771             // improved error message for this special case
1772             abort("HBase is having a problem with its Hadoop jars.  You may need to "
1773               + "recompile HBase against Hadoop version "
1774               + org.apache.hadoop.util.VersionInfo.getVersion()
1775               + " or change your hadoop jars to start properly", t);
1776           } else {
1777             abort("Unhandled exception. Starting shutdown.", t);
1778           }
1779         } finally {
1780           status.cleanup();
1781         }
1782       }
1783     }, getServerName().toShortString() + ".activeMasterManager"));
1784   }
1785 
1786   private void checkCompression(final HTableDescriptor htd)
1787   throws IOException {
1788     if (!this.masterCheckCompression) return;
1789     for (HColumnDescriptor hcd : htd.getColumnFamilies()) {
1790       checkCompression(hcd);
1791     }
1792   }
1793 
1794   private void checkCompression(final HColumnDescriptor hcd)
1795   throws IOException {
1796     if (!this.masterCheckCompression) return;
1797     CompressionTest.testCompression(hcd.getCompression());
1798     CompressionTest.testCompression(hcd.getCompactionCompression());
1799   }
1800 
1801   private void checkEncryption(final Configuration conf, final HTableDescriptor htd)
1802   throws IOException {
1803     if (!this.masterCheckEncryption) return;
1804     for (HColumnDescriptor hcd : htd.getColumnFamilies()) {
1805       checkEncryption(conf, hcd);
1806     }
1807   }
1808 
1809   private void checkEncryption(final Configuration conf, final HColumnDescriptor hcd)
1810   throws IOException {
1811     if (!this.masterCheckEncryption) return;
1812     EncryptionTest.testEncryption(conf, hcd.getEncryptionType(), hcd.getEncryptionKey());
1813   }
1814 
1815   private void checkClassLoading(final Configuration conf, final HTableDescriptor htd)
1816   throws IOException {
1817     RegionSplitPolicy.getSplitPolicyClass(htd, conf);
1818     RegionCoprocessorHost.testTableCoprocessorAttrs(conf, htd);
1819   }
1820 
1821   private static boolean isCatalogTable(final TableName tableName) {
1822     return tableName.equals(TableName.META_TABLE_NAME);
1823   }
1824 
1825   @Override
1826   public long deleteTable(
1827       final TableName tableName,
1828       final long nonceGroup,
1829       final long nonce) throws IOException {
1830     checkInitialized();
1831     if (cpHost != null) {
1832       cpHost.preDeleteTable(tableName);
1833     }
1834     LOG.info(getClientIdAuditPrefix() + " delete " + tableName);
1835 
1836     // TODO: We can handle/merge duplicate request
1837     ProcedurePrepareLatch latch = ProcedurePrepareLatch.createLatch();
1838     long procId = this.procedureExecutor.submitProcedure(
1839       new DeleteTableProcedure(procedureExecutor.getEnvironment(), tableName, latch),
1840       nonceGroup,
1841       nonce);
1842     latch.await();
1843 
1844     if (cpHost != null) {
1845       cpHost.postDeleteTable(tableName);
1846     }
1847 
1848     return procId;
1849   }
1850 
1851   @Override
1852   public void truncateTable(
1853       final TableName tableName,
1854       final boolean preserveSplits,
1855       final long nonceGroup,
1856       final long nonce) throws IOException {
1857     checkInitialized();
1858     if (cpHost != null) {
1859       cpHost.preTruncateTable(tableName);
1860     }
1861     LOG.info(getClientIdAuditPrefix() + " truncate " + tableName);
1862 
1863     long procId = this.procedureExecutor.submitProcedure(
1864       new TruncateTableProcedure(procedureExecutor.getEnvironment(), tableName, preserveSplits),
1865       nonceGroup,
1866       nonce);
1867     ProcedureSyncWait.waitForProcedureToComplete(procedureExecutor, procId);
1868 
1869     if (cpHost != null) {
1870       cpHost.postTruncateTable(tableName);
1871     }
1872   }
1873 
1874   @Override
1875   public void addColumn(
1876       final TableName tableName,
1877       final HColumnDescriptor columnDescriptor,
1878       final long nonceGroup,
1879       final long nonce)
1880       throws IOException {
1881     checkInitialized();
1882     checkCompression(columnDescriptor);
1883     checkEncryption(conf, columnDescriptor);
1884     if (cpHost != null) {
1885       if (cpHost.preAddColumn(tableName, columnDescriptor)) {
1886         return;
1887       }
1888     }
1889     // Execute the operation synchronously - wait for the operation to complete before continuing.
1890     long procId = this.procedureExecutor.submitProcedure(
1891       new AddColumnFamilyProcedure(procedureExecutor.getEnvironment(), tableName, columnDescriptor),
1892       nonceGroup,
1893       nonce);
1894     ProcedureSyncWait.waitForProcedureToComplete(procedureExecutor, procId);
1895     if (cpHost != null) {
1896       cpHost.postAddColumn(tableName, columnDescriptor);
1897     }
1898   }
1899 
1900   @Override
1901   public void modifyColumn(
1902       final TableName tableName,
1903       final HColumnDescriptor descriptor,
1904       final long nonceGroup,
1905       final long nonce)
1906       throws IOException {
1907     checkInitialized();
1908     checkCompression(descriptor);
1909     checkEncryption(conf, descriptor);
1910     if (cpHost != null) {
1911       if (cpHost.preModifyColumn(tableName, descriptor)) {
1912         return;
1913       }
1914     }
1915     LOG.info(getClientIdAuditPrefix() + " modify " + descriptor);
1916 
1917     // Execute the operation synchronously - wait for the operation to complete before continuing.
1918     long procId = this.procedureExecutor.submitProcedure(
1919       new ModifyColumnFamilyProcedure(procedureExecutor.getEnvironment(), tableName, descriptor),
1920       nonceGroup,
1921       nonce);
1922     ProcedureSyncWait.waitForProcedureToComplete(procedureExecutor, procId);
1923 
1924     if (cpHost != null) {
1925       cpHost.postModifyColumn(tableName, descriptor);
1926     }
1927   }
1928 
1929   @Override
1930   public void deleteColumn(
1931       final TableName tableName,
1932       final byte[] columnName,
1933       final long nonceGroup,
1934       final long nonce)
1935       throws IOException {
1936     checkInitialized();
1937     if (cpHost != null) {
1938       if (cpHost.preDeleteColumn(tableName, columnName)) {
1939         return;
1940       }
1941     }
1942     LOG.info(getClientIdAuditPrefix() + " delete " + Bytes.toString(columnName));
1943 
1944     // Execute the operation synchronously - wait for the operation to complete before continuing.
1945     long procId = this.procedureExecutor.submitProcedure(
1946       new DeleteColumnFamilyProcedure(procedureExecutor.getEnvironment(), tableName, columnName),
1947       nonceGroup,
1948       nonce);
1949     ProcedureSyncWait.waitForProcedureToComplete(procedureExecutor, procId);
1950 
1951     if (cpHost != null) {
1952       cpHost.postDeleteColumn(tableName, columnName);
1953     }
1954   }
1955 
1956   @Override
1957   public long enableTable(
1958       final TableName tableName,
1959       final long nonceGroup,
1960       final long nonce) throws IOException {
1961     checkInitialized();
1962     if (cpHost != null) {
1963       cpHost.preEnableTable(tableName);
1964     }
1965     LOG.info(getClientIdAuditPrefix() + " enable " + tableName);
1966 
1967     // Execute the operation asynchronously - client will check the progress of the operation
1968     final ProcedurePrepareLatch prepareLatch = ProcedurePrepareLatch.createLatch();
1969     long procId = this.procedureExecutor.submitProcedure(
1970       new EnableTableProcedure(procedureExecutor.getEnvironment(), tableName, false, prepareLatch),
1971       nonceGroup,
1972       nonce);
1973     // Before returning to client, we want to make sure that the table is prepared to be
1974     // enabled (the table is locked and the table state is set).
1975     //
1976     // Note: if the procedure throws exception, we will catch it and rethrow.
1977     prepareLatch.await();
1978 
1979     if (cpHost != null) {
1980       cpHost.postEnableTable(tableName);
1981     }
1982 
1983     return procId;
1984   }
1985 
1986   @Override
1987   public long disableTable(
1988       final TableName tableName,
1989       final long nonceGroup,
1990       final long nonce) throws IOException {
1991     checkInitialized();
1992     if (cpHost != null) {
1993       cpHost.preDisableTable(tableName);
1994     }
1995     LOG.info(getClientIdAuditPrefix() + " disable " + tableName);
1996 
1997     // Execute the operation asynchronously - client will check the progress of the operation
1998     final ProcedurePrepareLatch prepareLatch = ProcedurePrepareLatch.createLatch();
1999     // Execute the operation asynchronously - client will check the progress of the operation
2000     long procId = this.procedureExecutor.submitProcedure(
2001       new DisableTableProcedure(procedureExecutor.getEnvironment(), tableName, false, prepareLatch),
2002       nonceGroup,
2003       nonce);
2004     // Before returning to client, we want to make sure that the table is prepared to be
2005     // enabled (the table is locked and the table state is set).
2006     //
2007     // Note: if the procedure throws exception, we will catch it and rethrow.
2008     prepareLatch.await();
2009 
2010     if (cpHost != null) {
2011       cpHost.postDisableTable(tableName);
2012     }
2013 
2014     return procId;
2015   }
2016 
2017   /**
2018    * Return the region and current deployment for the region containing
2019    * the given row. If the region cannot be found, returns null. If it
2020    * is found, but not currently deployed, the second element of the pair
2021    * may be null.
2022    */
2023   @VisibleForTesting // Used by TestMaster.
2024   Pair<HRegionInfo, ServerName> getTableRegionForRow(
2025       final TableName tableName, final byte [] rowKey)
2026   throws IOException {
2027     final AtomicReference<Pair<HRegionInfo, ServerName>> result =
2028       new AtomicReference<Pair<HRegionInfo, ServerName>>(null);
2029 
2030     MetaScannerVisitor visitor =
2031       new MetaScannerVisitorBase() {
2032         @Override
2033         public boolean processRow(Result data) throws IOException {
2034           if (data == null || data.size() <= 0) {
2035             return true;
2036           }
2037           Pair<HRegionInfo, ServerName> pair = HRegionInfo.getHRegionInfoAndServerName(data);
2038           if (pair == null) {
2039             return false;
2040           }
2041           if (!pair.getFirst().getTable().equals(tableName)) {
2042             return false;
2043           }
2044           result.set(pair);
2045           return true;
2046         }
2047     };
2048 
2049     MetaScanner.metaScan(clusterConnection, visitor, tableName, rowKey, 1);
2050     return result.get();
2051   }
2052 
2053   @Override
2054   public void modifyTable(
2055       final TableName tableName,
2056       final HTableDescriptor descriptor,
2057       final long nonceGroup,
2058       final long nonce)
2059       throws IOException {
2060     checkInitialized();
2061     sanityCheckTableDescriptor(descriptor);
2062     if (cpHost != null) {
2063       cpHost.preModifyTable(tableName, descriptor);
2064     }
2065 
2066     LOG.info(getClientIdAuditPrefix() + " modify " + tableName);
2067 
2068     // Execute the operation synchronously - wait for the operation completes before continuing.
2069     long procId = this.procedureExecutor.submitProcedure(
2070       new ModifyTableProcedure(procedureExecutor.getEnvironment(), descriptor),
2071       nonceGroup,
2072       nonce);
2073 
2074     ProcedureSyncWait.waitForProcedureToComplete(procedureExecutor, procId);
2075 
2076     if (cpHost != null) {
2077       cpHost.postModifyTable(tableName, descriptor);
2078     }
2079   }
2080 
2081   @Override
2082   public void checkTableModifiable(final TableName tableName)
2083       throws IOException, TableNotFoundException, TableNotDisabledException {
2084     if (isCatalogTable(tableName)) {
2085       throw new IOException("Can't modify catalog tables");
2086     }
2087     if (!MetaTableAccessor.tableExists(getConnection(), tableName)) {
2088       throw new TableNotFoundException(tableName);
2089     }
2090     if (!getAssignmentManager().getTableStateManager().
2091         isTableState(tableName, ZooKeeperProtos.Table.State.DISABLED)) {
2092       throw new TableNotDisabledException(tableName);
2093     }
2094   }
2095 
2096   /**
2097    * @return cluster status
2098    */
2099   public ClusterStatus getClusterStatus() throws InterruptedIOException {
2100     // Build Set of backup masters from ZK nodes
2101     List<String> backupMasterStrings;
2102     try {
2103       backupMasterStrings = ZKUtil.listChildrenNoWatch(this.zooKeeper,
2104         this.zooKeeper.backupMasterAddressesZNode);
2105     } catch (KeeperException e) {
2106       LOG.warn(this.zooKeeper.prefix("Unable to list backup servers"), e);
2107       backupMasterStrings = null;
2108     }
2109 
2110     List<ServerName> backupMasters = null;
2111     if (backupMasterStrings != null && !backupMasterStrings.isEmpty()) {
2112       backupMasters = new ArrayList<ServerName>(backupMasterStrings.size());
2113       for (String s: backupMasterStrings) {
2114         try {
2115           byte [] bytes;
2116           try {
2117             bytes = ZKUtil.getData(this.zooKeeper, ZKUtil.joinZNode(
2118                 this.zooKeeper.backupMasterAddressesZNode, s));
2119           } catch (InterruptedException e) {
2120             throw new InterruptedIOException();
2121           }
2122           if (bytes != null) {
2123             ServerName sn;
2124             try {
2125               sn = ServerName.parseFrom(bytes);
2126             } catch (DeserializationException e) {
2127               LOG.warn("Failed parse, skipping registering backup server", e);
2128               continue;
2129             }
2130             backupMasters.add(sn);
2131           }
2132         } catch (KeeperException e) {
2133           LOG.warn(this.zooKeeper.prefix("Unable to get information about " +
2134                    "backup servers"), e);
2135         }
2136       }
2137       Collections.sort(backupMasters, new Comparator<ServerName>() {
2138         @Override
2139         public int compare(ServerName s1, ServerName s2) {
2140           return s1.getServerName().compareTo(s2.getServerName());
2141         }});
2142     }
2143 
2144     String clusterId = fileSystemManager != null ?
2145       fileSystemManager.getClusterId().toString() : null;
2146     Map<String, RegionState> regionsInTransition = assignmentManager != null ?
2147       assignmentManager.getRegionStates().getRegionsInTransition() : null;
2148     String[] coprocessors = cpHost != null ? getMasterCoprocessors() : null;
2149     boolean balancerOn = loadBalancerTracker != null ?
2150       loadBalancerTracker.isBalancerOn() : false;
2151     Map<ServerName, ServerLoad> onlineServers = null;
2152     Set<ServerName> deadServers = null;
2153     if (serverManager != null) {
2154       deadServers = serverManager.getDeadServers().copyServerNames();
2155       onlineServers = serverManager.getOnlineServers();
2156     }
2157     return new ClusterStatus(VersionInfo.getVersion(), clusterId,
2158       onlineServers, deadServers, serverName, backupMasters,
2159       regionsInTransition, coprocessors, balancerOn);
2160   }
2161 
2162   /**
2163    * The set of loaded coprocessors is stored in a static set. Since it's
2164    * statically allocated, it does not require that HMaster's cpHost be
2165    * initialized prior to accessing it.
2166    * @return a String representation of the set of names of the loaded
2167    * coprocessors.
2168    */
2169   public static String getLoadedCoprocessors() {
2170     return CoprocessorHost.getLoadedCoprocessors().toString();
2171   }
2172 
2173   /**
2174    * @return timestamp in millis when HMaster was started.
2175    */
2176   public long getMasterStartTime() {
2177     return startcode;
2178   }
2179 
2180   /**
2181    * @return timestamp in millis when HMaster became the active master.
2182    */
2183   public long getMasterActiveTime() {
2184     return masterActiveTime;
2185   }
2186 
2187   public int getRegionServerInfoPort(final ServerName sn) {
2188     RegionServerInfo info = this.regionServerTracker.getRegionServerInfo(sn);
2189     if (info == null || info.getInfoPort() == 0) {
2190       return conf.getInt(HConstants.REGIONSERVER_INFO_PORT,
2191         HConstants.DEFAULT_REGIONSERVER_INFOPORT);
2192     }
2193     return info.getInfoPort();
2194   }
2195 
2196   public String getRegionServerVersion(final ServerName sn) {
2197     RegionServerInfo info = this.regionServerTracker.getRegionServerInfo(sn);
2198     if (info != null && info.hasVersionInfo()) {
2199       return info.getVersionInfo().getVersion();
2200     }
2201     return "Unknown";
2202   }
2203 
2204   /**
2205    * @return array of coprocessor SimpleNames.
2206    */
2207   public String[] getMasterCoprocessors() {
2208     Set<String> masterCoprocessors = getMasterCoprocessorHost().getCoprocessors();
2209     return masterCoprocessors.toArray(new String[masterCoprocessors.size()]);
2210   }
2211 
2212   @Override
2213   public void abort(final String msg, final Throwable t) {
2214     if (isAborted() || isStopped()) {
2215       return;
2216     }
2217     if (cpHost != null) {
2218       // HBASE-4014: dump a list of loaded coprocessors.
2219       LOG.fatal("Master server abort: loaded coprocessors are: " +
2220           getLoadedCoprocessors());
2221     }
2222     if (t != null) LOG.fatal(msg, t);
2223     stop(msg);
2224   }
2225 
2226   @Override
2227   public ZooKeeperWatcher getZooKeeper() {
2228     return zooKeeper;
2229   }
2230 
2231   @Override
2232   public MasterCoprocessorHost getMasterCoprocessorHost() {
2233     return cpHost;
2234   }
2235 
2236   @Override
2237   public MasterQuotaManager getMasterQuotaManager() {
2238     return quotaManager;
2239   }
2240 
2241   @Override
2242   public ProcedureExecutor<MasterProcedureEnv> getMasterProcedureExecutor() {
2243     return procedureExecutor;
2244   }
2245 
2246   @Override
2247   public ServerName getServerName() {
2248     return this.serverName;
2249   }
2250 
2251   @Override
2252   public AssignmentManager getAssignmentManager() {
2253     return this.assignmentManager;
2254   }
2255 
2256   public MemoryBoundedLogMessageBuffer getRegionServerFatalLogBuffer() {
2257     return rsFatals;
2258   }
2259 
2260   public void shutdown() throws IOException {
2261     if (cpHost != null) {
2262       cpHost.preShutdown();
2263     }
2264 
2265     if (this.serverManager != null) {
2266       this.serverManager.shutdownCluster();
2267     }
2268     if (this.clusterStatusTracker != null){
2269       try {
2270         this.clusterStatusTracker.setClusterDown();
2271       } catch (KeeperException e) {
2272         LOG.error("ZooKeeper exception trying to set cluster as down in ZK", e);
2273       }
2274     }
2275   }
2276 
2277   public void stopMaster() throws IOException {
2278     if (cpHost != null) {
2279       cpHost.preStopMaster();
2280     }
2281     stop("Stopped by " + Thread.currentThread().getName());
2282   }
2283 
2284   void checkServiceStarted() throws ServerNotRunningYetException {
2285     if (!serviceStarted) {
2286       throw new ServerNotRunningYetException("Server is not running yet");
2287     }
2288   }
2289 
2290   void checkInitialized() throws PleaseHoldException, ServerNotRunningYetException {
2291     checkServiceStarted();
2292     if (!isInitialized()) {
2293       throw new PleaseHoldException("Master is initializing");
2294     }
2295   }
2296 
2297   void checkNamespaceManagerReady() throws IOException {
2298     checkInitialized();
2299     if (tableNamespaceManager == null ||
2300         !tableNamespaceManager.isTableAvailableAndInitialized()) {
2301       throw new IOException("Table Namespace Manager not ready yet, try again later");
2302     }
2303   }
2304   /**
2305    * Report whether this master is currently the active master or not.
2306    * If not active master, we are parked on ZK waiting to become active.
2307    *
2308    * This method is used for testing.
2309    *
2310    * @return true if active master, false if not.
2311    */
2312   public boolean isActiveMaster() {
2313     return isActiveMaster;
2314   }
2315 
2316   /**
2317    * Report whether this master has completed with its initialization and is
2318    * ready.  If ready, the master is also the active master.  A standby master
2319    * is never ready.
2320    *
2321    * This method is used for testing.
2322    *
2323    * @return true if master is ready to go, false if not.
2324    */
2325   @Override
2326   public boolean isInitialized() {
2327     return initialized.isReady();
2328   }
2329 
2330   @VisibleForTesting
2331   public void setInitialized(boolean isInitialized) {
2332     procedureExecutor.getEnvironment().setEventReady(initialized, isInitialized);
2333   }
2334 
2335   public ProcedureEvent getInitializedEvent() {
2336     return initialized;
2337   }
2338 
2339   /**
2340    * ServerCrashProcessingEnabled is set false before completing assignMeta to prevent processing
2341    * of crashed servers.
2342    * @return true if assignMeta has completed;
2343    */
2344   @Override
2345   public boolean isServerCrashProcessingEnabled() {
2346     return serverCrashProcessingEnabled.isReady();
2347   }
2348 
2349   @VisibleForTesting
2350   public void setServerCrashProcessingEnabled(final boolean b) {
2351     procedureExecutor.getEnvironment().setEventReady(serverCrashProcessingEnabled, b);
2352   }
2353 
2354   public ProcedureEvent getServerCrashProcessingEnabledEvent() {
2355     return serverCrashProcessingEnabled;
2356   }
2357 
2358   /**
2359    * Report whether this master has started initialization and is about to do meta region assignment
2360    * @return true if master is in initialization &amp; about to assign hbase:meta regions
2361    */
2362   public boolean isInitializationStartsMetaRegionAssignment() {
2363     return this.initializationBeforeMetaAssignment;
2364   }
2365 
2366   public void assignRegion(HRegionInfo hri) {
2367     assignmentManager.assign(hri, true);
2368   }
2369 
2370   /**
2371    * Compute the average load across all region servers.
2372    * Currently, this uses a very naive computation - just uses the number of
2373    * regions being served, ignoring stats about number of requests.
2374    * @return the average load
2375    */
2376   public double getAverageLoad() {
2377     if (this.assignmentManager == null) {
2378       return 0;
2379     }
2380 
2381     RegionStates regionStates = this.assignmentManager.getRegionStates();
2382     if (regionStates == null) {
2383       return 0;
2384     }
2385     return regionStates.getAverageLoad();
2386   }
2387 
2388   @Override
2389   public boolean registerService(Service instance) {
2390     /*
2391      * No stacking of instances is allowed for a single service name
2392      */
2393     Descriptors.ServiceDescriptor serviceDesc = instance.getDescriptorForType();
2394     if (coprocessorServiceHandlers.containsKey(serviceDesc.getFullName())) {
2395       LOG.error("Coprocessor service "+serviceDesc.getFullName()+
2396           " already registered, rejecting request from "+instance
2397       );
2398       return false;
2399     }
2400 
2401     coprocessorServiceHandlers.put(serviceDesc.getFullName(), instance);
2402     if (LOG.isDebugEnabled()) {
2403       LOG.debug("Registered master coprocessor service: service="+serviceDesc.getFullName());
2404     }
2405     return true;
2406   }
2407 
2408   /**
2409    * Utility for constructing an instance of the passed HMaster class.
2410    * @param masterClass
2411    * @param conf
2412    * @return HMaster instance.
2413    */
2414   public static HMaster constructMaster(Class<? extends HMaster> masterClass,
2415       final Configuration conf, final CoordinatedStateManager cp)  {
2416     try {
2417       Constructor<? extends HMaster> c =
2418         masterClass.getConstructor(Configuration.class, CoordinatedStateManager.class);
2419       return c.newInstance(conf, cp);
2420     } catch(Exception e) {
2421       Throwable error = e;
2422       if (e instanceof InvocationTargetException &&
2423           ((InvocationTargetException)e).getTargetException() != null) {
2424         error = ((InvocationTargetException)e).getTargetException();
2425       }
2426       throw new RuntimeException("Failed construction of Master: " + masterClass.toString() + ". "
2427         , error);
2428     }
2429   }
2430 
2431   /**
2432    * @see org.apache.hadoop.hbase.master.HMasterCommandLine
2433    */
2434   public static void main(String [] args) {
2435     VersionInfo.logVersion();
2436     new HMasterCommandLine(HMaster.class).doMain(args);
2437   }
2438 
2439   public HFileCleaner getHFileCleaner() {
2440     return this.hfileCleaner;
2441   }
2442 
2443   /**
2444    * Exposed for TESTING!
2445    * @return the underlying snapshot manager
2446    */
2447   public SnapshotManager getSnapshotManagerForTesting() {
2448     return this.snapshotManager;
2449   }
2450 
2451   @Override
2452   public void createNamespace(NamespaceDescriptor descriptor) throws IOException {
2453     TableName.isLegalNamespaceName(Bytes.toBytes(descriptor.getName()));
2454     checkNamespaceManagerReady();
2455     if (cpHost != null) {
2456       if (cpHost.preCreateNamespace(descriptor)) {
2457         return;
2458       }
2459     }
2460     LOG.info(getClientIdAuditPrefix() + " creating " + descriptor);
2461     tableNamespaceManager.create(descriptor);
2462     if (cpHost != null) {
2463       cpHost.postCreateNamespace(descriptor);
2464     }
2465   }
2466 
2467   @Override
2468   public void modifyNamespace(NamespaceDescriptor descriptor) throws IOException {
2469     TableName.isLegalNamespaceName(Bytes.toBytes(descriptor.getName()));
2470     checkNamespaceManagerReady();
2471     if (cpHost != null) {
2472       if (cpHost.preModifyNamespace(descriptor)) {
2473         return;
2474       }
2475     }
2476     LOG.info(getClientIdAuditPrefix() + " modify " + descriptor);
2477     tableNamespaceManager.update(descriptor);
2478     if (cpHost != null) {
2479       cpHost.postModifyNamespace(descriptor);
2480     }
2481   }
2482 
2483   @Override
2484   public void deleteNamespace(String name) throws IOException {
2485     checkNamespaceManagerReady();
2486     if (cpHost != null) {
2487       if (cpHost.preDeleteNamespace(name)) {
2488         return;
2489       }
2490     }
2491     LOG.info(getClientIdAuditPrefix() + " delete " + name);
2492     tableNamespaceManager.remove(name);
2493     if (cpHost != null) {
2494       cpHost.postDeleteNamespace(name);
2495     }
2496   }
2497 
2498   /**
2499    * Ensure that the specified namespace exists, otherwise throws a NamespaceNotFoundException
2500    *
2501    * @param name the namespace to check
2502    * @throws IOException if the namespace manager is not ready yet.
2503    * @throws NamespaceNotFoundException if the namespace does not exists
2504    */
2505   protected void ensureNamespaceExists(final String name)
2506       throws IOException, NamespaceNotFoundException {
2507     checkNamespaceManagerReady();
2508     NamespaceDescriptor nsd = tableNamespaceManager.get(name);
2509     if (nsd == null) {
2510       throw new NamespaceNotFoundException(name);
2511     }
2512   }
2513 
2514   @Override
2515   public NamespaceDescriptor getNamespaceDescriptor(String name) throws IOException {
2516     checkNamespaceManagerReady();
2517 
2518     if (cpHost != null) {
2519       cpHost.preGetNamespaceDescriptor(name);
2520     }
2521 
2522     NamespaceDescriptor nsd = tableNamespaceManager.get(name);
2523     if (nsd == null) {
2524       throw new NamespaceNotFoundException(name);
2525     }
2526 
2527     if (cpHost != null) {
2528       cpHost.postGetNamespaceDescriptor(nsd);
2529     }
2530 
2531     return nsd;
2532   }
2533 
2534   @Override
2535   public List<NamespaceDescriptor> listNamespaceDescriptors() throws IOException {
2536     checkNamespaceManagerReady();
2537 
2538     final List<NamespaceDescriptor> descriptors = new ArrayList<NamespaceDescriptor>();
2539     boolean bypass = false;
2540     if (cpHost != null) {
2541       bypass = cpHost.preListNamespaceDescriptors(descriptors);
2542     }
2543 
2544     if (!bypass) {
2545       descriptors.addAll(tableNamespaceManager.list());
2546 
2547       if (cpHost != null) {
2548         cpHost.postListNamespaceDescriptors(descriptors);
2549       }
2550     }
2551     return descriptors;
2552   }
2553 
2554   @Override
2555   public boolean abortProcedure(final long procId, final boolean mayInterruptIfRunning)
2556       throws IOException {
2557     if (cpHost != null) {
2558       cpHost.preAbortProcedure(this.procedureExecutor, procId);
2559     }
2560 
2561     final boolean result = this.procedureExecutor.abort(procId, mayInterruptIfRunning);
2562 
2563     if (cpHost != null) {
2564       cpHost.postAbortProcedure();
2565     }
2566 
2567     return result;
2568   }
2569 
2570   @Override
2571   public List<ProcedureInfo> listProcedures() throws IOException {
2572     if (cpHost != null) {
2573       cpHost.preListProcedures();
2574     }
2575 
2576     final List<ProcedureInfo> procInfoList = this.procedureExecutor.listProcedures();
2577 
2578     if (cpHost != null) {
2579       cpHost.postListProcedures(procInfoList);
2580     }
2581 
2582     return procInfoList;
2583   }
2584 
2585   @Override
2586   public List<HTableDescriptor> listTableDescriptorsByNamespace(String name) throws IOException {
2587     ensureNamespaceExists(name);
2588     return listTableDescriptors(name, null, null, true);
2589   }
2590 
2591   @Override
2592   public List<TableName> listTableNamesByNamespace(String name) throws IOException {
2593     ensureNamespaceExists(name);
2594     return listTableNames(name, null, true);
2595   }
2596 
2597   /**
2598    * Returns the list of table descriptors that match the specified request
2599    *
2600    * @param namespace the namespace to query, or null if querying for all
2601    * @param regex The regular expression to match against, or null if querying for all
2602    * @param tableNameList the list of table names, or null if querying for all
2603    * @param includeSysTables False to match only against userspace tables
2604    * @return the list of table descriptors
2605    */
2606   public List<HTableDescriptor> listTableDescriptors(final String namespace, final String regex,
2607       final List<TableName> tableNameList, final boolean includeSysTables)
2608       throws IOException {
2609     final List<HTableDescriptor> descriptors = new ArrayList<HTableDescriptor>();
2610 
2611     boolean bypass = false;
2612     if (cpHost != null) {
2613       bypass = cpHost.preGetTableDescriptors(tableNameList, descriptors);
2614       // method required for AccessController.
2615       bypass |= cpHost.preGetTableDescriptors(tableNameList, descriptors, regex);
2616     }
2617 
2618     if (!bypass) {
2619       if (tableNameList == null || tableNameList.size() == 0) {
2620         // request for all TableDescriptors
2621         Collection<HTableDescriptor> htds;
2622         if (namespace != null && namespace.length() > 0) {
2623           htds = tableDescriptors.getByNamespace(namespace).values();
2624         } else {
2625           htds = tableDescriptors.getAll().values();
2626         }
2627 
2628         for (HTableDescriptor desc: htds) {
2629           if (includeSysTables || !desc.getTableName().isSystemTable()) {
2630             descriptors.add(desc);
2631           }
2632         }
2633       } else {
2634         for (TableName s: tableNameList) {
2635           HTableDescriptor desc = tableDescriptors.get(s);
2636           if (desc != null) {
2637             descriptors.add(desc);
2638           }
2639         }
2640       }
2641 
2642       // Retains only those matched by regular expression.
2643       if (regex != null) {
2644         filterTablesByRegex(descriptors, Pattern.compile(regex));
2645       }
2646 
2647       if (cpHost != null) {
2648         cpHost.postGetTableDescriptors(descriptors);
2649         // method required for AccessController.
2650         cpHost.postGetTableDescriptors(tableNameList, descriptors, regex);
2651       }
2652     }
2653     return descriptors;
2654   }
2655 
2656   /**
2657    * Returns the list of table names that match the specified request
2658    * @param regex The regular expression to match against, or null if querying for all
2659    * @param namespace the namespace to query, or null if querying for all
2660    * @param includeSysTables False to match only against userspace tables
2661    * @return the list of table names
2662    */
2663   public List<TableName> listTableNames(final String namespace, final String regex,
2664       final boolean includeSysTables) throws IOException {
2665     final List<HTableDescriptor> descriptors = new ArrayList<HTableDescriptor>();
2666 
2667     boolean bypass = false;
2668     if (cpHost != null) {
2669       bypass = cpHost.preGetTableNames(descriptors, regex);
2670     }
2671 
2672     if (!bypass) {
2673       // get all descriptors
2674       Collection<HTableDescriptor> htds;
2675       if (namespace != null && namespace.length() > 0) {
2676         htds = tableDescriptors.getByNamespace(namespace).values();
2677       } else {
2678         htds = tableDescriptors.getAll().values();
2679       }
2680 
2681       for (HTableDescriptor htd: htds) {
2682         if (includeSysTables || !htd.getTableName().isSystemTable()) {
2683           descriptors.add(htd);
2684         }
2685       }
2686 
2687       // Retains only those matched by regular expression.
2688       if (regex != null) {
2689         filterTablesByRegex(descriptors, Pattern.compile(regex));
2690       }
2691 
2692       if (cpHost != null) {
2693         cpHost.postGetTableNames(descriptors, regex);
2694       }
2695     }
2696 
2697     List<TableName> result = new ArrayList<TableName>(descriptors.size());
2698     for (HTableDescriptor htd: descriptors) {
2699       result.add(htd.getTableName());
2700     }
2701     return result;
2702   }
2703 
2704 
2705   /**
2706    * Removes the table descriptors that don't match the pattern.
2707    * @param descriptors list of table descriptors to filter
2708    * @param pattern the regex to use
2709    */
2710   private static void filterTablesByRegex(final Collection<HTableDescriptor> descriptors,
2711       final Pattern pattern) {
2712     final String defaultNS = NamespaceDescriptor.DEFAULT_NAMESPACE_NAME_STR;
2713     Iterator<HTableDescriptor> itr = descriptors.iterator();
2714     while (itr.hasNext()) {
2715       HTableDescriptor htd = itr.next();
2716       String tableName = htd.getTableName().getNameAsString();
2717       boolean matched = pattern.matcher(tableName).matches();
2718       if (!matched && htd.getTableName().getNamespaceAsString().equals(defaultNS)) {
2719         matched = pattern.matcher(defaultNS + TableName.NAMESPACE_DELIM + tableName).matches();
2720       }
2721       if (!matched) {
2722         itr.remove();
2723       }
2724     }
2725   }
2726 
2727   @Override
2728   public long getLastMajorCompactionTimestamp(TableName table) throws IOException {
2729     return getClusterStatus().getLastMajorCompactionTsForTable(table);
2730   }
2731 
2732   @Override
2733   public long getLastMajorCompactionTimestampForRegion(byte[] regionName) throws IOException {
2734     return getClusterStatus().getLastMajorCompactionTsForRegion(regionName);
2735   }
2736 
2737   /**
2738    * Queries the state of the {@link LoadBalancerTracker}. If the balancer is not initialized,
2739    * false is returned.
2740    *
2741    * @return The state of the load balancer, or false if the load balancer isn't defined.
2742    */
2743   public boolean isBalancerOn() {
2744     if (null == loadBalancerTracker) return false;
2745     return loadBalancerTracker.isBalancerOn();
2746   }
2747 
2748   /**
2749    * Queries the state of the {@link RegionNormalizerTracker}. If it's not initialized,
2750    * false is returned.
2751    */
2752    public boolean isNormalizerOn() {
2753     if (null == regionNormalizerTracker) {
2754       return false;
2755     }
2756     return regionNormalizerTracker.isNormalizerOn();
2757   }
2758 
2759   /**
2760    * Fetch the configured {@link LoadBalancer} class name. If none is set, a default is returned.
2761    *
2762    * @return The name of the {@link LoadBalancer} in use.
2763    */
2764   public String getLoadBalancerClassName() {
2765     return conf.get(HConstants.HBASE_MASTER_LOADBALANCER_CLASS, LoadBalancerFactory
2766         .getDefaultLoadBalancerClass().getName());
2767   }
2768 
2769   /**
2770    * @return RegionNormalizerTracker instance
2771    */
2772   public RegionNormalizerTracker getRegionNormalizerTracker() {
2773     return regionNormalizerTracker;
2774   }
2775 }