001/**
002 * Licensed to the Apache Software Foundation (ASF) under one
003 * or more contributor license agreements.  See the NOTICE file
004 * distributed with this work for additional information
005 * regarding copyright ownership.  The ASF licenses this file
006 * to you under the Apache License, Version 2.0 (the
007 * "License"); you may not use this file except in compliance
008 * with the License.  You may obtain a copy of the License at
009 *
010 *     http://www.apache.org/licenses/LICENSE-2.0
011 *
012 * Unless required by applicable law or agreed to in writing, software
013 * distributed under the License is distributed on an "AS IS" BASIS,
014 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
015 * See the License for the specific language governing permissions and
016 * limitations under the License.
017 */
018package org.apache.hadoop.hbase.master;
019
020import static org.apache.hadoop.hbase.HConstants.DEFAULT_HBASE_SPLIT_COORDINATED_BY_ZK;
021import static org.apache.hadoop.hbase.HConstants.HBASE_MASTER_LOGCLEANER_PLUGINS;
022import static org.apache.hadoop.hbase.HConstants.HBASE_SPLIT_WAL_COORDINATED_BY_ZK;
023
024import com.google.protobuf.Descriptors;
025import com.google.protobuf.Service;
026import java.io.IOException;
027import java.io.InterruptedIOException;
028import java.lang.reflect.Constructor;
029import java.lang.reflect.InvocationTargetException;
030import java.net.InetAddress;
031import java.net.InetSocketAddress;
032import java.net.UnknownHostException;
033import java.util.ArrayList;
034import java.util.Arrays;
035import java.util.Collection;
036import java.util.Collections;
037import java.util.Comparator;
038import java.util.EnumSet;
039import java.util.HashMap;
040import java.util.Iterator;
041import java.util.List;
042import java.util.Map;
043import java.util.Map.Entry;
044import java.util.Objects;
045import java.util.Optional;
046import java.util.Set;
047import java.util.concurrent.ExecutionException;
048import java.util.concurrent.Future;
049import java.util.concurrent.TimeUnit;
050import java.util.concurrent.TimeoutException;
051import java.util.concurrent.atomic.AtomicInteger;
052import java.util.function.Function;
053import java.util.regex.Pattern;
054import java.util.stream.Collectors;
055import javax.servlet.ServletException;
056import javax.servlet.http.HttpServlet;
057import javax.servlet.http.HttpServletRequest;
058import javax.servlet.http.HttpServletResponse;
059import org.apache.commons.lang3.StringUtils;
060import org.apache.hadoop.conf.Configuration;
061import org.apache.hadoop.fs.Path;
062import org.apache.hadoop.hbase.ChoreService;
063import org.apache.hadoop.hbase.ClusterId;
064import org.apache.hadoop.hbase.ClusterMetrics;
065import org.apache.hadoop.hbase.ClusterMetrics.Option;
066import org.apache.hadoop.hbase.ClusterMetricsBuilder;
067import org.apache.hadoop.hbase.DoNotRetryIOException;
068import org.apache.hadoop.hbase.HBaseIOException;
069import org.apache.hadoop.hbase.HBaseInterfaceAudience;
070import org.apache.hadoop.hbase.HConstants;
071import org.apache.hadoop.hbase.InvalidFamilyOperationException;
072import org.apache.hadoop.hbase.MasterNotRunningException;
073import org.apache.hadoop.hbase.MetaTableAccessor;
074import org.apache.hadoop.hbase.NamespaceDescriptor;
075import org.apache.hadoop.hbase.PleaseHoldException;
076import org.apache.hadoop.hbase.ReplicationPeerNotFoundException;
077import org.apache.hadoop.hbase.ServerName;
078import org.apache.hadoop.hbase.TableName;
079import org.apache.hadoop.hbase.TableNotDisabledException;
080import org.apache.hadoop.hbase.TableNotFoundException;
081import org.apache.hadoop.hbase.UnknownRegionException;
082import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
083import org.apache.hadoop.hbase.client.MasterSwitchType;
084import org.apache.hadoop.hbase.client.RegionInfo;
085import org.apache.hadoop.hbase.client.RegionInfoBuilder;
086import org.apache.hadoop.hbase.client.RegionStatesCount;
087import org.apache.hadoop.hbase.client.TableDescriptor;
088import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
089import org.apache.hadoop.hbase.client.TableState;
090import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
091import org.apache.hadoop.hbase.exceptions.DeserializationException;
092import org.apache.hadoop.hbase.executor.ExecutorType;
093import org.apache.hadoop.hbase.favored.FavoredNodesManager;
094import org.apache.hadoop.hbase.favored.FavoredNodesPromoter;
095import org.apache.hadoop.hbase.http.InfoServer;
096import org.apache.hadoop.hbase.ipc.CoprocessorRpcUtils;
097import org.apache.hadoop.hbase.ipc.RpcServer;
098import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
099import org.apache.hadoop.hbase.log.HBaseMarkers;
100import org.apache.hadoop.hbase.master.MasterRpcServices.BalanceSwitchMode;
101import org.apache.hadoop.hbase.master.assignment.AssignProcedure;
102import org.apache.hadoop.hbase.master.assignment.AssignmentManager;
103import org.apache.hadoop.hbase.master.assignment.MergeTableRegionsProcedure;
104import org.apache.hadoop.hbase.master.assignment.MoveRegionProcedure;
105import org.apache.hadoop.hbase.master.assignment.RegionStateNode;
106import org.apache.hadoop.hbase.master.assignment.RegionStates;
107import org.apache.hadoop.hbase.master.assignment.TransitRegionStateProcedure;
108import org.apache.hadoop.hbase.master.assignment.UnassignProcedure;
109import org.apache.hadoop.hbase.master.balancer.BalancerChore;
110import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer;
111import org.apache.hadoop.hbase.master.balancer.ClusterStatusChore;
112import org.apache.hadoop.hbase.master.balancer.LoadBalancerFactory;
113import org.apache.hadoop.hbase.master.cleaner.DirScanPool;
114import org.apache.hadoop.hbase.master.cleaner.HFileCleaner;
115import org.apache.hadoop.hbase.master.cleaner.LogCleaner;
116import org.apache.hadoop.hbase.master.cleaner.ReplicationBarrierCleaner;
117import org.apache.hadoop.hbase.master.cleaner.SnapshotCleanerChore;
118import org.apache.hadoop.hbase.master.locking.LockManager;
119import org.apache.hadoop.hbase.master.normalizer.NormalizationPlan;
120import org.apache.hadoop.hbase.master.normalizer.NormalizationPlan.PlanType;
121import org.apache.hadoop.hbase.master.normalizer.RegionNormalizer;
122import org.apache.hadoop.hbase.master.normalizer.RegionNormalizerChore;
123import org.apache.hadoop.hbase.master.normalizer.RegionNormalizerFactory;
124import org.apache.hadoop.hbase.master.procedure.CreateTableProcedure;
125import org.apache.hadoop.hbase.master.procedure.DeleteNamespaceProcedure;
126import org.apache.hadoop.hbase.master.procedure.DeleteTableProcedure;
127import org.apache.hadoop.hbase.master.procedure.DisableTableProcedure;
128import org.apache.hadoop.hbase.master.procedure.EnableTableProcedure;
129import org.apache.hadoop.hbase.master.procedure.InitMetaProcedure;
130import org.apache.hadoop.hbase.master.procedure.MasterProcedureConstants;
131import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
132import org.apache.hadoop.hbase.master.procedure.MasterProcedureScheduler;
133import org.apache.hadoop.hbase.master.procedure.MasterProcedureUtil;
134import org.apache.hadoop.hbase.master.procedure.MasterProcedureUtil.NonceProcedureRunnable;
135import org.apache.hadoop.hbase.master.procedure.ModifyTableProcedure;
136import org.apache.hadoop.hbase.master.procedure.ProcedurePrepareLatch;
137import org.apache.hadoop.hbase.master.procedure.ProcedureSyncWait;
138import org.apache.hadoop.hbase.master.procedure.RecoverMetaProcedure;
139import org.apache.hadoop.hbase.master.procedure.ReopenTableRegionsProcedure;
140import org.apache.hadoop.hbase.master.procedure.ServerCrashProcedure;
141import org.apache.hadoop.hbase.master.procedure.TruncateTableProcedure;
142import org.apache.hadoop.hbase.master.replication.AbstractPeerProcedure;
143import org.apache.hadoop.hbase.master.replication.AddPeerProcedure;
144import org.apache.hadoop.hbase.master.replication.DisablePeerProcedure;
145import org.apache.hadoop.hbase.master.replication.EnablePeerProcedure;
146import org.apache.hadoop.hbase.master.replication.RemovePeerProcedure;
147import org.apache.hadoop.hbase.master.replication.ReplicationPeerManager;
148import org.apache.hadoop.hbase.master.replication.SyncReplicationReplayWALManager;
149import org.apache.hadoop.hbase.master.replication.TransitPeerSyncReplicationStateProcedure;
150import org.apache.hadoop.hbase.master.replication.UpdatePeerConfigProcedure;
151import org.apache.hadoop.hbase.master.snapshot.SnapshotManager;
152import org.apache.hadoop.hbase.master.zksyncer.MasterAddressSyncer;
153import org.apache.hadoop.hbase.master.zksyncer.MetaLocationSyncer;
154import org.apache.hadoop.hbase.mob.MobConstants;
155import org.apache.hadoop.hbase.monitoring.MemoryBoundedLogMessageBuffer;
156import org.apache.hadoop.hbase.monitoring.MonitoredTask;
157import org.apache.hadoop.hbase.monitoring.TaskMonitor;
158import org.apache.hadoop.hbase.procedure.MasterProcedureManagerHost;
159import org.apache.hadoop.hbase.procedure.flush.MasterFlushTableProcedureManager;
160import org.apache.hadoop.hbase.procedure2.LockedResource;
161import org.apache.hadoop.hbase.procedure2.Procedure;
162import org.apache.hadoop.hbase.procedure2.ProcedureEvent;
163import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
164import org.apache.hadoop.hbase.procedure2.RemoteProcedureDispatcher.RemoteProcedure;
165import org.apache.hadoop.hbase.procedure2.RemoteProcedureException;
166import org.apache.hadoop.hbase.procedure2.store.ProcedureStore.ProcedureStoreListener;
167import org.apache.hadoop.hbase.procedure2.store.wal.WALProcedureStore;
168import org.apache.hadoop.hbase.quotas.MasterQuotaManager;
169import org.apache.hadoop.hbase.quotas.MasterQuotasObserver;
170import org.apache.hadoop.hbase.quotas.QuotaObserverChore;
171import org.apache.hadoop.hbase.quotas.QuotaTableUtil;
172import org.apache.hadoop.hbase.quotas.QuotaUtil;
173import org.apache.hadoop.hbase.quotas.SnapshotQuotaObserverChore;
174import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshot;
175import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshot.SpaceQuotaStatus;
176import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshotNotifier;
177import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshotNotifierFactory;
178import org.apache.hadoop.hbase.quotas.SpaceViolationPolicy;
179import org.apache.hadoop.hbase.regionserver.HRegionServer;
180import org.apache.hadoop.hbase.regionserver.RSRpcServices;
181import org.apache.hadoop.hbase.replication.ReplicationException;
182import org.apache.hadoop.hbase.replication.ReplicationLoadSource;
183import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
184import org.apache.hadoop.hbase.replication.ReplicationPeerDescription;
185import org.apache.hadoop.hbase.replication.ReplicationUtils;
186import org.apache.hadoop.hbase.replication.SyncReplicationState;
187import org.apache.hadoop.hbase.replication.master.ReplicationHFileCleaner;
188import org.apache.hadoop.hbase.replication.master.ReplicationLogCleaner;
189import org.apache.hadoop.hbase.replication.master.ReplicationPeerConfigUpgrader;
190import org.apache.hadoop.hbase.replication.regionserver.ReplicationStatus;
191import org.apache.hadoop.hbase.security.AccessDeniedException;
192import org.apache.hadoop.hbase.security.SecurityConstants;
193import org.apache.hadoop.hbase.security.UserProvider;
194import org.apache.hadoop.hbase.trace.TraceUtil;
195import org.apache.hadoop.hbase.util.Addressing;
196import org.apache.hadoop.hbase.util.Bytes;
197import org.apache.hadoop.hbase.util.FutureUtils;
198import org.apache.hadoop.hbase.util.HBaseFsck;
199import org.apache.hadoop.hbase.util.HFileArchiveUtil;
200import org.apache.hadoop.hbase.util.HasThread;
201import org.apache.hadoop.hbase.util.IdLock;
202import org.apache.hadoop.hbase.util.ModifyRegionUtils;
203import org.apache.hadoop.hbase.util.Pair;
204import org.apache.hadoop.hbase.util.RetryCounter;
205import org.apache.hadoop.hbase.util.RetryCounterFactory;
206import org.apache.hadoop.hbase.util.TableDescriptorChecker;
207import org.apache.hadoop.hbase.util.Threads;
208import org.apache.hadoop.hbase.util.VersionInfo;
209import org.apache.hadoop.hbase.zookeeper.LoadBalancerTracker;
210import org.apache.hadoop.hbase.zookeeper.MasterAddressTracker;
211import org.apache.hadoop.hbase.zookeeper.RegionNormalizerTracker;
212import org.apache.hadoop.hbase.zookeeper.SnapshotCleanupTracker;
213import org.apache.hadoop.hbase.zookeeper.ZKClusterId;
214import org.apache.hadoop.hbase.zookeeper.ZKUtil;
215import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
216import org.apache.hadoop.hbase.zookeeper.ZNodePaths;
217import org.apache.yetus.audience.InterfaceAudience;
218import org.apache.zookeeper.KeeperException;
219import org.eclipse.jetty.server.Server;
220import org.eclipse.jetty.server.ServerConnector;
221import org.eclipse.jetty.servlet.ServletHolder;
222import org.eclipse.jetty.webapp.WebAppContext;
223import org.slf4j.Logger;
224import org.slf4j.LoggerFactory;
225
226import org.apache.hbase.thirdparty.com.google.common.annotations.VisibleForTesting;
227import org.apache.hbase.thirdparty.com.google.common.collect.ImmutableSet;
228import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
229import org.apache.hbase.thirdparty.com.google.common.collect.Maps;
230
231import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
232import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
233import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetRegionInfoResponse.CompactionState;
234import org.apache.hadoop.hbase.shaded.protobuf.generated.SnapshotProtos.SnapshotDescription;
235
236/**
237 * HMaster is the "master server" for HBase. An HBase cluster has one active
238 * master.  If many masters are started, all compete.  Whichever wins goes on to
239 * run the cluster.  All others park themselves in their constructor until
240 * master or cluster shutdown or until the active master loses its lease in
241 * zookeeper.  Thereafter, all running master jostle to take over master role.
242 *
243 * <p>The Master can be asked shutdown the cluster. See {@link #shutdown()}.  In
244 * this case it will tell all regionservers to go down and then wait on them
245 * all reporting in that they are down.  This master will then shut itself down.
246 *
247 * <p>You can also shutdown just this master.  Call {@link #stopMaster()}.
248 *
249 * @see org.apache.zookeeper.Watcher
250 */
251@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS)
252@SuppressWarnings("deprecation")
253public class HMaster extends HRegionServer implements MasterServices {
254  private static Logger LOG = LoggerFactory.getLogger(HMaster.class);
255
256  /**
257   * Protection against zombie master. Started once Master accepts active responsibility and
258   * starts taking over responsibilities. Allows a finite time window before giving up ownership.
259   */
260  private static class InitializationMonitor extends HasThread {
261    /** The amount of time in milliseconds to sleep before checking initialization status. */
262    public static final String TIMEOUT_KEY = "hbase.master.initializationmonitor.timeout";
263    public static final long TIMEOUT_DEFAULT = TimeUnit.MILLISECONDS.convert(15, TimeUnit.MINUTES);
264
265    /**
266     * When timeout expired and initialization has not complete, call {@link System#exit(int)} when
267     * true, do nothing otherwise.
268     */
269    public static final String HALT_KEY = "hbase.master.initializationmonitor.haltontimeout";
270    public static final boolean HALT_DEFAULT = false;
271
272    private final HMaster master;
273    private final long timeout;
274    private final boolean haltOnTimeout;
275
276    /** Creates a Thread that monitors the {@link #isInitialized()} state. */
277    InitializationMonitor(HMaster master) {
278      super("MasterInitializationMonitor");
279      this.master = master;
280      this.timeout = master.getConfiguration().getLong(TIMEOUT_KEY, TIMEOUT_DEFAULT);
281      this.haltOnTimeout = master.getConfiguration().getBoolean(HALT_KEY, HALT_DEFAULT);
282      this.setDaemon(true);
283    }
284
285    @Override
286    public void run() {
287      try {
288        while (!master.isStopped() && master.isActiveMaster()) {
289          Thread.sleep(timeout);
290          if (master.isInitialized()) {
291            LOG.debug("Initialization completed within allotted tolerance. Monitor exiting.");
292          } else {
293            LOG.error("Master failed to complete initialization after " + timeout + "ms. Please"
294                + " consider submitting a bug report including a thread dump of this process.");
295            if (haltOnTimeout) {
296              LOG.error("Zombie Master exiting. Thread dump to stdout");
297              Threads.printThreadInfo(System.out, "Zombie HMaster");
298              System.exit(-1);
299            }
300          }
301        }
302      } catch (InterruptedException ie) {
303        LOG.trace("InitMonitor thread interrupted. Existing.");
304      }
305    }
306  }
307
308  // MASTER is name of the webapp and the attribute name used stuffing this
309  //instance into web context.
310  public static final String MASTER = "master";
311
312  // Manager and zk listener for master election
313  private final ActiveMasterManager activeMasterManager;
314  // Region server tracker
315  private RegionServerTracker regionServerTracker;
316  // Draining region server tracker
317  private DrainingServerTracker drainingServerTracker;
318  // Tracker for load balancer state
319  LoadBalancerTracker loadBalancerTracker;
320  // Tracker for meta location, if any client ZK quorum specified
321  MetaLocationSyncer metaLocationSyncer;
322  // Tracker for active master location, if any client ZK quorum specified
323  MasterAddressSyncer masterAddressSyncer;
324  // Tracker for auto snapshot cleanup state
325  SnapshotCleanupTracker snapshotCleanupTracker;
326
327  // Tracker for split and merge state
328  private SplitOrMergeTracker splitOrMergeTracker;
329
330  // Tracker for region normalizer state
331  private RegionNormalizerTracker regionNormalizerTracker;
332
333  private ClusterSchemaService clusterSchemaService;
334
335  public static final String HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS =
336    "hbase.master.wait.on.service.seconds";
337  public static final int DEFAULT_HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS = 5 * 60;
338
339  // Metrics for the HMaster
340  final MetricsMaster metricsMaster;
341  // file system manager for the master FS operations
342  private MasterFileSystem fileSystemManager;
343  private MasterWalManager walManager;
344
345  // manager to manage procedure-based WAL splitting, can be null if current
346  // is zk-based WAL splitting. SplitWALManager will replace SplitLogManager
347  // and MasterWalManager, which means zk-based WAL splitting code will be
348  // useless after we switch to the procedure-based one. our eventual goal
349  // is to remove all the zk-based WAL splitting code.
350  private SplitWALManager splitWALManager;
351
352  // server manager to deal with region server info
353  private volatile ServerManager serverManager;
354
355  // manager of assignment nodes in zookeeper
356  private AssignmentManager assignmentManager;
357
358  // manager of replication
359  private ReplicationPeerManager replicationPeerManager;
360
361  private SyncReplicationReplayWALManager syncReplicationReplayWALManager;
362
363  // buffer for "fatal error" notices from region servers
364  // in the cluster. This is only used for assisting
365  // operations/debugging.
366  MemoryBoundedLogMessageBuffer rsFatals;
367
368  // flag set after we become the active master (used for testing)
369  private volatile boolean activeMaster = false;
370
371  // flag set after we complete initialization once active
372  private final ProcedureEvent<?> initialized = new ProcedureEvent<>("master initialized");
373
374  // flag set after master services are started,
375  // initialization may have not completed yet.
376  volatile boolean serviceStarted = false;
377
378  // Maximum time we should run balancer for
379  private final int maxBlancingTime;
380  // Maximum percent of regions in transition when balancing
381  private final double maxRitPercent;
382
383  private final LockManager lockManager = new LockManager(this);
384
385  private LoadBalancer balancer;
386  private RegionNormalizer normalizer;
387  private BalancerChore balancerChore;
388  private RegionNormalizerChore normalizerChore;
389  private ClusterStatusChore clusterStatusChore;
390  private ClusterStatusPublisher clusterStatusPublisherChore = null;
391  private SnapshotCleanerChore snapshotCleanerChore = null;
392
393  private HbckChore hbckChore;
394  CatalogJanitor catalogJanitorChore;
395  private DirScanPool cleanerPool;
396  private LogCleaner logCleaner;
397  private HFileCleaner hfileCleaner;
398  private ReplicationBarrierCleaner replicationBarrierCleaner;
399  private ExpiredMobFileCleanerChore expiredMobFileCleanerChore;
400  private MobCompactionChore mobCompactChore;
401  private MasterMobCompactionThread mobCompactThread;
402  // used to synchronize the mobCompactionStates
403  private final IdLock mobCompactionLock = new IdLock();
404  // save the information of mob compactions in tables.
405  // the key is table name, the value is the number of compactions in that table.
406  private Map<TableName, AtomicInteger> mobCompactionStates = Maps.newConcurrentMap();
407
408  MasterCoprocessorHost cpHost;
409
410  private final boolean preLoadTableDescriptors;
411
412  // Time stamps for when a hmaster became active
413  private long masterActiveTime;
414
415  // Time stamp for when HMaster finishes becoming Active Master
416  private long masterFinishedInitializationTime;
417
418  Map<String, Service> coprocessorServiceHandlers = Maps.newHashMap();
419
420  // monitor for snapshot of hbase tables
421  SnapshotManager snapshotManager;
422  // monitor for distributed procedures
423  private MasterProcedureManagerHost mpmHost;
424
425  private RegionsRecoveryChore regionsRecoveryChore = null;
426
427  private RegionsRecoveryConfigManager regionsRecoveryConfigManager = null;
428  // it is assigned after 'initialized' guard set to true, so should be volatile
429  private volatile MasterQuotaManager quotaManager;
430  private SpaceQuotaSnapshotNotifier spaceQuotaSnapshotNotifier;
431  private QuotaObserverChore quotaObserverChore;
432  private SnapshotQuotaObserverChore snapshotQuotaChore;
433
434  private ProcedureExecutor<MasterProcedureEnv> procedureExecutor;
435  private WALProcedureStore procedureStore;
436
437  // handle table states
438  private TableStateManager tableStateManager;
439
440  private long splitPlanCount;
441  private long mergePlanCount;
442
443  /* Handle favored nodes information */
444  private FavoredNodesManager favoredNodesManager;
445
446  /** jetty server for master to redirect requests to regionserver infoServer */
447  private Server masterJettyServer;
448
449  // Determine if we should do normal startup or minimal "single-user" mode with no region
450  // servers and no user tables. Useful for repair and recovery of hbase:meta
451  private final boolean maintenanceMode;
452  static final String MAINTENANCE_MODE = "hbase.master.maintenance_mode";
453
454  // Cached clusterId on stand by masters to serve clusterID requests from clients.
455  private final CachedClusterId cachedClusterId;
456
457  public static class RedirectServlet extends HttpServlet {
458    private static final long serialVersionUID = 2894774810058302473L;
459    private final int regionServerInfoPort;
460    private final String regionServerHostname;
461
462    /**
463     * @param infoServer that we're trying to send all requests to
464     * @param hostname may be null. if given, will be used for redirects instead of host from client.
465     */
466    public RedirectServlet(InfoServer infoServer, String hostname) {
467       regionServerInfoPort = infoServer.getPort();
468       regionServerHostname = hostname;
469    }
470
471    @Override
472    public void doGet(HttpServletRequest request,
473        HttpServletResponse response) throws ServletException, IOException {
474      String redirectHost = regionServerHostname;
475      if(redirectHost == null) {
476        redirectHost = request.getServerName();
477        if(!Addressing.isLocalAddress(InetAddress.getByName(redirectHost))) {
478          LOG.warn("Couldn't resolve '" + redirectHost + "' as an address local to this node and '" +
479              MASTER_HOSTNAME_KEY + "' is not set; client will get an HTTP 400 response. If " +
480              "your HBase deployment relies on client accessible names that the region server process " +
481              "can't resolve locally, then you should set the previously mentioned configuration variable " +
482              "to an appropriate hostname.");
483          // no sending client provided input back to the client, so the goal host is just in the logs.
484          response.sendError(400, "Request was to a host that I can't resolve for any of the network interfaces on " +
485              "this node. If this is due to an intermediary such as an HTTP load balancer or other proxy, your HBase " +
486              "administrator can set '" + MASTER_HOSTNAME_KEY + "' to point to the correct hostname.");
487          return;
488        }
489      }
490      // TODO this scheme should come from looking at the scheme registered in the infoserver's http server for the
491      // host and port we're using, but it's buried way too deep to do that ATM.
492      String redirectUrl = request.getScheme() + "://"
493        + redirectHost + ":" + regionServerInfoPort
494        + request.getRequestURI();
495      response.sendRedirect(redirectUrl);
496    }
497  }
498
499  /**
500   * Initializes the HMaster. The steps are as follows:
501   * <p>
502   * <ol>
503   * <li>Initialize the local HRegionServer
504   * <li>Start the ActiveMasterManager.
505   * </ol>
506   * <p>
507   * Remaining steps of initialization occur in
508   * #finishActiveMasterInitialization(MonitoredTask) after
509   * the master becomes the active one.
510   */
511  public HMaster(final Configuration conf)
512      throws IOException, KeeperException {
513    super(conf);
514    TraceUtil.initTracer(conf);
515    try {
516      if (conf.getBoolean(MAINTENANCE_MODE, false)) {
517        LOG.info("Detected {}=true via configuration.", MAINTENANCE_MODE);
518        maintenanceMode = true;
519      } else if (Boolean.getBoolean(MAINTENANCE_MODE)) {
520        LOG.info("Detected {}=true via environment variables.", MAINTENANCE_MODE);
521        maintenanceMode = true;
522      } else {
523        maintenanceMode = false;
524      }
525
526      this.rsFatals = new MemoryBoundedLogMessageBuffer(
527          conf.getLong("hbase.master.buffer.for.rs.fatals", 1 * 1024 * 1024));
528      LOG.info("hbase.rootdir={}, hbase.cluster.distributed={}", getDataRootDir(),
529          this.conf.getBoolean(HConstants.CLUSTER_DISTRIBUTED, false));
530
531      // Disable usage of meta replicas in the master
532      this.conf.setBoolean(HConstants.USE_META_REPLICAS, false);
533
534      decorateMasterConfiguration(this.conf);
535
536      // Hack! Maps DFSClient => Master for logs.  HDFS made this
537      // config param for task trackers, but we can piggyback off of it.
538      if (this.conf.get("mapreduce.task.attempt.id") == null) {
539        this.conf.set("mapreduce.task.attempt.id", "hb_m_" + this.serverName.toString());
540      }
541
542      this.metricsMaster = new MetricsMaster(new MetricsMasterWrapperImpl(this));
543
544      // preload table descriptor at startup
545      this.preLoadTableDescriptors = conf.getBoolean("hbase.master.preload.tabledescriptors", true);
546
547      this.maxBlancingTime = getMaxBalancingTime();
548      this.maxRitPercent = conf.getDouble(HConstants.HBASE_MASTER_BALANCER_MAX_RIT_PERCENT,
549          HConstants.DEFAULT_HBASE_MASTER_BALANCER_MAX_RIT_PERCENT);
550
551      // Do we publish the status?
552
553      boolean shouldPublish = conf.getBoolean(HConstants.STATUS_PUBLISHED,
554          HConstants.STATUS_PUBLISHED_DEFAULT);
555      Class<? extends ClusterStatusPublisher.Publisher> publisherClass =
556          conf.getClass(ClusterStatusPublisher.STATUS_PUBLISHER_CLASS,
557              ClusterStatusPublisher.DEFAULT_STATUS_PUBLISHER_CLASS,
558              ClusterStatusPublisher.Publisher.class);
559
560      if (shouldPublish) {
561        if (publisherClass == null) {
562          LOG.warn(HConstants.STATUS_PUBLISHED + " is true, but " +
563              ClusterStatusPublisher.DEFAULT_STATUS_PUBLISHER_CLASS +
564              " is not set - not publishing status");
565        } else {
566          clusterStatusPublisherChore = new ClusterStatusPublisher(this, conf, publisherClass);
567          getChoreService().scheduleChore(clusterStatusPublisherChore);
568        }
569      }
570
571      // Some unit tests don't need a cluster, so no zookeeper at all
572      if (!conf.getBoolean("hbase.testing.nocluster", false)) {
573        this.activeMasterManager = new ActiveMasterManager(zooKeeper, this.serverName, this);
574      } else {
575        this.activeMasterManager = null;
576      }
577      cachedClusterId = new CachedClusterId(conf);
578    } catch (Throwable t) {
579      // Make sure we log the exception. HMaster is often started via reflection and the
580      // cause of failed startup is lost.
581      LOG.error("Failed construction of Master", t);
582      throw t;
583    }
584  }
585
586  @Override
587  protected String getUseThisHostnameInstead(Configuration conf) {
588    return conf.get(MASTER_HOSTNAME_KEY);
589  }
590
591  // Main run loop. Calls through to the regionserver run loop AFTER becoming active Master; will
592  // block in here until then.
593  @Override
594  public void run() {
595    try {
596      if (!conf.getBoolean("hbase.testing.nocluster", false)) {
597        Threads.setDaemonThreadRunning(new Thread(() -> {
598          try {
599            int infoPort = putUpJettyServer();
600            startActiveMasterManager(infoPort);
601          } catch (Throwable t) {
602            // Make sure we log the exception.
603            String error = "Failed to become Active Master";
604            LOG.error(error, t);
605            // Abort should have been called already.
606            if (!isAborted()) {
607              abort(error, t);
608            }
609          }
610        }), getName() + ":becomeActiveMaster");
611      }
612      // Fall in here even if we have been aborted. Need to run the shutdown services and
613      // the super run call will do this for us.
614      super.run();
615    } finally {
616      if (this.clusterSchemaService != null) {
617        // If on way out, then we are no longer active master.
618        this.clusterSchemaService.stopAsync();
619        try {
620          this.clusterSchemaService.awaitTerminated(
621              getConfiguration().getInt(HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS,
622              DEFAULT_HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS), TimeUnit.SECONDS);
623        } catch (TimeoutException te) {
624          LOG.warn("Failed shutdown of clusterSchemaService", te);
625        }
626      }
627      this.activeMaster = false;
628    }
629  }
630
631  // return the actual infoPort, -1 means disable info server.
632  private int putUpJettyServer() throws IOException {
633    if (!conf.getBoolean("hbase.master.infoserver.redirect", true)) {
634      return -1;
635    }
636    final int infoPort = conf.getInt("hbase.master.info.port.orig",
637      HConstants.DEFAULT_MASTER_INFOPORT);
638    // -1 is for disabling info server, so no redirecting
639    if (infoPort < 0 || infoServer == null) {
640      return -1;
641    }
642    if(infoPort == infoServer.getPort()) {
643      return infoPort;
644    }
645    final String addr = conf.get("hbase.master.info.bindAddress", "0.0.0.0");
646    if (!Addressing.isLocalAddress(InetAddress.getByName(addr))) {
647      String msg =
648          "Failed to start redirecting jetty server. Address " + addr
649              + " does not belong to this host. Correct configuration parameter: "
650              + "hbase.master.info.bindAddress";
651      LOG.error(msg);
652      throw new IOException(msg);
653    }
654
655    // TODO I'm pretty sure we could just add another binding to the InfoServer run by
656    // the RegionServer and have it run the RedirectServlet instead of standing up
657    // a second entire stack here.
658    masterJettyServer = new Server();
659    final ServerConnector connector = new ServerConnector(masterJettyServer);
660    connector.setHost(addr);
661    connector.setPort(infoPort);
662    masterJettyServer.addConnector(connector);
663    masterJettyServer.setStopAtShutdown(true);
664
665    final String redirectHostname =
666        StringUtils.isBlank(useThisHostnameInstead) ? null : useThisHostnameInstead;
667
668    final RedirectServlet redirect = new RedirectServlet(infoServer, redirectHostname);
669    final WebAppContext context = new WebAppContext(null, "/", null, null, null, null, WebAppContext.NO_SESSIONS);
670    context.addServlet(new ServletHolder(redirect), "/*");
671    context.setServer(masterJettyServer);
672
673    try {
674      masterJettyServer.start();
675    } catch (Exception e) {
676      throw new IOException("Failed to start redirecting jetty server", e);
677    }
678    return connector.getLocalPort();
679  }
680
681  @Override
682  protected Function<TableDescriptorBuilder, TableDescriptorBuilder> getMetaTableObserver() {
683    return builder -> builder.setRegionReplication(conf.getInt(HConstants.META_REPLICAS_NUM, HConstants.DEFAULT_META_REPLICA_NUM));
684  }
685  /**
686   * For compatibility, if failed with regionserver credentials, try the master one
687   */
688  @Override
689  protected void login(UserProvider user, String host) throws IOException {
690    try {
691      super.login(user, host);
692    } catch (IOException ie) {
693      user.login(SecurityConstants.MASTER_KRB_KEYTAB_FILE,
694              SecurityConstants.MASTER_KRB_PRINCIPAL, host);
695    }
696  }
697
698  /**
699   * If configured to put regions on active master,
700   * wait till a backup master becomes active.
701   * Otherwise, loop till the server is stopped or aborted.
702   */
703  @Override
704  protected void waitForMasterActive(){
705    if (maintenanceMode) {
706      return;
707    }
708    boolean tablesOnMaster = LoadBalancer.isTablesOnMaster(conf);
709    while (!(tablesOnMaster && activeMaster) && !isStopped() && !isAborted()) {
710      sleeper.sleep();
711    }
712  }
713
714  @VisibleForTesting
715  public MasterRpcServices getMasterRpcServices() {
716    return (MasterRpcServices)rpcServices;
717  }
718
719  public boolean balanceSwitch(final boolean b) throws IOException {
720    return getMasterRpcServices().switchBalancer(b, BalanceSwitchMode.ASYNC);
721  }
722
723  @Override
724  protected String getProcessName() {
725    return MASTER;
726  }
727
728  @Override
729  protected boolean canCreateBaseZNode() {
730    return true;
731  }
732
733  @Override
734  protected boolean canUpdateTableDescriptor() {
735    return true;
736  }
737
738  @Override
739  protected RSRpcServices createRpcServices() throws IOException {
740    return new MasterRpcServices(this);
741  }
742
743  @Override
744  protected void configureInfoServer() {
745    infoServer.addServlet("master-status", "/master-status", MasterStatusServlet.class);
746    infoServer.setAttribute(MASTER, this);
747    if (LoadBalancer.isTablesOnMaster(conf)) {
748      super.configureInfoServer();
749    }
750  }
751
752  @Override
753  protected Class<? extends HttpServlet> getDumpServlet() {
754    return MasterDumpServlet.class;
755  }
756
757  @Override
758  public MetricsMaster getMasterMetrics() {
759    return metricsMaster;
760  }
761
762  /**
763   * <p>
764   * Initialize all ZK based system trackers. But do not include {@link RegionServerTracker}, it
765   * should have already been initialized along with {@link ServerManager}.
766   * </p>
767   * <p>
768   * Will be overridden in tests.
769   * </p>
770   */
771  @VisibleForTesting
772  protected void initializeZKBasedSystemTrackers()
773      throws IOException, InterruptedException, KeeperException, ReplicationException {
774    this.balancer = LoadBalancerFactory.getLoadBalancer(conf);
775    this.normalizer = RegionNormalizerFactory.getRegionNormalizer(conf);
776    this.normalizer.setMasterServices(this);
777    this.normalizer.setMasterRpcServices((MasterRpcServices)rpcServices);
778    this.loadBalancerTracker = new LoadBalancerTracker(zooKeeper, this);
779    this.loadBalancerTracker.start();
780
781    this.regionNormalizerTracker = new RegionNormalizerTracker(zooKeeper, this);
782    this.regionNormalizerTracker.start();
783
784    this.splitOrMergeTracker = new SplitOrMergeTracker(zooKeeper, conf, this);
785    this.splitOrMergeTracker.start();
786
787    this.replicationPeerManager = ReplicationPeerManager.create(zooKeeper, conf);
788
789    this.drainingServerTracker = new DrainingServerTracker(zooKeeper, this, this.serverManager);
790    this.drainingServerTracker.start();
791
792    this.snapshotCleanupTracker = new SnapshotCleanupTracker(zooKeeper, this);
793    this.snapshotCleanupTracker.start();
794
795    String clientQuorumServers = conf.get(HConstants.CLIENT_ZOOKEEPER_QUORUM);
796    boolean clientZkObserverMode = conf.getBoolean(HConstants.CLIENT_ZOOKEEPER_OBSERVER_MODE,
797      HConstants.DEFAULT_CLIENT_ZOOKEEPER_OBSERVER_MODE);
798    if (clientQuorumServers != null && !clientZkObserverMode) {
799      // we need to take care of the ZK information synchronization
800      // if given client ZK are not observer nodes
801      ZKWatcher clientZkWatcher = new ZKWatcher(conf,
802          getProcessName() + ":" + rpcServices.getSocketAddress().getPort() + "-clientZK", this,
803          false, true);
804      this.metaLocationSyncer = new MetaLocationSyncer(zooKeeper, clientZkWatcher, this);
805      this.metaLocationSyncer.start();
806      this.masterAddressSyncer = new MasterAddressSyncer(zooKeeper, clientZkWatcher, this);
807      this.masterAddressSyncer.start();
808      // set cluster id is a one-go effort
809      ZKClusterId.setClusterId(clientZkWatcher, fileSystemManager.getClusterId());
810    }
811
812    // Set the cluster as up.  If new RSs, they'll be waiting on this before
813    // going ahead with their startup.
814    boolean wasUp = this.clusterStatusTracker.isClusterUp();
815    if (!wasUp) this.clusterStatusTracker.setClusterUp();
816
817    LOG.info("Active/primary master=" + this.serverName +
818        ", sessionid=0x" +
819        Long.toHexString(this.zooKeeper.getRecoverableZooKeeper().getSessionId()) +
820        ", setting cluster-up flag (Was=" + wasUp + ")");
821
822    // create/initialize the snapshot manager and other procedure managers
823    this.snapshotManager = new SnapshotManager();
824    this.mpmHost = new MasterProcedureManagerHost();
825    this.mpmHost.register(this.snapshotManager);
826    this.mpmHost.register(new MasterFlushTableProcedureManager());
827    this.mpmHost.loadProcedures(conf);
828    this.mpmHost.initialize(this, this.metricsMaster);
829  }
830
831  private static final ImmutableSet<Class<? extends Procedure>> UNSUPPORTED_PROCEDURES =
832    ImmutableSet.of(RecoverMetaProcedure.class, AssignProcedure.class, UnassignProcedure.class,
833      MoveRegionProcedure.class);
834
835  /**
836   * In HBASE-20811, we have introduced a new TRSP to assign/unassign/move regions, and it is
837   * incompatible with the old AssignProcedure/UnassignProcedure/MoveRegionProcedure. So we need to
838   * make sure that there are none these procedures when upgrading. If there are, the master will
839   * quit, you need to go back to the old version to finish these procedures first before upgrading.
840   */
841  private void checkUnsupportedProcedure(
842      Map<Class<? extends Procedure>, List<Procedure<MasterProcedureEnv>>> procsByType)
843      throws HBaseIOException {
844    // Confirm that we do not have unfinished assign/unassign related procedures. It is not easy to
845    // support both the old assign/unassign procedures and the new TransitRegionStateProcedure as
846    // there will be conflict in the code for AM. We should finish all these procedures before
847    // upgrading.
848    for (Class<? extends Procedure> clazz : UNSUPPORTED_PROCEDURES) {
849      List<Procedure<MasterProcedureEnv>> procs = procsByType.get(clazz);
850      if (procs != null) {
851        LOG.error(
852          "Unsupported procedure type {} found, please rollback your master to the old" +
853            " version to finish them, and then try to upgrade again. The full procedure list: {}",
854          clazz, procs);
855        throw new HBaseIOException("Unsupported procedure type " + clazz + " found");
856      }
857    }
858    // A special check for SCP, as we do not support RecoverMetaProcedure any more so we need to
859    // make sure that no one will try to schedule it but SCP does have a state which will schedule
860    // it.
861    if (procsByType.getOrDefault(ServerCrashProcedure.class, Collections.emptyList()).stream()
862      .map(p -> (ServerCrashProcedure) p).anyMatch(ServerCrashProcedure::isInRecoverMetaState)) {
863      LOG.error("At least one ServerCrashProcedure is going to schedule a RecoverMetaProcedure," +
864        " which is not supported any more. Please rollback your master to the old version to" +
865        " finish them, and then try to upgrade again.");
866      throw new HBaseIOException("Unsupported procedure state found for ServerCrashProcedure");
867    }
868  }
869
870  // Will be overriden in test to inject customized AssignmentManager
871  @VisibleForTesting
872  protected AssignmentManager createAssignmentManager(MasterServices master) {
873    return new AssignmentManager(master);
874  }
875
876  /**
877   * Finish initialization of HMaster after becoming the primary master.
878   * <p/>
879   * The startup order is a bit complicated but very important, do not change it unless you know
880   * what you are doing.
881   * <ol>
882   * <li>Initialize file system based components - file system manager, wal manager, table
883   * descriptors, etc</li>
884   * <li>Publish cluster id</li>
885   * <li>Here comes the most complicated part - initialize server manager, assignment manager and
886   * region server tracker
887   * <ol type='i'>
888   * <li>Create server manager</li>
889   * <li>Create procedure executor, load the procedures, but do not start workers. We will start it
890   * later after we finish scheduling SCPs to avoid scheduling duplicated SCPs for the same
891   * server</li>
892   * <li>Create assignment manager and start it, load the meta region state, but do not load data
893   * from meta region</li>
894   * <li>Start region server tracker, construct the online servers set and find out dead servers and
895   * schedule SCP for them. The online servers will be constructed by scanning zk, and we will also
896   * scan the wal directory to find out possible live region servers, and the differences between
897   * these two sets are the dead servers</li>
898   * </ol>
899   * </li>
900   * <li>If this is a new deploy, schedule a InitMetaProcedure to initialize meta</li>
901   * <li>Start necessary service threads - balancer, catalog janior, executor services, and also the
902   * procedure executor, etc. Notice that the balancer must be created first as assignment manager
903   * may use it when assigning regions.</li>
904   * <li>Wait for meta to be initialized if necesssary, start table state manager.</li>
905   * <li>Wait for enough region servers to check-in</li>
906   * <li>Let assignment manager load data from meta and construct region states</li>
907   * <li>Start all other things such as chore services, etc</li>
908   * </ol>
909   * <p/>
910   * Notice that now we will not schedule a special procedure to make meta online(unless the first
911   * time where meta has not been created yet), we will rely on SCP to bring meta online.
912   */
913  private void finishActiveMasterInitialization(MonitoredTask status) throws IOException,
914          InterruptedException, KeeperException, ReplicationException {
915    /*
916     * We are active master now... go initialize components we need to run.
917     */
918    status.setStatus("Initializing Master file system");
919
920    this.masterActiveTime = System.currentTimeMillis();
921    // TODO: Do this using Dependency Injection, using PicoContainer, Guice or Spring.
922
923    // Only initialize the MemStoreLAB when master carry table
924    if (LoadBalancer.isTablesOnMaster(conf)) {
925      initializeMemStoreChunkCreator();
926    }
927    this.fileSystemManager = new MasterFileSystem(conf);
928    this.walManager = new MasterWalManager(this);
929
930    // enable table descriptors cache
931    this.tableDescriptors.setCacheOn();
932
933    // warm-up HTDs cache on master initialization
934    if (preLoadTableDescriptors) {
935      status.setStatus("Pre-loading table descriptors");
936      this.tableDescriptors.getAll();
937    }
938
939    // Publish cluster ID; set it in Master too. The superclass RegionServer does this later but
940    // only after it has checked in with the Master. At least a few tests ask Master for clusterId
941    // before it has called its run method and before RegionServer has done the reportForDuty.
942    ClusterId clusterId = fileSystemManager.getClusterId();
943    status.setStatus("Publishing Cluster ID " + clusterId + " in ZooKeeper");
944    ZKClusterId.setClusterId(this.zooKeeper, fileSystemManager.getClusterId());
945    this.clusterId = clusterId.toString();
946
947    // Precaution. Put in place the old hbck1 lock file to fence out old hbase1s running their
948    // hbck1s against an hbase2 cluster; it could do damage. To skip this behavior, set
949    // hbase.write.hbck1.lock.file to false.
950    if (this.conf.getBoolean("hbase.write.hbck1.lock.file", true)) {
951      HBaseFsck.checkAndMarkRunningHbck(this.conf,
952          HBaseFsck.createLockRetryCounterFactory(this.conf).create());
953    }
954
955    status.setStatus("Initialize ServerManager and schedule SCP for crash servers");
956    // The below two managers must be created before loading procedures, as they will be used during
957    // loading.
958    this.serverManager = createServerManager(this);
959    this.syncReplicationReplayWALManager = new SyncReplicationReplayWALManager(this);
960    if (!conf.getBoolean(HBASE_SPLIT_WAL_COORDINATED_BY_ZK,
961      DEFAULT_HBASE_SPLIT_COORDINATED_BY_ZK)) {
962      this.splitWALManager = new SplitWALManager(this);
963    }
964    createProcedureExecutor();
965    @SuppressWarnings("rawtypes")
966    Map<Class<? extends Procedure>, List<Procedure<MasterProcedureEnv>>> procsByType =
967      procedureExecutor.getActiveProceduresNoCopy().stream()
968        .collect(Collectors.groupingBy(p -> p.getClass()));
969
970    checkUnsupportedProcedure(procsByType);
971
972    // Create Assignment Manager
973    this.assignmentManager = createAssignmentManager(this);
974    this.assignmentManager.start();
975    // TODO: TRSP can perform as the sub procedure for other procedures, so even if it is marked as
976    // completed, it could still be in the procedure list. This is a bit strange but is another
977    // story, need to verify the implementation for ProcedureExecutor and ProcedureStore.
978    List<TransitRegionStateProcedure> ritList =
979      procsByType.getOrDefault(TransitRegionStateProcedure.class, Collections.emptyList()).stream()
980        .filter(p -> !p.isFinished()).map(p -> (TransitRegionStateProcedure) p)
981        .collect(Collectors.toList());
982    this.assignmentManager.setupRIT(ritList);
983
984    // Start RegionServerTracker with listing of servers found with exiting SCPs -- these should
985    // be registered in the deadServers set -- and with the list of servernames out on the
986    // filesystem that COULD BE 'alive' (we'll schedule SCPs for each and let SCP figure it out).
987    // We also pass dirs that are already 'splitting'... so we can do some checks down in tracker.
988    // TODO: Generate the splitting and live Set in one pass instead of two as we currently do.
989    this.regionServerTracker = new RegionServerTracker(zooKeeper, this, this.serverManager);
990    this.regionServerTracker.start(
991      procsByType.getOrDefault(ServerCrashProcedure.class, Collections.emptyList()).stream()
992        .map(p -> (ServerCrashProcedure) p).map(p -> p.getServerName()).collect(Collectors.toSet()),
993      walManager.getLiveServersFromWALDir(), walManager.getSplittingServersFromWALDir());
994    // This manager will be started AFTER hbase:meta is confirmed on line.
995    // hbase.mirror.table.state.to.zookeeper is so hbase1 clients can connect. They read table
996    // state from zookeeper while hbase2 reads it from hbase:meta. Disable if no hbase1 clients.
997    this.tableStateManager =
998      this.conf.getBoolean(MirroringTableStateManager.MIRROR_TABLE_STATE_TO_ZK_KEY, true)
999        ?
1000        new MirroringTableStateManager(this):
1001        new TableStateManager(this);
1002
1003    status.setStatus("Initializing ZK system trackers");
1004    initializeZKBasedSystemTrackers();
1005    status.setStatus("Loading last flushed sequence id of regions");
1006    try {
1007      this.serverManager.loadLastFlushedSequenceIds();
1008    } catch (IOException e) {
1009      LOG.info("Failed to load last flushed sequence id of regions"
1010          + " from file system", e);
1011    }
1012    // Set ourselves as active Master now our claim has succeeded up in zk.
1013    this.activeMaster = true;
1014
1015    // Start the Zombie master detector after setting master as active, see HBASE-21535
1016    Thread zombieDetector = new Thread(new InitializationMonitor(this),
1017        "ActiveMasterInitializationMonitor-" + System.currentTimeMillis());
1018    zombieDetector.setDaemon(true);
1019    zombieDetector.start();
1020
1021    // This is for backwards compatibility
1022    // See HBASE-11393
1023    status.setStatus("Update TableCFs node in ZNode");
1024    ReplicationPeerConfigUpgrader tableCFsUpdater =
1025        new ReplicationPeerConfigUpgrader(zooKeeper, conf);
1026    tableCFsUpdater.copyTableCFs();
1027
1028    if (!maintenanceMode) {
1029      // Add the Observer to delete quotas on table deletion before starting all CPs by
1030      // default with quota support, avoiding if user specifically asks to not load this Observer.
1031      if (QuotaUtil.isQuotaEnabled(conf)) {
1032        updateConfigurationForQuotasObserver(conf);
1033      }
1034      // initialize master side coprocessors before we start handling requests
1035      status.setStatus("Initializing master coprocessors");
1036      this.cpHost = new MasterCoprocessorHost(this, this.conf);
1037    }
1038
1039    // Checking if meta needs initializing.
1040    status.setStatus("Initializing meta table if this is a new deploy");
1041    InitMetaProcedure initMetaProc = null;
1042    // Print out state of hbase:meta on startup; helps debugging.
1043    RegionState rs = this.assignmentManager.getRegionStates().
1044        getRegionState(RegionInfoBuilder.FIRST_META_REGIONINFO);
1045    LOG.info("hbase:meta {}", rs);
1046    if (rs.isOffline()) {
1047      Optional<InitMetaProcedure> optProc = procedureExecutor.getProcedures().stream()
1048        .filter(p -> p instanceof InitMetaProcedure).map(o -> (InitMetaProcedure) o).findAny();
1049      initMetaProc = optProc.orElseGet(() -> {
1050        // schedule an init meta procedure if meta has not been deployed yet
1051        InitMetaProcedure temp = new InitMetaProcedure();
1052        procedureExecutor.submitProcedure(temp);
1053        return temp;
1054      });
1055    }
1056    if (this.balancer instanceof FavoredNodesPromoter) {
1057      favoredNodesManager = new FavoredNodesManager(this);
1058    }
1059
1060    // initialize load balancer
1061    this.balancer.setMasterServices(this);
1062    this.balancer.setClusterMetrics(getClusterMetricsWithoutCoprocessor());
1063    this.balancer.initialize();
1064
1065    // start up all service threads.
1066    status.setStatus("Initializing master service threads");
1067    startServiceThreads();
1068    // wait meta to be initialized after we start procedure executor
1069    if (initMetaProc != null) {
1070      initMetaProc.await();
1071    }
1072    // Wake up this server to check in
1073    sleeper.skipSleepCycle();
1074
1075    // Wait for region servers to report in.
1076    // With this as part of master initialization, it precludes our being able to start a single
1077    // server that is both Master and RegionServer. Needs more thought. TODO.
1078    String statusStr = "Wait for region servers to report in";
1079    status.setStatus(statusStr);
1080    LOG.info(Objects.toString(status));
1081    waitForRegionServers(status);
1082
1083    // Check if master is shutting down because issue initializing regionservers or balancer.
1084    if (isStopped()) {
1085      return;
1086    }
1087
1088    status.setStatus("Starting assignment manager");
1089    // FIRST HBASE:META READ!!!!
1090    // The below cannot make progress w/o hbase:meta being online.
1091    // This is the FIRST attempt at going to hbase:meta. Meta on-lining is going on in background
1092    // as procedures run -- in particular SCPs for crashed servers... One should put up hbase:meta
1093    // if it is down. It may take a while to come online. So, wait here until meta if for sure
1094    // available. That's what waitForMetaOnline does.
1095    if (!waitForMetaOnline()) {
1096      return;
1097    }
1098    this.assignmentManager.joinCluster();
1099    // The below depends on hbase:meta being online.
1100    this.tableStateManager.start();
1101    // Below has to happen after tablestatemanager has started in the case where this hbase-2.x
1102    // is being started over an hbase-1.x dataset. tablestatemanager runs a migration as part
1103    // of its 'start' moving table state from zookeeper to hbase:meta. This migration needs to
1104    // complete before we do this next step processing offline regions else it fails reading
1105    // table states messing up master launch (namespace table, etc., are not assigned).
1106    this.assignmentManager.processOfflineRegions();
1107    // Initialize after meta is up as below scans meta
1108    if (favoredNodesManager != null && !maintenanceMode) {
1109      SnapshotOfRegionAssignmentFromMeta snapshotOfRegionAssignment =
1110          new SnapshotOfRegionAssignmentFromMeta(getConnection());
1111      snapshotOfRegionAssignment.initialize();
1112      favoredNodesManager.initialize(snapshotOfRegionAssignment);
1113    }
1114
1115    // set cluster status again after user regions are assigned
1116    this.balancer.setClusterMetrics(getClusterMetricsWithoutCoprocessor());
1117
1118    // Start balancer and meta catalog janitor after meta and regions have been assigned.
1119    status.setStatus("Starting balancer and catalog janitor");
1120    this.clusterStatusChore = new ClusterStatusChore(this, balancer);
1121    getChoreService().scheduleChore(clusterStatusChore);
1122    this.balancerChore = new BalancerChore(this);
1123    getChoreService().scheduleChore(balancerChore);
1124    this.normalizerChore = new RegionNormalizerChore(this);
1125    getChoreService().scheduleChore(normalizerChore);
1126    this.catalogJanitorChore = new CatalogJanitor(this);
1127    getChoreService().scheduleChore(catalogJanitorChore);
1128    this.hbckChore = new HbckChore(this);
1129    getChoreService().scheduleChore(hbckChore);
1130    this.serverManager.startChore();
1131
1132    // Only for rolling upgrade, where we need to migrate the data in namespace table to meta table.
1133    if (!waitForNamespaceOnline()) {
1134      return;
1135    }
1136    status.setStatus("Starting cluster schema service");
1137    initClusterSchemaService();
1138
1139    if (this.cpHost != null) {
1140      try {
1141        this.cpHost.preMasterInitialization();
1142      } catch (IOException e) {
1143        LOG.error("Coprocessor preMasterInitialization() hook failed", e);
1144      }
1145    }
1146
1147    status.markComplete("Initialization successful");
1148    LOG.info(String.format("Master has completed initialization %.3fsec",
1149       (System.currentTimeMillis() - masterActiveTime) / 1000.0f));
1150    this.masterFinishedInitializationTime = System.currentTimeMillis();
1151    configurationManager.registerObserver(this.balancer);
1152    configurationManager.registerObserver(this.cleanerPool);
1153    configurationManager.registerObserver(this.hfileCleaner);
1154    configurationManager.registerObserver(this.logCleaner);
1155    configurationManager.registerObserver(this.regionsRecoveryConfigManager);
1156    // Set master as 'initialized'.
1157    setInitialized(true);
1158
1159    if (maintenanceMode) {
1160      LOG.info("Detected repair mode, skipping final initialization steps.");
1161      return;
1162    }
1163
1164    assignmentManager.checkIfShouldMoveSystemRegionAsync();
1165    status.setStatus("Assign meta replicas");
1166    MasterMetaBootstrap metaBootstrap = createMetaBootstrap();
1167    metaBootstrap.assignMetaReplicas();
1168    status.setStatus("Starting quota manager");
1169    initQuotaManager();
1170    if (QuotaUtil.isQuotaEnabled(conf)) {
1171      // Create the quota snapshot notifier
1172      spaceQuotaSnapshotNotifier = createQuotaSnapshotNotifier();
1173      spaceQuotaSnapshotNotifier.initialize(getConnection());
1174      this.quotaObserverChore = new QuotaObserverChore(this, getMasterMetrics());
1175      // Start the chore to read the region FS space reports and act on them
1176      getChoreService().scheduleChore(quotaObserverChore);
1177
1178      this.snapshotQuotaChore = new SnapshotQuotaObserverChore(this, getMasterMetrics());
1179      // Start the chore to read snapshots and add their usage to table/NS quotas
1180      getChoreService().scheduleChore(snapshotQuotaChore);
1181    }
1182
1183    // clear the dead servers with same host name and port of online server because we are not
1184    // removing dead server with same hostname and port of rs which is trying to check in before
1185    // master initialization. See HBASE-5916.
1186    this.serverManager.clearDeadServersWithSameHostNameAndPortOfOnlineServer();
1187
1188    // Check and set the znode ACLs if needed in case we are overtaking a non-secure configuration
1189    status.setStatus("Checking ZNode ACLs");
1190    zooKeeper.checkAndSetZNodeAcls();
1191
1192    status.setStatus("Initializing MOB Cleaner");
1193    initMobCleaner();
1194
1195    status.setStatus("Calling postStartMaster coprocessors");
1196    if (this.cpHost != null) {
1197      // don't let cp initialization errors kill the master
1198      try {
1199        this.cpHost.postStartMaster();
1200      } catch (IOException ioe) {
1201        LOG.error("Coprocessor postStartMaster() hook failed", ioe);
1202      }
1203    }
1204
1205    zombieDetector.interrupt();
1206
1207    /*
1208     * After master has started up, lets do balancer post startup initialization. Since this runs
1209     * in activeMasterManager thread, it should be fine.
1210     */
1211    long start = System.currentTimeMillis();
1212    this.balancer.postMasterStartupInitialize();
1213    if (LOG.isDebugEnabled()) {
1214      LOG.debug("Balancer post startup initialization complete, took " + (
1215          (System.currentTimeMillis() - start) / 1000) + " seconds");
1216    }
1217  }
1218
1219  /**
1220   * Check hbase:meta is up and ready for reading. For use during Master startup only.
1221   * @return True if meta is UP and online and startup can progress. Otherwise, meta is not online
1222   *   and we will hold here until operator intervention.
1223   */
1224  @VisibleForTesting
1225  public boolean waitForMetaOnline() throws InterruptedException {
1226    return isRegionOnline(RegionInfoBuilder.FIRST_META_REGIONINFO);
1227  }
1228
1229  /**
1230   * @return True if region is online and scannable else false if an error or shutdown (Otherwise
1231   *   we just block in here holding up all forward-progess).
1232   */
1233  private boolean isRegionOnline(RegionInfo ri) throws InterruptedException {
1234    RetryCounter rc = null;
1235    while (!isStopped()) {
1236      RegionState rs = this.assignmentManager.getRegionStates().getRegionState(ri);
1237      if (rs.isOpened()) {
1238        if (this.getServerManager().isServerOnline(rs.getServerName())) {
1239          return true;
1240        }
1241      }
1242      // Region is not OPEN.
1243      Optional<Procedure<MasterProcedureEnv>> optProc = this.procedureExecutor.getProcedures().
1244          stream().filter(p -> p instanceof ServerCrashProcedure).findAny();
1245      // TODO: Add a page to refguide on how to do repair. Have this log message point to it.
1246      // Page will talk about loss of edits, how to schedule at least the meta WAL recovery, and
1247      // then how to assign including how to break region lock if one held.
1248      LOG.warn("{} is NOT online; state={}; ServerCrashProcedures={}. Master startup cannot " +
1249          "progress, in holding-pattern until region onlined.",
1250          ri.getRegionNameAsString(), rs, optProc.isPresent());
1251      // Check once-a-minute.
1252      if (rc == null) {
1253        rc = new RetryCounterFactory(1000).create();
1254      }
1255      Threads.sleep(rc.getBackoffTimeAndIncrementAttempts());
1256    }
1257    return false;
1258  }
1259
1260  /**
1261   * Check hbase:namespace table is assigned. If not, startup will hang looking for the ns table
1262   * <p/>
1263   * This is for rolling upgrading, later we will migrate the data in ns table to the ns family of
1264   * meta table. And if this is a new clsuter, this method will return immediately as there will be
1265   * no namespace table/region.
1266   * @return True if namespace table is up/online.
1267   */
1268  private boolean waitForNamespaceOnline() throws InterruptedException, IOException {
1269    TableState nsTableState =
1270      MetaTableAccessor.getTableState(getConnection(), TableName.NAMESPACE_TABLE_NAME);
1271    if (nsTableState == null || nsTableState.isDisabled()) {
1272      // this means we have already migrated the data and disabled or deleted the namespace table,
1273      // or this is a new depliy which does not have a namespace table from the beginning.
1274      return true;
1275    }
1276    List<RegionInfo> ris =
1277      this.assignmentManager.getRegionStates().getRegionsOfTable(TableName.NAMESPACE_TABLE_NAME);
1278    if (ris.isEmpty()) {
1279      // maybe this will not happen any more, but anyway, no harm to add a check here...
1280      return true;
1281    }
1282    // Else there are namespace regions up in meta. Ensure they are assigned before we go on.
1283    for (RegionInfo ri : ris) {
1284      isRegionOnline(ri);
1285    }
1286    return true;
1287  }
1288
1289  /**
1290   * Adds the {@code MasterQuotasObserver} to the list of configured Master observers to
1291   * automatically remove quotas for a table when that table is deleted.
1292   */
1293  @VisibleForTesting
1294  public void updateConfigurationForQuotasObserver(Configuration conf) {
1295    // We're configured to not delete quotas on table deletion, so we don't need to add the obs.
1296    if (!conf.getBoolean(
1297          MasterQuotasObserver.REMOVE_QUOTA_ON_TABLE_DELETE,
1298          MasterQuotasObserver.REMOVE_QUOTA_ON_TABLE_DELETE_DEFAULT)) {
1299      return;
1300    }
1301    String[] masterCoprocs = conf.getStrings(CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY);
1302    final int length = null == masterCoprocs ? 0 : masterCoprocs.length;
1303    String[] updatedCoprocs = new String[length + 1];
1304    if (length > 0) {
1305      System.arraycopy(masterCoprocs, 0, updatedCoprocs, 0, masterCoprocs.length);
1306    }
1307    updatedCoprocs[length] = MasterQuotasObserver.class.getName();
1308    conf.setStrings(CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY, updatedCoprocs);
1309  }
1310
1311  private void initMobCleaner() {
1312    this.expiredMobFileCleanerChore = new ExpiredMobFileCleanerChore(this);
1313    getChoreService().scheduleChore(expiredMobFileCleanerChore);
1314
1315    int mobCompactionPeriod = conf.getInt(MobConstants.MOB_COMPACTION_CHORE_PERIOD,
1316        MobConstants.DEFAULT_MOB_COMPACTION_CHORE_PERIOD);
1317    this.mobCompactChore = new MobCompactionChore(this, mobCompactionPeriod);
1318    getChoreService().scheduleChore(mobCompactChore);
1319    this.mobCompactThread = new MasterMobCompactionThread(this);
1320  }
1321
1322  /**
1323   * <p>
1324   * Create a {@link MasterMetaBootstrap} instance.
1325   * </p>
1326   * <p>
1327   * Will be overridden in tests.
1328   * </p>
1329   */
1330  @VisibleForTesting
1331  protected MasterMetaBootstrap createMetaBootstrap() {
1332    // We put this out here in a method so can do a Mockito.spy and stub it out
1333    // w/ a mocked up MasterMetaBootstrap.
1334    return new MasterMetaBootstrap(this);
1335  }
1336
1337  /**
1338   * <p>
1339   * Create a {@link ServerManager} instance.
1340   * </p>
1341   * <p>
1342   * Will be overridden in tests.
1343   * </p>
1344   */
1345  @VisibleForTesting
1346  protected ServerManager createServerManager(final MasterServices master) throws IOException {
1347    // We put this out here in a method so can do a Mockito.spy and stub it out
1348    // w/ a mocked up ServerManager.
1349    setupClusterConnection();
1350    return new ServerManager(master);
1351  }
1352
1353  private void waitForRegionServers(final MonitoredTask status)
1354      throws IOException, InterruptedException {
1355    this.serverManager.waitForRegionServers(status);
1356  }
1357
1358  // Will be overridden in tests
1359  @VisibleForTesting
1360  protected void initClusterSchemaService() throws IOException, InterruptedException {
1361    this.clusterSchemaService = new ClusterSchemaServiceImpl(this);
1362    this.clusterSchemaService.startAsync();
1363    try {
1364      this.clusterSchemaService.awaitRunning(getConfiguration().getInt(
1365        HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS,
1366        DEFAULT_HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS), TimeUnit.SECONDS);
1367    } catch (TimeoutException toe) {
1368      throw new IOException("Timedout starting ClusterSchemaService", toe);
1369    }
1370  }
1371
1372  private void initQuotaManager() throws IOException {
1373    MasterQuotaManager quotaManager = new MasterQuotaManager(this);
1374    quotaManager.start();
1375    this.quotaManager = quotaManager;
1376  }
1377
1378  private SpaceQuotaSnapshotNotifier createQuotaSnapshotNotifier() {
1379    SpaceQuotaSnapshotNotifier notifier =
1380        SpaceQuotaSnapshotNotifierFactory.getInstance().create(getConfiguration());
1381    return notifier;
1382  }
1383
1384  boolean isCatalogJanitorEnabled() {
1385    return catalogJanitorChore != null ?
1386      catalogJanitorChore.getEnabled() : false;
1387  }
1388
1389  boolean isCleanerChoreEnabled() {
1390    boolean hfileCleanerFlag = true, logCleanerFlag = true;
1391
1392    if (hfileCleaner != null) {
1393      hfileCleanerFlag = hfileCleaner.getEnabled();
1394    }
1395
1396    if (logCleaner != null) {
1397      logCleanerFlag = logCleaner.getEnabled();
1398    }
1399
1400    return (hfileCleanerFlag && logCleanerFlag);
1401  }
1402
1403  @Override
1404  public ServerManager getServerManager() {
1405    return this.serverManager;
1406  }
1407
1408  @Override
1409  public MasterFileSystem getMasterFileSystem() {
1410    return this.fileSystemManager;
1411  }
1412
1413  @Override
1414  public MasterWalManager getMasterWalManager() {
1415    return this.walManager;
1416  }
1417
1418  @Override
1419  public SplitWALManager getSplitWALManager() {
1420    return splitWALManager;
1421  }
1422
1423  @Override
1424  public TableStateManager getTableStateManager() {
1425    return tableStateManager;
1426  }
1427
1428  /*
1429   * Start up all services. If any of these threads gets an unhandled exception
1430   * then they just die with a logged message.  This should be fine because
1431   * in general, we do not expect the master to get such unhandled exceptions
1432   *  as OOMEs; it should be lightly loaded. See what HRegionServer does if
1433   *  need to install an unexpected exception handler.
1434   */
1435  private void startServiceThreads() throws IOException {
1436    // Start the executor service pools
1437    this.executorService.startExecutorService(ExecutorType.MASTER_OPEN_REGION, conf.getInt(
1438      HConstants.MASTER_OPEN_REGION_THREADS, HConstants.MASTER_OPEN_REGION_THREADS_DEFAULT));
1439    this.executorService.startExecutorService(ExecutorType.MASTER_CLOSE_REGION, conf.getInt(
1440      HConstants.MASTER_CLOSE_REGION_THREADS, HConstants.MASTER_CLOSE_REGION_THREADS_DEFAULT));
1441    this.executorService.startExecutorService(ExecutorType.MASTER_SERVER_OPERATIONS,
1442      conf.getInt(HConstants.MASTER_SERVER_OPERATIONS_THREADS,
1443        HConstants.MASTER_SERVER_OPERATIONS_THREADS_DEFAULT));
1444    this.executorService.startExecutorService(ExecutorType.MASTER_META_SERVER_OPERATIONS,
1445      conf.getInt(HConstants.MASTER_META_SERVER_OPERATIONS_THREADS,
1446        HConstants.MASTER_META_SERVER_OPERATIONS_THREADS_DEFAULT));
1447    this.executorService.startExecutorService(ExecutorType.M_LOG_REPLAY_OPS, conf.getInt(
1448      HConstants.MASTER_LOG_REPLAY_OPS_THREADS, HConstants.MASTER_LOG_REPLAY_OPS_THREADS_DEFAULT));
1449    this.executorService.startExecutorService(ExecutorType.MASTER_SNAPSHOT_OPERATIONS, conf.getInt(
1450      SnapshotManager.SNAPSHOT_POOL_THREADS_KEY, SnapshotManager.SNAPSHOT_POOL_THREADS_DEFAULT));
1451
1452   // We depend on there being only one instance of this executor running
1453   // at a time.  To do concurrency, would need fencing of enable/disable of
1454   // tables.
1455   // Any time changing this maxThreads to > 1, pls see the comment at
1456   // AccessController#postCompletedCreateTableAction
1457   this.executorService.startExecutorService(ExecutorType.MASTER_TABLE_OPERATIONS, 1);
1458   startProcedureExecutor();
1459
1460    // Create cleaner thread pool
1461    cleanerPool = new DirScanPool(conf);
1462    // Start log cleaner thread
1463    int cleanerInterval = conf.getInt("hbase.master.cleaner.interval", 600 * 1000);
1464    this.logCleaner = new LogCleaner(cleanerInterval, this, conf,
1465      getMasterWalManager().getFileSystem(), getMasterWalManager().getOldLogDir(), cleanerPool);
1466    getChoreService().scheduleChore(logCleaner);
1467
1468    // start the hfile archive cleaner thread
1469    Path archiveDir = HFileArchiveUtil.getArchivePath(conf);
1470    Map<String, Object> params = new HashMap<>();
1471    params.put(MASTER, this);
1472    this.hfileCleaner = new HFileCleaner(cleanerInterval, this, conf,
1473      getMasterFileSystem().getFileSystem(), archiveDir, cleanerPool, params);
1474    getChoreService().scheduleChore(hfileCleaner);
1475
1476    // Regions Reopen based on very high storeFileRefCount is considered enabled
1477    // only if hbase.regions.recovery.store.file.ref.count has value > 0
1478    final int maxStoreFileRefCount = conf.getInt(
1479      HConstants.STORE_FILE_REF_COUNT_THRESHOLD,
1480      HConstants.DEFAULT_STORE_FILE_REF_COUNT_THRESHOLD);
1481    if (maxStoreFileRefCount > 0) {
1482      this.regionsRecoveryChore = new RegionsRecoveryChore(this, conf, this);
1483      getChoreService().scheduleChore(this.regionsRecoveryChore);
1484    } else {
1485      LOG.info("Reopening regions with very high storeFileRefCount is disabled. " +
1486          "Provide threshold value > 0 for {} to enable it.",
1487        HConstants.STORE_FILE_REF_COUNT_THRESHOLD);
1488    }
1489
1490    this.regionsRecoveryConfigManager = new RegionsRecoveryConfigManager(this);
1491
1492    replicationBarrierCleaner = new ReplicationBarrierCleaner(conf, this, getConnection(),
1493      replicationPeerManager);
1494    getChoreService().scheduleChore(replicationBarrierCleaner);
1495
1496    final boolean isSnapshotChoreEnabled = this.snapshotCleanupTracker
1497        .isSnapshotCleanupEnabled();
1498    this.snapshotCleanerChore = new SnapshotCleanerChore(this, conf, getSnapshotManager());
1499    if (isSnapshotChoreEnabled) {
1500      getChoreService().scheduleChore(this.snapshotCleanerChore);
1501    } else {
1502      if (LOG.isTraceEnabled()) {
1503        LOG.trace("Snapshot Cleaner Chore is disabled. Not starting up the chore..");
1504      }
1505    }
1506    serviceStarted = true;
1507    if (LOG.isTraceEnabled()) {
1508      LOG.trace("Started service threads");
1509    }
1510  }
1511
1512  @Override
1513  protected void stopServiceThreads() {
1514    if (masterJettyServer != null) {
1515      LOG.info("Stopping master jetty server");
1516      try {
1517        masterJettyServer.stop();
1518      } catch (Exception e) {
1519        LOG.error("Failed to stop master jetty server", e);
1520      }
1521    }
1522    stopChores();
1523    if (this.mobCompactThread != null) {
1524      this.mobCompactThread.close();
1525    }
1526    super.stopServiceThreads();
1527    if (cleanerPool != null) {
1528      cleanerPool.shutdownNow();
1529      cleanerPool = null;
1530    }
1531
1532    LOG.debug("Stopping service threads");
1533
1534    if (this.quotaManager != null) {
1535      this.quotaManager.stop();
1536    }
1537
1538    if (this.activeMasterManager != null) {
1539      this.activeMasterManager.stop();
1540    }
1541    if (this.serverManager != null) {
1542      this.serverManager.stop();
1543    }
1544    if (this.assignmentManager != null) {
1545      this.assignmentManager.stop();
1546    }
1547
1548    stopProcedureExecutor();
1549
1550    if (this.walManager != null) {
1551      this.walManager.stop();
1552    }
1553    if (this.fileSystemManager != null) {
1554      this.fileSystemManager.stop();
1555    }
1556    if (this.mpmHost != null) {
1557      this.mpmHost.stop("server shutting down.");
1558    }
1559    if (this.regionServerTracker != null) {
1560      this.regionServerTracker.stop();
1561    }
1562  }
1563
1564  private void createProcedureExecutor() throws IOException {
1565    MasterProcedureEnv procEnv = new MasterProcedureEnv(this);
1566    procedureStore =
1567      new WALProcedureStore(conf, new MasterProcedureEnv.WALStoreLeaseRecovery(this));
1568    procedureStore.registerListener(new ProcedureStoreListener() {
1569
1570      @Override
1571      public void abortProcess() {
1572        abort("The Procedure Store lost the lease", null);
1573      }
1574    });
1575    MasterProcedureScheduler procedureScheduler = procEnv.getProcedureScheduler();
1576    procedureExecutor = new ProcedureExecutor<>(conf, procEnv, procedureStore, procedureScheduler);
1577    configurationManager.registerObserver(procEnv);
1578
1579    int cpus = Runtime.getRuntime().availableProcessors();
1580    final int numThreads = conf.getInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, Math.max(
1581      (cpus > 0 ? cpus / 4 : 0), MasterProcedureConstants.DEFAULT_MIN_MASTER_PROCEDURE_THREADS));
1582    final boolean abortOnCorruption =
1583      conf.getBoolean(MasterProcedureConstants.EXECUTOR_ABORT_ON_CORRUPTION,
1584        MasterProcedureConstants.DEFAULT_EXECUTOR_ABORT_ON_CORRUPTION);
1585    procedureStore.start(numThreads);
1586    // Just initialize it but do not start the workers, we will start the workers later by calling
1587    // startProcedureExecutor. See the javadoc for finishActiveMasterInitialization for more
1588    // details.
1589    procedureExecutor.init(numThreads, abortOnCorruption);
1590    procEnv.getRemoteDispatcher().start();
1591  }
1592
1593  private void startProcedureExecutor() throws IOException {
1594    procedureExecutor.startWorkers();
1595  }
1596
1597  /**
1598   * Turn on/off Snapshot Cleanup Chore
1599   *
1600   * @param on indicates whether Snapshot Cleanup Chore is to be run
1601   */
1602  void switchSnapshotCleanup(final boolean on, final boolean synchronous) {
1603    if (synchronous) {
1604      synchronized (this.snapshotCleanerChore) {
1605        switchSnapshotCleanup(on);
1606      }
1607    } else {
1608      switchSnapshotCleanup(on);
1609    }
1610  }
1611
1612  private void switchSnapshotCleanup(final boolean on) {
1613    try {
1614      snapshotCleanupTracker.setSnapshotCleanupEnabled(on);
1615      if (on) {
1616        if (!getChoreService().isChoreScheduled(this.snapshotCleanerChore)) {
1617          getChoreService().scheduleChore(this.snapshotCleanerChore);
1618        }
1619      } else {
1620        getChoreService().cancelChore(this.snapshotCleanerChore);
1621      }
1622    } catch (KeeperException e) {
1623      LOG.error("Error updating snapshot cleanup mode to {}", on, e);
1624    }
1625  }
1626
1627
1628  private void stopProcedureExecutor() {
1629    if (procedureExecutor != null) {
1630      configurationManager.deregisterObserver(procedureExecutor.getEnvironment());
1631      procedureExecutor.getEnvironment().getRemoteDispatcher().stop();
1632      procedureExecutor.stop();
1633      procedureExecutor.join();
1634      procedureExecutor = null;
1635    }
1636
1637    if (procedureStore != null) {
1638      procedureStore.stop(isAborted());
1639      procedureStore = null;
1640    }
1641  }
1642
1643  private void stopChores() {
1644    ChoreService choreService = getChoreService();
1645    if (choreService != null) {
1646      choreService.cancelChore(this.expiredMobFileCleanerChore);
1647      choreService.cancelChore(this.mobCompactChore);
1648      choreService.cancelChore(this.balancerChore);
1649      choreService.cancelChore(this.normalizerChore);
1650      choreService.cancelChore(this.clusterStatusChore);
1651      choreService.cancelChore(this.catalogJanitorChore);
1652      choreService.cancelChore(this.clusterStatusPublisherChore);
1653      choreService.cancelChore(this.snapshotQuotaChore);
1654      choreService.cancelChore(this.logCleaner);
1655      choreService.cancelChore(this.hfileCleaner);
1656      choreService.cancelChore(this.replicationBarrierCleaner);
1657      choreService.cancelChore(this.snapshotCleanerChore);
1658      choreService.cancelChore(this.hbckChore);
1659      choreService.cancelChore(this.regionsRecoveryChore);
1660    }
1661  }
1662
1663  /**
1664   * @return Get remote side's InetAddress
1665   */
1666  InetAddress getRemoteInetAddress(final int port,
1667      final long serverStartCode) throws UnknownHostException {
1668    // Do it out here in its own little method so can fake an address when
1669    // mocking up in tests.
1670    InetAddress ia = RpcServer.getRemoteIp();
1671
1672    // The call could be from the local regionserver,
1673    // in which case, there is no remote address.
1674    if (ia == null && serverStartCode == startcode) {
1675      InetSocketAddress isa = rpcServices.getSocketAddress();
1676      if (isa != null && isa.getPort() == port) {
1677        ia = isa.getAddress();
1678      }
1679    }
1680    return ia;
1681  }
1682
1683  /**
1684   * @return Maximum time we should run balancer for
1685   */
1686  private int getMaxBalancingTime() {
1687    int maxBalancingTime = getConfiguration().getInt(HConstants.HBASE_BALANCER_MAX_BALANCING, -1);
1688    if (maxBalancingTime == -1) {
1689      // if max balancing time isn't set, defaulting it to period time
1690      maxBalancingTime = getConfiguration().getInt(HConstants.HBASE_BALANCER_PERIOD,
1691        HConstants.DEFAULT_HBASE_BALANCER_PERIOD);
1692    }
1693    return maxBalancingTime;
1694  }
1695
1696  /**
1697   * @return Maximum number of regions in transition
1698   */
1699  private int getMaxRegionsInTransition() {
1700    int numRegions = this.assignmentManager.getRegionStates().getRegionAssignments().size();
1701    return Math.max((int) Math.floor(numRegions * this.maxRitPercent), 1);
1702  }
1703
1704  /**
1705   * It first sleep to the next balance plan start time. Meanwhile, throttling by the max
1706   * number regions in transition to protect availability.
1707   * @param nextBalanceStartTime The next balance plan start time
1708   * @param maxRegionsInTransition max number of regions in transition
1709   * @param cutoffTime when to exit balancer
1710   */
1711  private void balanceThrottling(long nextBalanceStartTime, int maxRegionsInTransition,
1712      long cutoffTime) {
1713    boolean interrupted = false;
1714
1715    // Sleep to next balance plan start time
1716    // But if there are zero regions in transition, it can skip sleep to speed up.
1717    while (!interrupted && System.currentTimeMillis() < nextBalanceStartTime
1718        && this.assignmentManager.getRegionStates().hasRegionsInTransition()) {
1719      try {
1720        Thread.sleep(100);
1721      } catch (InterruptedException ie) {
1722        interrupted = true;
1723      }
1724    }
1725
1726    // Throttling by max number regions in transition
1727    while (!interrupted
1728        && maxRegionsInTransition > 0
1729        && this.assignmentManager.getRegionStates().getRegionsInTransitionCount()
1730        >= maxRegionsInTransition && System.currentTimeMillis() <= cutoffTime) {
1731      try {
1732        // sleep if the number of regions in transition exceeds the limit
1733        Thread.sleep(100);
1734      } catch (InterruptedException ie) {
1735        interrupted = true;
1736      }
1737    }
1738
1739    if (interrupted) Thread.currentThread().interrupt();
1740  }
1741
1742  public boolean balance() throws IOException {
1743    return balance(false);
1744  }
1745
1746  public boolean balance(boolean force) throws IOException {
1747    // if master not initialized, don't run balancer.
1748    if (!isInitialized()) {
1749      LOG.debug("Master has not been initialized, don't run balancer.");
1750      return false;
1751    }
1752
1753    if (isInMaintenanceMode()) {
1754      LOG.info("Master is in maintenanceMode mode, don't run balancer.");
1755      return false;
1756    }
1757
1758    synchronized (this.balancer) {
1759      // If balance not true, don't run balancer.
1760      if (!this.loadBalancerTracker.isBalancerOn()) return false;
1761        // Only allow one balance run at at time.
1762      if (this.assignmentManager.hasRegionsInTransition()) {
1763        List<RegionStateNode> regionsInTransition = assignmentManager.getRegionsInTransition();
1764        // if hbase:meta region is in transition, result of assignment cannot be recorded
1765        // ignore the force flag in that case
1766        boolean metaInTransition = assignmentManager.isMetaRegionInTransition();
1767        String prefix = force && !metaInTransition ? "R" : "Not r";
1768        List<RegionStateNode> toPrint = regionsInTransition;
1769        int max = 5;
1770        boolean truncated = false;
1771        if (regionsInTransition.size() > max) {
1772          toPrint = regionsInTransition.subList(0, max);
1773          truncated = true;
1774        }
1775        LOG.info(prefix + "unning balancer because " + regionsInTransition.size() +
1776          " region(s) in transition: " + toPrint + (truncated? "(truncated list)": ""));
1777        if (!force || metaInTransition) return false;
1778      }
1779      if (this.serverManager.areDeadServersInProgress()) {
1780        LOG.info("Not running balancer because processing dead regionserver(s): " +
1781          this.serverManager.getDeadServers());
1782        return false;
1783      }
1784
1785      if (this.cpHost != null) {
1786        try {
1787          if (this.cpHost.preBalance()) {
1788            LOG.debug("Coprocessor bypassing balancer request");
1789            return false;
1790          }
1791        } catch (IOException ioe) {
1792          LOG.error("Error invoking master coprocessor preBalance()", ioe);
1793          return false;
1794        }
1795      }
1796
1797      boolean isByTable = getConfiguration().getBoolean("hbase.master.loadbalance.bytable", false);
1798      Map<TableName, Map<ServerName, List<RegionInfo>>> assignments =
1799        this.assignmentManager.getRegionStates()
1800          .getAssignmentsForBalancer(tableStateManager, isByTable);
1801      for (Map<ServerName, List<RegionInfo>> serverMap : assignments.values()) {
1802        serverMap.keySet().removeAll(this.serverManager.getDrainingServersList());
1803      }
1804
1805      //Give the balancer the current cluster state.
1806      this.balancer.setClusterMetrics(getClusterMetricsWithoutCoprocessor());
1807      this.balancer.setClusterLoad(assignments);
1808
1809      List<RegionPlan> plans = new ArrayList<>();
1810      for (Entry<TableName, Map<ServerName, List<RegionInfo>>> e : assignments.entrySet()) {
1811        List<RegionPlan> partialPlans = this.balancer.balanceCluster(e.getKey(), e.getValue());
1812        if (partialPlans != null) {
1813          plans.addAll(partialPlans);
1814        }
1815      }
1816
1817      List<RegionPlan> sucRPs = executeRegionPlansWithThrottling(plans);
1818
1819      if (this.cpHost != null) {
1820        try {
1821          this.cpHost.postBalance(sucRPs);
1822        } catch (IOException ioe) {
1823          // balancing already succeeded so don't change the result
1824          LOG.error("Error invoking master coprocessor postBalance()", ioe);
1825        }
1826      }
1827    }
1828    // If LoadBalancer did not generate any plans, it means the cluster is already balanced.
1829    // Return true indicating a success.
1830    return true;
1831  }
1832
1833  public List<RegionPlan> executeRegionPlansWithThrottling(List<RegionPlan> plans) {
1834    List<RegionPlan> sucRPs = new ArrayList<>();
1835    int maxRegionsInTransition = getMaxRegionsInTransition();
1836    long balanceStartTime = System.currentTimeMillis();
1837    long cutoffTime = balanceStartTime + this.maxBlancingTime;
1838    int rpCount = 0;  // number of RegionPlans balanced so far
1839    if (plans != null && !plans.isEmpty()) {
1840      int balanceInterval = this.maxBlancingTime / plans.size();
1841      LOG.info("Balancer plans size is " + plans.size() + ", the balance interval is "
1842          + balanceInterval + " ms, and the max number regions in transition is "
1843          + maxRegionsInTransition);
1844
1845      for (RegionPlan plan: plans) {
1846        LOG.info("balance " + plan);
1847        //TODO: bulk assign
1848        try {
1849          this.assignmentManager.moveAsync(plan);
1850        } catch (HBaseIOException hioe) {
1851          //should ignore failed plans here, avoiding the whole balance plans be aborted
1852          //later calls of balance() can fetch up the failed and skipped plans
1853          LOG.warn("Failed balance plan: {}, just skip it", plan, hioe);
1854        }
1855        //rpCount records balance plans processed, does not care if a plan succeeds
1856        rpCount++;
1857
1858        balanceThrottling(balanceStartTime + rpCount * balanceInterval, maxRegionsInTransition,
1859            cutoffTime);
1860
1861        // if performing next balance exceeds cutoff time, exit the loop
1862        if (rpCount < plans.size() && System.currentTimeMillis() > cutoffTime) {
1863          // TODO: After balance, there should not be a cutoff time (keeping it as
1864          // a security net for now)
1865          LOG.debug("No more balancing till next balance run; maxBalanceTime="
1866              + this.maxBlancingTime);
1867          break;
1868        }
1869      }
1870    }
1871    return sucRPs;
1872  }
1873
1874  @Override
1875  @VisibleForTesting
1876  public RegionNormalizer getRegionNormalizer() {
1877    return this.normalizer;
1878  }
1879
1880  /**
1881   * Perform normalization of cluster (invoked by {@link RegionNormalizerChore}).
1882   *
1883   * @return true if normalization step was performed successfully, false otherwise
1884   *    (specifically, if HMaster hasn't been initialized properly or normalization
1885   *    is globally disabled)
1886   */
1887  public boolean normalizeRegions() throws IOException {
1888    if (!isInitialized()) {
1889      LOG.debug("Master has not been initialized, don't run region normalizer.");
1890      return false;
1891    }
1892    if (this.getServerManager().isClusterShutdown()) {
1893      LOG.info("Cluster is shutting down, don't run region normalizer.");
1894      return false;
1895    }
1896    if (isInMaintenanceMode()) {
1897      LOG.info("Master is in maintenance mode, don't run region normalizer.");
1898      return false;
1899    }
1900    if (!this.regionNormalizerTracker.isNormalizerOn()) {
1901      LOG.debug("Region normalization is disabled, don't run region normalizer.");
1902      return false;
1903    }
1904
1905    synchronized (this.normalizer) {
1906      // Don't run the normalizer concurrently
1907      List<TableName> allEnabledTables = new ArrayList<>(
1908        this.tableStateManager.getTablesInStates(TableState.State.ENABLED));
1909
1910      Collections.shuffle(allEnabledTables);
1911
1912      for (TableName table : allEnabledTables) {
1913        if (isInMaintenanceMode()) {
1914          LOG.debug("Master is in maintenance mode, stop running region normalizer.");
1915          return false;
1916        }
1917
1918        TableDescriptor tblDesc = getTableDescriptors().get(table);
1919        if (table.isSystemTable() || (tblDesc != null &&
1920            !tblDesc.isNormalizationEnabled())) {
1921          LOG.trace("Skipping normalization for {}, as it's either system"
1922              + " table or doesn't have auto normalization turned on", table);
1923          continue;
1924        }
1925        List<NormalizationPlan> plans = this.normalizer.computePlanForTable(table);
1926        if (plans != null) {
1927          for (NormalizationPlan plan : plans) {
1928            plan.execute(asyncClusterConnection.toConnection().getAdmin());
1929            if (plan.getType() == PlanType.SPLIT) {
1930              splitPlanCount++;
1931            } else if (plan.getType() == PlanType.MERGE) {
1932              mergePlanCount++;
1933            }
1934          }
1935        }
1936      }
1937    }
1938    // If Region did not generate any plans, it means the cluster is already balanced.
1939    // Return true indicating a success.
1940    return true;
1941  }
1942
1943  /**
1944   * @return Client info for use as prefix on an audit log string; who did an action
1945   */
1946  @Override
1947  public String getClientIdAuditPrefix() {
1948    return "Client=" + RpcServer.getRequestUserName().orElse(null)
1949        + "/" + RpcServer.getRemoteAddress().orElse(null);
1950  }
1951
1952  /**
1953   * Switch for the background CatalogJanitor thread.
1954   * Used for testing.  The thread will continue to run.  It will just be a noop
1955   * if disabled.
1956   * @param b If false, the catalog janitor won't do anything.
1957   */
1958  public void setCatalogJanitorEnabled(final boolean b) {
1959    this.catalogJanitorChore.setEnabled(b);
1960  }
1961
1962  @Override
1963  public long mergeRegions(
1964      final RegionInfo[] regionsToMerge,
1965      final boolean forcible,
1966      final long ng,
1967      final long nonce) throws IOException {
1968    checkInitialized();
1969
1970    final String mergeRegionsStr = Arrays.stream(regionsToMerge).
1971      map(r -> RegionInfo.getShortNameToLog(r)).collect(Collectors.joining(", "));
1972    return MasterProcedureUtil.submitProcedure(new NonceProcedureRunnable(this, ng, nonce) {
1973      @Override
1974      protected void run() throws IOException {
1975        getMaster().getMasterCoprocessorHost().preMergeRegions(regionsToMerge);
1976        String aid = getClientIdAuditPrefix();
1977        LOG.info("{} merge regions {}", aid, mergeRegionsStr);
1978        submitProcedure(new MergeTableRegionsProcedure(procedureExecutor.getEnvironment(),
1979            regionsToMerge, forcible));
1980        getMaster().getMasterCoprocessorHost().postMergeRegions(regionsToMerge);
1981      }
1982
1983      @Override
1984      protected String getDescription() {
1985        return "MergeTableProcedure";
1986      }
1987    });
1988  }
1989
1990  @Override
1991  public long splitRegion(final RegionInfo regionInfo, final byte[] splitRow,
1992      final long nonceGroup, final long nonce)
1993  throws IOException {
1994    checkInitialized();
1995    return MasterProcedureUtil.submitProcedure(
1996        new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
1997      @Override
1998      protected void run() throws IOException {
1999        getMaster().getMasterCoprocessorHost().preSplitRegion(regionInfo.getTable(), splitRow);
2000        LOG.info(getClientIdAuditPrefix() + " split " + regionInfo.getRegionNameAsString());
2001
2002        // Execute the operation asynchronously
2003        submitProcedure(getAssignmentManager().createSplitProcedure(regionInfo, splitRow));
2004      }
2005
2006      @Override
2007      protected String getDescription() {
2008        return "SplitTableProcedure";
2009      }
2010    });
2011  }
2012
2013  private void warmUpRegion(ServerName server, RegionInfo region) {
2014    FutureUtils.addListener(asyncClusterConnection.getRegionServerAdmin(server)
2015      .warmupRegion(RequestConverter.buildWarmupRegionRequest(region)), (r, e) -> {
2016        if (e != null) {
2017          LOG.warn("Failed to warm up region {} on server {}", region, server, e);
2018        }
2019      });
2020  }
2021
2022  // Public so can be accessed by tests. Blocks until move is done.
2023  // Replace with an async implementation from which you can get
2024  // a success/failure result.
2025  @VisibleForTesting
2026  public void move(final byte[] encodedRegionName, byte[] destServerName) throws HBaseIOException {
2027    RegionState regionState = assignmentManager.getRegionStates().
2028      getRegionState(Bytes.toString(encodedRegionName));
2029
2030    RegionInfo hri;
2031    if (regionState != null) {
2032      hri = regionState.getRegion();
2033    } else {
2034      throw new UnknownRegionException(Bytes.toStringBinary(encodedRegionName));
2035    }
2036
2037    ServerName dest;
2038    List<ServerName> exclude = hri.getTable().isSystemTable() ? assignmentManager.getExcludedServersForSystemTable()
2039        : new ArrayList<>(1);
2040    if (destServerName != null && exclude.contains(ServerName.valueOf(Bytes.toString(destServerName)))) {
2041      LOG.info(
2042          Bytes.toString(encodedRegionName) + " can not move to " + Bytes.toString(destServerName)
2043              + " because the server is in exclude list");
2044      destServerName = null;
2045    }
2046    if (destServerName == null || destServerName.length == 0) {
2047      LOG.info("Passed destination servername is null/empty so " +
2048        "choosing a server at random");
2049      exclude.add(regionState.getServerName());
2050      final List<ServerName> destServers = this.serverManager.createDestinationServersList(exclude);
2051      dest = balancer.randomAssignment(hri, destServers);
2052      if (dest == null) {
2053        LOG.debug("Unable to determine a plan to assign " + hri);
2054        return;
2055      }
2056    } else {
2057      ServerName candidate = ServerName.valueOf(Bytes.toString(destServerName));
2058      dest = balancer.randomAssignment(hri, Lists.newArrayList(candidate));
2059      if (dest == null) {
2060        LOG.debug("Unable to determine a plan to assign " + hri);
2061        return;
2062      }
2063      // TODO: What is this? I don't get it.
2064      if (dest.equals(serverName) && balancer instanceof BaseLoadBalancer
2065          && !((BaseLoadBalancer)balancer).shouldBeOnMaster(hri)) {
2066        // To avoid unnecessary region moving later by balancer. Don't put user
2067        // regions on master.
2068        LOG.debug("Skipping move of region " + hri.getRegionNameAsString()
2069          + " to avoid unnecessary region moving later by load balancer,"
2070          + " because it should not be on master");
2071        return;
2072      }
2073    }
2074
2075    if (dest.equals(regionState.getServerName())) {
2076      LOG.debug("Skipping move of region " + hri.getRegionNameAsString()
2077        + " because region already assigned to the same server " + dest + ".");
2078      return;
2079    }
2080
2081    // Now we can do the move
2082    RegionPlan rp = new RegionPlan(hri, regionState.getServerName(), dest);
2083    assert rp.getDestination() != null: rp.toString() + " " + dest;
2084
2085    try {
2086      checkInitialized();
2087      if (this.cpHost != null) {
2088        this.cpHost.preMove(hri, rp.getSource(), rp.getDestination());
2089      }
2090
2091      TransitRegionStateProcedure proc =
2092        this.assignmentManager.createMoveRegionProcedure(rp.getRegionInfo(), rp.getDestination());
2093      // Warmup the region on the destination before initiating the move.
2094      // A region server could reject the close request because it either does not
2095      // have the specified region or the region is being split.
2096      warmUpRegion(rp.getDestination(), hri);
2097
2098      LOG.info(getClientIdAuditPrefix() + " move " + rp + ", running balancer");
2099      Future<byte[]> future = ProcedureSyncWait.submitProcedure(this.procedureExecutor, proc);
2100      try {
2101        // Is this going to work? Will we throw exception on error?
2102        // TODO: CompletableFuture rather than this stunted Future.
2103        future.get();
2104      } catch (InterruptedException | ExecutionException e) {
2105        throw new HBaseIOException(e);
2106      }
2107      if (this.cpHost != null) {
2108        this.cpHost.postMove(hri, rp.getSource(), rp.getDestination());
2109      }
2110    } catch (IOException ioe) {
2111      if (ioe instanceof HBaseIOException) {
2112        throw (HBaseIOException)ioe;
2113      }
2114      throw new HBaseIOException(ioe);
2115    }
2116  }
2117
2118  @Override
2119  public long createTable(final TableDescriptor tableDescriptor, final byte[][] splitKeys,
2120      final long nonceGroup, final long nonce) throws IOException {
2121    checkInitialized();
2122    TableDescriptor desc = getMasterCoprocessorHost().preCreateTableRegionsInfos(tableDescriptor);
2123    if (desc == null) {
2124      throw new IOException("Creation for " + tableDescriptor + " is canceled by CP");
2125    }
2126    String namespace = desc.getTableName().getNamespaceAsString();
2127    this.clusterSchemaService.getNamespace(namespace);
2128
2129    RegionInfo[] newRegions = ModifyRegionUtils.createRegionInfos(desc, splitKeys);
2130    TableDescriptorChecker.sanityCheck(conf, desc);
2131
2132    return MasterProcedureUtil
2133      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
2134        @Override
2135        protected void run() throws IOException {
2136          getMaster().getMasterCoprocessorHost().preCreateTable(desc, newRegions);
2137
2138          LOG.info(getClientIdAuditPrefix() + " create " + desc);
2139
2140          // TODO: We can handle/merge duplicate requests, and differentiate the case of
2141          // TableExistsException by saying if the schema is the same or not.
2142          //
2143          // We need to wait for the procedure to potentially fail due to "prepare" sanity
2144          // checks. This will block only the beginning of the procedure. See HBASE-19953.
2145          ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch();
2146          submitProcedure(
2147            new CreateTableProcedure(procedureExecutor.getEnvironment(), desc, newRegions, latch));
2148          latch.await();
2149
2150          getMaster().getMasterCoprocessorHost().postCreateTable(desc, newRegions);
2151        }
2152
2153        @Override
2154        protected String getDescription() {
2155          return "CreateTableProcedure";
2156        }
2157      });
2158  }
2159
2160  @Override
2161  public long createSystemTable(final TableDescriptor tableDescriptor) throws IOException {
2162    if (isStopped()) {
2163      throw new MasterNotRunningException();
2164    }
2165
2166    TableName tableName = tableDescriptor.getTableName();
2167    if (!(tableName.isSystemTable())) {
2168      throw new IllegalArgumentException(
2169        "Only system table creation can use this createSystemTable API");
2170    }
2171
2172    RegionInfo[] newRegions = ModifyRegionUtils.createRegionInfos(tableDescriptor, null);
2173
2174    LOG.info(getClientIdAuditPrefix() + " create " + tableDescriptor);
2175
2176    // This special create table is called locally to master.  Therefore, no RPC means no need
2177    // to use nonce to detect duplicated RPC call.
2178    long procId = this.procedureExecutor.submitProcedure(
2179      new CreateTableProcedure(procedureExecutor.getEnvironment(), tableDescriptor, newRegions));
2180
2181    return procId;
2182  }
2183
2184  private void startActiveMasterManager(int infoPort) throws KeeperException {
2185    String backupZNode = ZNodePaths.joinZNode(
2186      zooKeeper.getZNodePaths().backupMasterAddressesZNode, serverName.toString());
2187    /*
2188    * Add a ZNode for ourselves in the backup master directory since we
2189    * may not become the active master. If so, we want the actual active
2190    * master to know we are backup masters, so that it won't assign
2191    * regions to us if so configured.
2192    *
2193    * If we become the active master later, ActiveMasterManager will delete
2194    * this node explicitly.  If we crash before then, ZooKeeper will delete
2195    * this node for us since it is ephemeral.
2196    */
2197    LOG.info("Adding backup master ZNode " + backupZNode);
2198    if (!MasterAddressTracker.setMasterAddress(zooKeeper, backupZNode, serverName, infoPort)) {
2199      LOG.warn("Failed create of " + backupZNode + " by " + serverName);
2200    }
2201    this.activeMasterManager.setInfoPort(infoPort);
2202    int timeout = conf.getInt(HConstants.ZK_SESSION_TIMEOUT, HConstants.DEFAULT_ZK_SESSION_TIMEOUT);
2203    // If we're a backup master, stall until a primary to write this address
2204    if (conf.getBoolean(HConstants.MASTER_TYPE_BACKUP, HConstants.DEFAULT_MASTER_TYPE_BACKUP)) {
2205      LOG.debug("HMaster started in backup mode. Stalling until master znode is written.");
2206      // This will only be a minute or so while the cluster starts up,
2207      // so don't worry about setting watches on the parent znode
2208      while (!activeMasterManager.hasActiveMaster()) {
2209        LOG.debug("Waiting for master address and cluster state znode to be written.");
2210        Threads.sleep(timeout);
2211      }
2212    }
2213    MonitoredTask status = TaskMonitor.get().createStatus("Master startup");
2214    status.setDescription("Master startup");
2215    try {
2216      if (activeMasterManager.blockUntilBecomingActiveMaster(timeout, status)) {
2217        finishActiveMasterInitialization(status);
2218      }
2219    } catch (Throwable t) {
2220      status.setStatus("Failed to become active: " + t.getMessage());
2221      LOG.error(HBaseMarkers.FATAL, "Failed to become active master", t);
2222      // HBASE-5680: Likely hadoop23 vs hadoop 20.x/1.x incompatibility
2223      if (t instanceof NoClassDefFoundError && t.getMessage().
2224          contains("org/apache/hadoop/hdfs/protocol/HdfsConstants$SafeModeAction")) {
2225        // improved error message for this special case
2226        abort("HBase is having a problem with its Hadoop jars.  You may need to recompile " +
2227          "HBase against Hadoop version " + org.apache.hadoop.util.VersionInfo.getVersion() +
2228          " or change your hadoop jars to start properly", t);
2229      } else {
2230        abort("Unhandled exception. Starting shutdown.", t);
2231      }
2232    } finally {
2233      status.cleanup();
2234    }
2235  }
2236
2237  private static boolean isCatalogTable(final TableName tableName) {
2238    return tableName.equals(TableName.META_TABLE_NAME);
2239  }
2240
2241  @Override
2242  public long deleteTable(
2243      final TableName tableName,
2244      final long nonceGroup,
2245      final long nonce) throws IOException {
2246    checkInitialized();
2247
2248    return MasterProcedureUtil.submitProcedure(
2249        new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
2250      @Override
2251      protected void run() throws IOException {
2252        getMaster().getMasterCoprocessorHost().preDeleteTable(tableName);
2253
2254        LOG.info(getClientIdAuditPrefix() + " delete " + tableName);
2255
2256        // TODO: We can handle/merge duplicate request
2257        //
2258        // We need to wait for the procedure to potentially fail due to "prepare" sanity
2259        // checks. This will block only the beginning of the procedure. See HBASE-19953.
2260        ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch();
2261        submitProcedure(new DeleteTableProcedure(procedureExecutor.getEnvironment(),
2262            tableName, latch));
2263        latch.await();
2264
2265        getMaster().getMasterCoprocessorHost().postDeleteTable(tableName);
2266      }
2267
2268      @Override
2269      protected String getDescription() {
2270        return "DeleteTableProcedure";
2271      }
2272    });
2273  }
2274
2275  @Override
2276  public long truncateTable(
2277      final TableName tableName,
2278      final boolean preserveSplits,
2279      final long nonceGroup,
2280      final long nonce) throws IOException {
2281    checkInitialized();
2282
2283    return MasterProcedureUtil.submitProcedure(
2284        new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
2285      @Override
2286      protected void run() throws IOException {
2287        getMaster().getMasterCoprocessorHost().preTruncateTable(tableName);
2288
2289        LOG.info(getClientIdAuditPrefix() + " truncate " + tableName);
2290        ProcedurePrepareLatch latch = ProcedurePrepareLatch.createLatch(2, 0);
2291        submitProcedure(new TruncateTableProcedure(procedureExecutor.getEnvironment(),
2292            tableName, preserveSplits, latch));
2293        latch.await();
2294
2295        getMaster().getMasterCoprocessorHost().postTruncateTable(tableName);
2296      }
2297
2298      @Override
2299      protected String getDescription() {
2300        return "TruncateTableProcedure";
2301      }
2302    });
2303  }
2304
2305  @Override
2306  public long addColumn(final TableName tableName, final ColumnFamilyDescriptor column,
2307      final long nonceGroup, final long nonce) throws IOException {
2308    checkInitialized();
2309    checkTableExists(tableName);
2310
2311    return modifyTable(tableName, new TableDescriptorGetter() {
2312
2313      @Override
2314      public TableDescriptor get() throws IOException {
2315        TableDescriptor old = getTableDescriptors().get(tableName);
2316        if (old.hasColumnFamily(column.getName())) {
2317          throw new InvalidFamilyOperationException("Column family '" + column.getNameAsString()
2318              + "' in table '" + tableName + "' already exists so cannot be added");
2319        }
2320
2321        return TableDescriptorBuilder.newBuilder(old).setColumnFamily(column).build();
2322      }
2323    }, nonceGroup, nonce, true);
2324  }
2325
2326  /**
2327   * Implement to return TableDescriptor after pre-checks
2328   */
2329  protected interface TableDescriptorGetter {
2330    TableDescriptor get() throws IOException;
2331  }
2332
2333  @Override
2334  public long modifyColumn(final TableName tableName, final ColumnFamilyDescriptor descriptor,
2335      final long nonceGroup, final long nonce) throws IOException {
2336    checkInitialized();
2337    checkTableExists(tableName);
2338    return modifyTable(tableName, new TableDescriptorGetter() {
2339
2340      @Override
2341      public TableDescriptor get() throws IOException {
2342        TableDescriptor old = getTableDescriptors().get(tableName);
2343        if (!old.hasColumnFamily(descriptor.getName())) {
2344          throw new InvalidFamilyOperationException("Family '" + descriptor.getNameAsString()
2345              + "' does not exist, so it cannot be modified");
2346        }
2347
2348        return TableDescriptorBuilder.newBuilder(old).modifyColumnFamily(descriptor).build();
2349      }
2350    }, nonceGroup, nonce, true);
2351  }
2352
2353  @Override
2354  public long deleteColumn(final TableName tableName, final byte[] columnName,
2355      final long nonceGroup, final long nonce) throws IOException {
2356    checkInitialized();
2357    checkTableExists(tableName);
2358
2359    return modifyTable(tableName, new TableDescriptorGetter() {
2360
2361      @Override
2362      public TableDescriptor get() throws IOException {
2363        TableDescriptor old = getTableDescriptors().get(tableName);
2364
2365        if (!old.hasColumnFamily(columnName)) {
2366          throw new InvalidFamilyOperationException("Family '" + Bytes.toString(columnName)
2367              + "' does not exist, so it cannot be deleted");
2368        }
2369        if (old.getColumnFamilyCount() == 1) {
2370          throw new InvalidFamilyOperationException("Family '" + Bytes.toString(columnName)
2371              + "' is the only column family in the table, so it cannot be deleted");
2372        }
2373        return TableDescriptorBuilder.newBuilder(old).removeColumnFamily(columnName).build();
2374      }
2375    }, nonceGroup, nonce, true);
2376  }
2377
2378  @Override
2379  public long enableTable(final TableName tableName, final long nonceGroup, final long nonce)
2380      throws IOException {
2381    checkInitialized();
2382
2383    return MasterProcedureUtil.submitProcedure(
2384        new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
2385      @Override
2386      protected void run() throws IOException {
2387        getMaster().getMasterCoprocessorHost().preEnableTable(tableName);
2388
2389        // Normally, it would make sense for this authorization check to exist inside
2390        // AccessController, but because the authorization check is done based on internal state
2391        // (rather than explicit permissions) we'll do the check here instead of in the
2392        // coprocessor.
2393        MasterQuotaManager quotaManager = getMasterQuotaManager();
2394        if (quotaManager != null) {
2395          if (quotaManager.isQuotaInitialized()) {
2396              SpaceQuotaSnapshot currSnapshotOfTable =
2397                  QuotaTableUtil.getCurrentSnapshotFromQuotaTable(getConnection(), tableName);
2398              if (currSnapshotOfTable != null) {
2399                SpaceQuotaStatus quotaStatus = currSnapshotOfTable.getQuotaStatus();
2400                if (quotaStatus.isInViolation()
2401                    && SpaceViolationPolicy.DISABLE == quotaStatus.getPolicy().orElse(null)) {
2402                throw new AccessDeniedException("Enabling the table '" + tableName
2403                    + "' is disallowed due to a violated space quota.");
2404              }
2405            }
2406          } else if (LOG.isTraceEnabled()) {
2407            LOG.trace("Unable to check for space quotas as the MasterQuotaManager is not enabled");
2408          }
2409        }
2410
2411        LOG.info(getClientIdAuditPrefix() + " enable " + tableName);
2412
2413        // Execute the operation asynchronously - client will check the progress of the operation
2414        // In case the request is from a <1.1 client before returning,
2415        // we want to make sure that the table is prepared to be
2416        // enabled (the table is locked and the table state is set).
2417        // Note: if the procedure throws exception, we will catch it and rethrow.
2418        final ProcedurePrepareLatch prepareLatch = ProcedurePrepareLatch.createLatch();
2419        submitProcedure(new EnableTableProcedure(procedureExecutor.getEnvironment(),
2420            tableName, prepareLatch));
2421        prepareLatch.await();
2422
2423        getMaster().getMasterCoprocessorHost().postEnableTable(tableName);
2424      }
2425
2426      @Override
2427      protected String getDescription() {
2428        return "EnableTableProcedure";
2429      }
2430    });
2431  }
2432
2433  @Override
2434  public long disableTable(final TableName tableName, final long nonceGroup, final long nonce)
2435      throws IOException {
2436    checkInitialized();
2437
2438    return MasterProcedureUtil.submitProcedure(
2439        new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
2440      @Override
2441      protected void run() throws IOException {
2442        getMaster().getMasterCoprocessorHost().preDisableTable(tableName);
2443
2444        LOG.info(getClientIdAuditPrefix() + " disable " + tableName);
2445
2446        // Execute the operation asynchronously - client will check the progress of the operation
2447        // In case the request is from a <1.1 client before returning,
2448        // we want to make sure that the table is prepared to be
2449        // enabled (the table is locked and the table state is set).
2450        // Note: if the procedure throws exception, we will catch it and rethrow.
2451        //
2452        // We need to wait for the procedure to potentially fail due to "prepare" sanity
2453        // checks. This will block only the beginning of the procedure. See HBASE-19953.
2454        final ProcedurePrepareLatch prepareLatch = ProcedurePrepareLatch.createBlockingLatch();
2455        submitProcedure(new DisableTableProcedure(procedureExecutor.getEnvironment(),
2456            tableName, false, prepareLatch));
2457        prepareLatch.await();
2458
2459        getMaster().getMasterCoprocessorHost().postDisableTable(tableName);
2460      }
2461
2462      @Override
2463      protected String getDescription() {
2464        return "DisableTableProcedure";
2465      }
2466    });
2467  }
2468
2469  private long modifyTable(final TableName tableName,
2470      final TableDescriptorGetter newDescriptorGetter, final long nonceGroup, final long nonce,
2471      final boolean shouldCheckDescriptor) throws IOException {
2472    return MasterProcedureUtil
2473        .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
2474          @Override
2475          protected void run() throws IOException {
2476            TableDescriptor oldDescriptor = getMaster().getTableDescriptors().get(tableName);
2477            TableDescriptor newDescriptor = getMaster().getMasterCoprocessorHost()
2478                .preModifyTable(tableName, oldDescriptor, newDescriptorGetter.get());
2479            TableDescriptorChecker.sanityCheck(conf, newDescriptor);
2480            LOG.info("{} modify table {} from {} to {}", getClientIdAuditPrefix(), tableName,
2481                oldDescriptor, newDescriptor);
2482
2483            // Execute the operation synchronously - wait for the operation completes before
2484            // continuing.
2485            //
2486            // We need to wait for the procedure to potentially fail due to "prepare" sanity
2487            // checks. This will block only the beginning of the procedure. See HBASE-19953.
2488            ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch();
2489            submitProcedure(new ModifyTableProcedure(procedureExecutor.getEnvironment(),
2490                newDescriptor, latch, oldDescriptor, shouldCheckDescriptor));
2491            latch.await();
2492
2493            getMaster().getMasterCoprocessorHost().postModifyTable(tableName, oldDescriptor,
2494              newDescriptor);
2495          }
2496
2497          @Override
2498          protected String getDescription() {
2499            return "ModifyTableProcedure";
2500          }
2501        });
2502
2503  }
2504
2505  @Override
2506  public long modifyTable(final TableName tableName, final TableDescriptor newDescriptor,
2507      final long nonceGroup, final long nonce) throws IOException {
2508    checkInitialized();
2509    return modifyTable(tableName, new TableDescriptorGetter() {
2510      @Override
2511      public TableDescriptor get() throws IOException {
2512        return newDescriptor;
2513      }
2514    }, nonceGroup, nonce, false);
2515
2516  }
2517
2518  public long restoreSnapshot(final SnapshotDescription snapshotDesc,
2519      final long nonceGroup, final long nonce, final boolean restoreAcl) throws IOException {
2520    checkInitialized();
2521    getSnapshotManager().checkSnapshotSupport();
2522
2523    // Ensure namespace exists. Will throw exception if non-known NS.
2524    final TableName dstTable = TableName.valueOf(snapshotDesc.getTable());
2525    getClusterSchema().getNamespace(dstTable.getNamespaceAsString());
2526
2527    return MasterProcedureUtil.submitProcedure(
2528        new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
2529      @Override
2530      protected void run() throws IOException {
2531          setProcId(
2532            getSnapshotManager().restoreOrCloneSnapshot(snapshotDesc, getNonceKey(), restoreAcl));
2533      }
2534
2535      @Override
2536      protected String getDescription() {
2537        return "RestoreSnapshotProcedure";
2538      }
2539    });
2540  }
2541
2542  private void checkTableExists(final TableName tableName)
2543      throws IOException, TableNotFoundException {
2544    if (!MetaTableAccessor.tableExists(getConnection(), tableName)) {
2545      throw new TableNotFoundException(tableName);
2546    }
2547  }
2548
2549  @Override
2550  public void checkTableModifiable(final TableName tableName)
2551      throws IOException, TableNotFoundException, TableNotDisabledException {
2552    if (isCatalogTable(tableName)) {
2553      throw new IOException("Can't modify catalog tables");
2554    }
2555    checkTableExists(tableName);
2556    TableState ts = getTableStateManager().getTableState(tableName);
2557    if (!ts.isDisabled()) {
2558      throw new TableNotDisabledException("Not DISABLED; " + ts);
2559    }
2560  }
2561
2562  public ClusterMetrics getClusterMetricsWithoutCoprocessor() throws InterruptedIOException {
2563    return getClusterMetricsWithoutCoprocessor(EnumSet.allOf(Option.class));
2564  }
2565
2566  public ClusterMetrics getClusterMetricsWithoutCoprocessor(EnumSet<Option> options)
2567      throws InterruptedIOException {
2568    ClusterMetricsBuilder builder = ClusterMetricsBuilder.newBuilder();
2569    // given that hbase1 can't submit the request with Option,
2570    // we return all information to client if the list of Option is empty.
2571    if (options.isEmpty()) {
2572      options = EnumSet.allOf(Option.class);
2573    }
2574
2575    for (Option opt : options) {
2576      switch (opt) {
2577        case HBASE_VERSION: builder.setHBaseVersion(VersionInfo.getVersion()); break;
2578        case CLUSTER_ID: builder.setClusterId(getClusterId()); break;
2579        case MASTER: builder.setMasterName(getServerName()); break;
2580        case BACKUP_MASTERS: builder.setBackerMasterNames(getBackupMasters()); break;
2581        case LIVE_SERVERS: {
2582          if (serverManager != null) {
2583            builder.setLiveServerMetrics(serverManager.getOnlineServers().entrySet().stream()
2584              .collect(Collectors.toMap(e -> e.getKey(), e -> e.getValue())));
2585          }
2586          break;
2587        }
2588        case DEAD_SERVERS: {
2589          if (serverManager != null) {
2590            builder.setDeadServerNames(new ArrayList<>(
2591              serverManager.getDeadServers().copyServerNames()));
2592          }
2593          break;
2594        }
2595        case MASTER_COPROCESSORS: {
2596          if (cpHost != null) {
2597            builder.setMasterCoprocessorNames(Arrays.asList(getMasterCoprocessors()));
2598          }
2599          break;
2600        }
2601        case REGIONS_IN_TRANSITION: {
2602          if (assignmentManager != null) {
2603            builder.setRegionsInTransition(assignmentManager.getRegionStates()
2604                .getRegionsStateInTransition());
2605          }
2606          break;
2607        }
2608        case BALANCER_ON: {
2609          if (loadBalancerTracker != null) {
2610            builder.setBalancerOn(loadBalancerTracker.isBalancerOn());
2611          }
2612          break;
2613        }
2614        case MASTER_INFO_PORT: {
2615          if (infoServer != null) {
2616            builder.setMasterInfoPort(infoServer.getPort());
2617          }
2618          break;
2619        }
2620        case SERVERS_NAME: {
2621          if (serverManager != null) {
2622            builder.setServerNames(serverManager.getOnlineServersList());
2623          }
2624          break;
2625        }
2626        case TABLE_TO_REGIONS_COUNT: {
2627          if (isActiveMaster() && isInitialized() && assignmentManager != null) {
2628            try {
2629              Map<TableName, RegionStatesCount> tableRegionStatesCountMap = new HashMap<>();
2630              Map<String, TableDescriptor> tableDescriptorMap = getTableDescriptors().getAll();
2631              for (TableDescriptor tableDescriptor : tableDescriptorMap.values()) {
2632                TableName tableName = tableDescriptor.getTableName();
2633                RegionStatesCount regionStatesCount = assignmentManager
2634                  .getRegionStatesCount(tableName);
2635                tableRegionStatesCountMap.put(tableName, regionStatesCount);
2636              }
2637              builder.setTableRegionStatesCount(tableRegionStatesCountMap);
2638            } catch (IOException e) {
2639              LOG.error("Error while populating TABLE_TO_REGIONS_COUNT for Cluster Metrics..", e);
2640            }
2641          }
2642          break;
2643        }
2644      }
2645    }
2646    return builder.build();
2647  }
2648
2649  /**
2650   * @return cluster status
2651   */
2652  public ClusterMetrics getClusterMetrics() throws IOException {
2653    return getClusterMetrics(EnumSet.allOf(Option.class));
2654  }
2655
2656  public ClusterMetrics getClusterMetrics(EnumSet<Option> options) throws IOException {
2657    if (cpHost != null) {
2658      cpHost.preGetClusterMetrics();
2659    }
2660    ClusterMetrics status = getClusterMetricsWithoutCoprocessor(options);
2661    if (cpHost != null) {
2662      cpHost.postGetClusterMetrics(status);
2663    }
2664    return status;
2665  }
2666
2667  private List<ServerName> getBackupMasters() throws InterruptedIOException {
2668    // Build Set of backup masters from ZK nodes
2669    List<String> backupMasterStrings;
2670    try {
2671      backupMasterStrings = ZKUtil.listChildrenNoWatch(this.zooKeeper,
2672        this.zooKeeper.getZNodePaths().backupMasterAddressesZNode);
2673    } catch (KeeperException e) {
2674      LOG.warn(this.zooKeeper.prefix("Unable to list backup servers"), e);
2675      backupMasterStrings = null;
2676    }
2677
2678    List<ServerName> backupMasters = Collections.emptyList();
2679    if (backupMasterStrings != null && !backupMasterStrings.isEmpty()) {
2680      backupMasters = new ArrayList<>(backupMasterStrings.size());
2681      for (String s: backupMasterStrings) {
2682        try {
2683          byte [] bytes;
2684          try {
2685            bytes = ZKUtil.getData(this.zooKeeper, ZNodePaths.joinZNode(
2686                this.zooKeeper.getZNodePaths().backupMasterAddressesZNode, s));
2687          } catch (InterruptedException e) {
2688            throw new InterruptedIOException();
2689          }
2690          if (bytes != null) {
2691            ServerName sn;
2692            try {
2693              sn = ProtobufUtil.parseServerNameFrom(bytes);
2694            } catch (DeserializationException e) {
2695              LOG.warn("Failed parse, skipping registering backup server", e);
2696              continue;
2697            }
2698            backupMasters.add(sn);
2699          }
2700        } catch (KeeperException e) {
2701          LOG.warn(this.zooKeeper.prefix("Unable to get information about " +
2702                   "backup servers"), e);
2703        }
2704      }
2705      Collections.sort(backupMasters, new Comparator<ServerName>() {
2706        @Override
2707        public int compare(ServerName s1, ServerName s2) {
2708          return s1.getServerName().compareTo(s2.getServerName());
2709        }});
2710    }
2711    return backupMasters;
2712  }
2713
2714  /**
2715   * The set of loaded coprocessors is stored in a static set. Since it's
2716   * statically allocated, it does not require that HMaster's cpHost be
2717   * initialized prior to accessing it.
2718   * @return a String representation of the set of names of the loaded coprocessors.
2719   */
2720  public static String getLoadedCoprocessors() {
2721    return CoprocessorHost.getLoadedCoprocessors().toString();
2722  }
2723
2724  /**
2725   * @return timestamp in millis when HMaster was started.
2726   */
2727  public long getMasterStartTime() {
2728    return startcode;
2729  }
2730
2731  /**
2732   * @return timestamp in millis when HMaster became the active master.
2733   */
2734  public long getMasterActiveTime() {
2735    return masterActiveTime;
2736  }
2737
2738  /**
2739   * @return timestamp in millis when HMaster finished becoming the active master
2740   */
2741  public long getMasterFinishedInitializationTime() {
2742    return masterFinishedInitializationTime;
2743  }
2744
2745  public int getNumWALFiles() {
2746    return procedureStore != null ? procedureStore.getActiveLogs().size() : 0;
2747  }
2748
2749  public WALProcedureStore getWalProcedureStore() {
2750    return procedureStore;
2751  }
2752
2753  public int getRegionServerInfoPort(final ServerName sn) {
2754    int port = this.serverManager.getInfoPort(sn);
2755    return port == 0 ? conf.getInt(HConstants.REGIONSERVER_INFO_PORT,
2756      HConstants.DEFAULT_REGIONSERVER_INFOPORT) : port;
2757  }
2758
2759  @Override
2760  public String getRegionServerVersion(ServerName sn) {
2761    // Will return "0.0.0" if the server is not online to prevent move system region to unknown
2762    // version RS.
2763    return this.serverManager.getVersion(sn);
2764  }
2765
2766  @Override
2767  public void checkIfShouldMoveSystemRegionAsync() {
2768    assignmentManager.checkIfShouldMoveSystemRegionAsync();
2769  }
2770
2771  /**
2772   * @return array of coprocessor SimpleNames.
2773   */
2774  public String[] getMasterCoprocessors() {
2775    Set<String> masterCoprocessors = getMasterCoprocessorHost().getCoprocessors();
2776    return masterCoprocessors.toArray(new String[masterCoprocessors.size()]);
2777  }
2778
2779  @Override
2780  public void abort(String reason, Throwable cause) {
2781    if (isAborted() || isStopped()) {
2782      return;
2783    }
2784    setAbortRequested();
2785    if (cpHost != null) {
2786      // HBASE-4014: dump a list of loaded coprocessors.
2787      LOG.error(HBaseMarkers.FATAL, "Master server abort: loaded coprocessors are: " +
2788          getLoadedCoprocessors());
2789    }
2790    String msg = "***** ABORTING master " + this + ": " + reason + " *****";
2791    if (cause != null) {
2792      LOG.error(HBaseMarkers.FATAL, msg, cause);
2793    } else {
2794      LOG.error(HBaseMarkers.FATAL, msg);
2795    }
2796
2797    try {
2798      stopMaster();
2799    } catch (IOException e) {
2800      LOG.error("Exception occurred while stopping master", e);
2801    }
2802  }
2803
2804  @Override
2805  public ZKWatcher getZooKeeper() {
2806    return zooKeeper;
2807  }
2808
2809  @Override
2810  public MasterCoprocessorHost getMasterCoprocessorHost() {
2811    return cpHost;
2812  }
2813
2814  @Override
2815  public MasterQuotaManager getMasterQuotaManager() {
2816    return quotaManager;
2817  }
2818
2819  @Override
2820  public ProcedureExecutor<MasterProcedureEnv> getMasterProcedureExecutor() {
2821    return procedureExecutor;
2822  }
2823
2824  @Override
2825  public ServerName getServerName() {
2826    return this.serverName;
2827  }
2828
2829  @Override
2830  public AssignmentManager getAssignmentManager() {
2831    return this.assignmentManager;
2832  }
2833
2834  @Override
2835  public CatalogJanitor getCatalogJanitor() {
2836    return this.catalogJanitorChore;
2837  }
2838
2839  public MemoryBoundedLogMessageBuffer getRegionServerFatalLogBuffer() {
2840    return rsFatals;
2841  }
2842
2843  /**
2844   * Shutdown the cluster.
2845   * Master runs a coordinated stop of all RegionServers and then itself.
2846   */
2847  public void shutdown() throws IOException {
2848    if (cpHost != null) {
2849      cpHost.preShutdown();
2850    }
2851    // Tell the servermanager cluster shutdown has been called. This makes it so when Master is
2852    // last running server, it'll stop itself. Next, we broadcast the cluster shutdown by setting
2853    // the cluster status as down. RegionServers will notice this change in state and will start
2854    // shutting themselves down. When last has exited, Master can go down.
2855    if (this.serverManager != null) {
2856      this.serverManager.shutdownCluster();
2857    }
2858    if (this.clusterStatusTracker != null) {
2859      try {
2860        this.clusterStatusTracker.setClusterDown();
2861      } catch (KeeperException e) {
2862        LOG.error("ZooKeeper exception trying to set cluster as down in ZK", e);
2863      }
2864    }
2865    // Stop the procedure executor. Will stop any ongoing assign, unassign, server crash etc.,
2866    // processing so we can go down.
2867    if (this.procedureExecutor != null) {
2868      this.procedureExecutor.stop();
2869    }
2870    // Shutdown our cluster connection. This will kill any hosted RPCs that might be going on;
2871    // this is what we want especially if the Master is in startup phase doing call outs to
2872    // hbase:meta, etc. when cluster is down. Without ths connection close, we'd have to wait on
2873    // the rpc to timeout.
2874    if (this.asyncClusterConnection != null) {
2875      this.asyncClusterConnection.close();
2876    }
2877  }
2878
2879  public void stopMaster() throws IOException {
2880    if (cpHost != null) {
2881      cpHost.preStopMaster();
2882    }
2883    stop("Stopped by " + Thread.currentThread().getName());
2884  }
2885
2886  @Override
2887  public void stop(String msg) {
2888    if (!isStopped()) {
2889      super.stop(msg);
2890      if (this.activeMasterManager != null) {
2891        this.activeMasterManager.stop();
2892      }
2893    }
2894  }
2895
2896  @VisibleForTesting
2897  protected void checkServiceStarted() throws ServerNotRunningYetException {
2898    if (!serviceStarted) {
2899      throw new ServerNotRunningYetException("Server is not running yet");
2900    }
2901  }
2902
2903  public static class MasterStoppedException extends DoNotRetryIOException {
2904    MasterStoppedException() {
2905      super();
2906    }
2907  }
2908
2909  void checkInitialized() throws PleaseHoldException, ServerNotRunningYetException,
2910      MasterNotRunningException, MasterStoppedException {
2911    checkServiceStarted();
2912    if (!isInitialized()) {
2913      throw new PleaseHoldException("Master is initializing");
2914    }
2915    if (isStopped()) {
2916      throw new MasterStoppedException();
2917    }
2918  }
2919
2920  /**
2921   * Report whether this master is currently the active master or not.
2922   * If not active master, we are parked on ZK waiting to become active.
2923   *
2924   * This method is used for testing.
2925   *
2926   * @return true if active master, false if not.
2927   */
2928  @Override
2929  public boolean isActiveMaster() {
2930    return activeMaster;
2931  }
2932
2933  /**
2934   * Report whether this master has completed with its initialization and is
2935   * ready.  If ready, the master is also the active master.  A standby master
2936   * is never ready.
2937   *
2938   * This method is used for testing.
2939   *
2940   * @return true if master is ready to go, false if not.
2941   */
2942  @Override
2943  public boolean isInitialized() {
2944    return initialized.isReady();
2945  }
2946
2947  /**
2948   * Report whether this master is in maintenance mode.
2949   *
2950   * @return true if master is in maintenanceMode
2951   */
2952  @Override
2953  public boolean isInMaintenanceMode() {
2954    return maintenanceMode;
2955  }
2956
2957  @VisibleForTesting
2958  public void setInitialized(boolean isInitialized) {
2959    procedureExecutor.getEnvironment().setEventReady(initialized, isInitialized);
2960  }
2961
2962  @Override
2963  public ProcedureEvent<?> getInitializedEvent() {
2964    return initialized;
2965  }
2966
2967  /**
2968   * Compute the average load across all region servers.
2969   * Currently, this uses a very naive computation - just uses the number of
2970   * regions being served, ignoring stats about number of requests.
2971   * @return the average load
2972   */
2973  public double getAverageLoad() {
2974    if (this.assignmentManager == null) {
2975      return 0;
2976    }
2977
2978    RegionStates regionStates = this.assignmentManager.getRegionStates();
2979    if (regionStates == null) {
2980      return 0;
2981    }
2982    return regionStates.getAverageLoad();
2983  }
2984
2985  /*
2986   * @return the count of region split plans executed
2987   */
2988  public long getSplitPlanCount() {
2989    return splitPlanCount;
2990  }
2991
2992  /*
2993   * @return the count of region merge plans executed
2994   */
2995  public long getMergePlanCount() {
2996    return mergePlanCount;
2997  }
2998
2999  @Override
3000  public boolean registerService(Service instance) {
3001    /*
3002     * No stacking of instances is allowed for a single service name
3003     */
3004    Descriptors.ServiceDescriptor serviceDesc = instance.getDescriptorForType();
3005    String serviceName = CoprocessorRpcUtils.getServiceName(serviceDesc);
3006    if (coprocessorServiceHandlers.containsKey(serviceName)) {
3007      LOG.error("Coprocessor service "+serviceName+
3008          " already registered, rejecting request from "+instance
3009      );
3010      return false;
3011    }
3012
3013    coprocessorServiceHandlers.put(serviceName, instance);
3014    if (LOG.isDebugEnabled()) {
3015      LOG.debug("Registered master coprocessor service: service="+serviceName);
3016    }
3017    return true;
3018  }
3019
3020  /**
3021   * Utility for constructing an instance of the passed HMaster class.
3022   * @param masterClass
3023   * @return HMaster instance.
3024   */
3025  public static HMaster constructMaster(Class<? extends HMaster> masterClass,
3026      final Configuration conf)  {
3027    try {
3028      Constructor<? extends HMaster> c = masterClass.getConstructor(Configuration.class);
3029      return c.newInstance(conf);
3030    } catch(Exception e) {
3031      Throwable error = e;
3032      if (e instanceof InvocationTargetException &&
3033          ((InvocationTargetException)e).getTargetException() != null) {
3034        error = ((InvocationTargetException)e).getTargetException();
3035      }
3036      throw new RuntimeException("Failed construction of Master: " + masterClass.toString() + ". "
3037        , error);
3038    }
3039  }
3040
3041  /**
3042   * @see org.apache.hadoop.hbase.master.HMasterCommandLine
3043   */
3044  public static void main(String [] args) {
3045    LOG.info("STARTING service " + HMaster.class.getSimpleName());
3046    VersionInfo.logVersion();
3047    new HMasterCommandLine(HMaster.class).doMain(args);
3048  }
3049
3050  public HFileCleaner getHFileCleaner() {
3051    return this.hfileCleaner;
3052  }
3053
3054  public LogCleaner getLogCleaner() {
3055    return this.logCleaner;
3056  }
3057
3058  /**
3059   * @return the underlying snapshot manager
3060   */
3061  @Override
3062  public SnapshotManager getSnapshotManager() {
3063    return this.snapshotManager;
3064  }
3065
3066  /**
3067   * @return the underlying MasterProcedureManagerHost
3068   */
3069  @Override
3070  public MasterProcedureManagerHost getMasterProcedureManagerHost() {
3071    return mpmHost;
3072  }
3073
3074  @Override
3075  public ClusterSchema getClusterSchema() {
3076    return this.clusterSchemaService;
3077  }
3078
3079  /**
3080   * Create a new Namespace.
3081   * @param namespaceDescriptor descriptor for new Namespace
3082   * @param nonceGroup Identifier for the source of the request, a client or process.
3083   * @param nonce A unique identifier for this operation from the client or process identified by
3084   * <code>nonceGroup</code> (the source must ensure each operation gets a unique id).
3085   * @return procedure id
3086   */
3087  long createNamespace(final NamespaceDescriptor namespaceDescriptor, final long nonceGroup,
3088      final long nonce) throws IOException {
3089    checkInitialized();
3090
3091    TableName.isLegalNamespaceName(Bytes.toBytes(namespaceDescriptor.getName()));
3092
3093    return MasterProcedureUtil.submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this,
3094          nonceGroup, nonce) {
3095      @Override
3096      protected void run() throws IOException {
3097        getMaster().getMasterCoprocessorHost().preCreateNamespace(namespaceDescriptor);
3098        // We need to wait for the procedure to potentially fail due to "prepare" sanity
3099        // checks. This will block only the beginning of the procedure. See HBASE-19953.
3100        ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch();
3101        LOG.info(getClientIdAuditPrefix() + " creating " + namespaceDescriptor);
3102        // Execute the operation synchronously - wait for the operation to complete before
3103        // continuing.
3104        setProcId(getClusterSchema().createNamespace(namespaceDescriptor, getNonceKey(), latch));
3105        latch.await();
3106        getMaster().getMasterCoprocessorHost().postCreateNamespace(namespaceDescriptor);
3107      }
3108
3109      @Override
3110      protected String getDescription() {
3111        return "CreateNamespaceProcedure";
3112      }
3113    });
3114  }
3115
3116  /**
3117   * Modify an existing Namespace.
3118   * @param nonceGroup Identifier for the source of the request, a client or process.
3119   * @param nonce A unique identifier for this operation from the client or process identified by
3120   * <code>nonceGroup</code> (the source must ensure each operation gets a unique id).
3121   * @return procedure id
3122   */
3123  long modifyNamespace(final NamespaceDescriptor newNsDescriptor, final long nonceGroup,
3124      final long nonce) throws IOException {
3125    checkInitialized();
3126
3127    TableName.isLegalNamespaceName(Bytes.toBytes(newNsDescriptor.getName()));
3128
3129    return MasterProcedureUtil.submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this,
3130          nonceGroup, nonce) {
3131      @Override
3132      protected void run() throws IOException {
3133        NamespaceDescriptor oldNsDescriptor = getNamespace(newNsDescriptor.getName());
3134        getMaster().getMasterCoprocessorHost().preModifyNamespace(oldNsDescriptor, newNsDescriptor);
3135        // We need to wait for the procedure to potentially fail due to "prepare" sanity
3136        // checks. This will block only the beginning of the procedure. See HBASE-19953.
3137        ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch();
3138        LOG.info(getClientIdAuditPrefix() + " modify " + newNsDescriptor);
3139        // Execute the operation synchronously - wait for the operation to complete before
3140        // continuing.
3141        setProcId(getClusterSchema().modifyNamespace(newNsDescriptor, getNonceKey(), latch));
3142        latch.await();
3143        getMaster().getMasterCoprocessorHost().postModifyNamespace(oldNsDescriptor,
3144          newNsDescriptor);
3145      }
3146
3147      @Override
3148      protected String getDescription() {
3149        return "ModifyNamespaceProcedure";
3150      }
3151    });
3152  }
3153
3154  /**
3155   * Delete an existing Namespace. Only empty Namespaces (no tables) can be removed.
3156   * @param nonceGroup Identifier for the source of the request, a client or process.
3157   * @param nonce A unique identifier for this operation from the client or process identified by
3158   * <code>nonceGroup</code> (the source must ensure each operation gets a unique id).
3159   * @return procedure id
3160   */
3161  long deleteNamespace(final String name, final long nonceGroup, final long nonce)
3162      throws IOException {
3163    checkInitialized();
3164
3165    return MasterProcedureUtil.submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this,
3166          nonceGroup, nonce) {
3167      @Override
3168      protected void run() throws IOException {
3169        getMaster().getMasterCoprocessorHost().preDeleteNamespace(name);
3170        LOG.info(getClientIdAuditPrefix() + " delete " + name);
3171        // Execute the operation synchronously - wait for the operation to complete before
3172        // continuing.
3173        //
3174        // We need to wait for the procedure to potentially fail due to "prepare" sanity
3175        // checks. This will block only the beginning of the procedure. See HBASE-19953.
3176        ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch();
3177        setProcId(submitProcedure(
3178              new DeleteNamespaceProcedure(procedureExecutor.getEnvironment(), name, latch)));
3179        latch.await();
3180        // Will not be invoked in the face of Exception thrown by the Procedure's execution
3181        getMaster().getMasterCoprocessorHost().postDeleteNamespace(name);
3182      }
3183
3184      @Override
3185      protected String getDescription() {
3186        return "DeleteNamespaceProcedure";
3187      }
3188    });
3189  }
3190
3191  /**
3192   * Get a Namespace
3193   * @param name Name of the Namespace
3194   * @return Namespace descriptor for <code>name</code>
3195   */
3196  NamespaceDescriptor getNamespace(String name) throws IOException {
3197    checkInitialized();
3198    if (this.cpHost != null) this.cpHost.preGetNamespaceDescriptor(name);
3199    NamespaceDescriptor nsd = this.clusterSchemaService.getNamespace(name);
3200    if (this.cpHost != null) this.cpHost.postGetNamespaceDescriptor(nsd);
3201    return nsd;
3202  }
3203
3204  /**
3205   * Get all Namespaces
3206   * @return All Namespace descriptors
3207   */
3208  List<NamespaceDescriptor> getNamespaces() throws IOException {
3209    checkInitialized();
3210    final List<NamespaceDescriptor> nsds = new ArrayList<>();
3211    if (cpHost != null) {
3212      cpHost.preListNamespaceDescriptors(nsds);
3213    }
3214    nsds.addAll(this.clusterSchemaService.getNamespaces());
3215    if (this.cpHost != null) {
3216      this.cpHost.postListNamespaceDescriptors(nsds);
3217    }
3218    return nsds;
3219  }
3220
3221  /**
3222   * List namespace names
3223   * @return All namespace names
3224   */
3225  public List<String> listNamespaces() throws IOException {
3226    checkInitialized();
3227    List<String> namespaces = new ArrayList<>();
3228    if (cpHost != null) {
3229      cpHost.preListNamespaces(namespaces);
3230    }
3231    for (NamespaceDescriptor namespace : clusterSchemaService.getNamespaces()) {
3232      namespaces.add(namespace.getName());
3233    }
3234    if (cpHost != null) {
3235      cpHost.postListNamespaces(namespaces);
3236    }
3237    return namespaces;
3238  }
3239
3240  @Override
3241  public List<TableName> listTableNamesByNamespace(String name) throws IOException {
3242    checkInitialized();
3243    return listTableNames(name, null, true);
3244  }
3245
3246  @Override
3247  public List<TableDescriptor> listTableDescriptorsByNamespace(String name) throws IOException {
3248    checkInitialized();
3249    return listTableDescriptors(name, null, null, true);
3250  }
3251
3252  @Override
3253  public boolean abortProcedure(final long procId, final boolean mayInterruptIfRunning)
3254      throws IOException {
3255    if (cpHost != null) {
3256      cpHost.preAbortProcedure(this.procedureExecutor, procId);
3257    }
3258
3259    final boolean result = this.procedureExecutor.abort(procId, mayInterruptIfRunning);
3260
3261    if (cpHost != null) {
3262      cpHost.postAbortProcedure();
3263    }
3264
3265    return result;
3266  }
3267
3268  @Override
3269  public List<Procedure<?>> getProcedures() throws IOException {
3270    if (cpHost != null) {
3271      cpHost.preGetProcedures();
3272    }
3273
3274    @SuppressWarnings({ "unchecked", "rawtypes" })
3275    List<Procedure<?>> procList = (List) this.procedureExecutor.getProcedures();
3276
3277    if (cpHost != null) {
3278      cpHost.postGetProcedures(procList);
3279    }
3280
3281    return procList;
3282  }
3283
3284  @Override
3285  public List<LockedResource> getLocks() throws IOException {
3286    if (cpHost != null) {
3287      cpHost.preGetLocks();
3288    }
3289
3290    MasterProcedureScheduler procedureScheduler =
3291      procedureExecutor.getEnvironment().getProcedureScheduler();
3292
3293    final List<LockedResource> lockedResources = procedureScheduler.getLocks();
3294
3295    if (cpHost != null) {
3296      cpHost.postGetLocks(lockedResources);
3297    }
3298
3299    return lockedResources;
3300  }
3301
3302  /**
3303   * Returns the list of table descriptors that match the specified request
3304   * @param namespace the namespace to query, or null if querying for all
3305   * @param regex The regular expression to match against, or null if querying for all
3306   * @param tableNameList the list of table names, or null if querying for all
3307   * @param includeSysTables False to match only against userspace tables
3308   * @return the list of table descriptors
3309   */
3310  public List<TableDescriptor> listTableDescriptors(final String namespace, final String regex,
3311      final List<TableName> tableNameList, final boolean includeSysTables)
3312  throws IOException {
3313    List<TableDescriptor> htds = new ArrayList<>();
3314    if (cpHost != null) {
3315      cpHost.preGetTableDescriptors(tableNameList, htds, regex);
3316    }
3317    htds = getTableDescriptors(htds, namespace, regex, tableNameList, includeSysTables);
3318    if (cpHost != null) {
3319      cpHost.postGetTableDescriptors(tableNameList, htds, regex);
3320    }
3321    return htds;
3322  }
3323
3324  /**
3325   * Returns the list of table names that match the specified request
3326   * @param regex The regular expression to match against, or null if querying for all
3327   * @param namespace the namespace to query, or null if querying for all
3328   * @param includeSysTables False to match only against userspace tables
3329   * @return the list of table names
3330   */
3331  public List<TableName> listTableNames(final String namespace, final String regex,
3332      final boolean includeSysTables) throws IOException {
3333    List<TableDescriptor> htds = new ArrayList<>();
3334    if (cpHost != null) {
3335      cpHost.preGetTableNames(htds, regex);
3336    }
3337    htds = getTableDescriptors(htds, namespace, regex, null, includeSysTables);
3338    if (cpHost != null) {
3339      cpHost.postGetTableNames(htds, regex);
3340    }
3341    List<TableName> result = new ArrayList<>(htds.size());
3342    for (TableDescriptor htd: htds) result.add(htd.getTableName());
3343    return result;
3344  }
3345
3346  /**
3347   * @return list of table table descriptors after filtering by regex and whether to include system
3348   *    tables, etc.
3349   * @throws IOException
3350   */
3351  private List<TableDescriptor> getTableDescriptors(final List<TableDescriptor> htds,
3352      final String namespace, final String regex, final List<TableName> tableNameList,
3353      final boolean includeSysTables)
3354  throws IOException {
3355    if (tableNameList == null || tableNameList.isEmpty()) {
3356      // request for all TableDescriptors
3357      Collection<TableDescriptor> allHtds;
3358      if (namespace != null && namespace.length() > 0) {
3359        // Do a check on the namespace existence. Will fail if does not exist.
3360        this.clusterSchemaService.getNamespace(namespace);
3361        allHtds = tableDescriptors.getByNamespace(namespace).values();
3362      } else {
3363        allHtds = tableDescriptors.getAll().values();
3364      }
3365      for (TableDescriptor desc: allHtds) {
3366        if (tableStateManager.isTablePresent(desc.getTableName())
3367            && (includeSysTables || !desc.getTableName().isSystemTable())) {
3368          htds.add(desc);
3369        }
3370      }
3371    } else {
3372      for (TableName s: tableNameList) {
3373        if (tableStateManager.isTablePresent(s)) {
3374          TableDescriptor desc = tableDescriptors.get(s);
3375          if (desc != null) {
3376            htds.add(desc);
3377          }
3378        }
3379      }
3380    }
3381
3382    // Retains only those matched by regular expression.
3383    if (regex != null) filterTablesByRegex(htds, Pattern.compile(regex));
3384    return htds;
3385  }
3386
3387  /**
3388   * Removes the table descriptors that don't match the pattern.
3389   * @param descriptors list of table descriptors to filter
3390   * @param pattern the regex to use
3391   */
3392  private static void filterTablesByRegex(final Collection<TableDescriptor> descriptors,
3393      final Pattern pattern) {
3394    final String defaultNS = NamespaceDescriptor.DEFAULT_NAMESPACE_NAME_STR;
3395    Iterator<TableDescriptor> itr = descriptors.iterator();
3396    while (itr.hasNext()) {
3397      TableDescriptor htd = itr.next();
3398      String tableName = htd.getTableName().getNameAsString();
3399      boolean matched = pattern.matcher(tableName).matches();
3400      if (!matched && htd.getTableName().getNamespaceAsString().equals(defaultNS)) {
3401        matched = pattern.matcher(defaultNS + TableName.NAMESPACE_DELIM + tableName).matches();
3402      }
3403      if (!matched) {
3404        itr.remove();
3405      }
3406    }
3407  }
3408
3409  @Override
3410  public long getLastMajorCompactionTimestamp(TableName table) throws IOException {
3411    return getClusterMetrics(EnumSet.of(Option.LIVE_SERVERS))
3412        .getLastMajorCompactionTimestamp(table);
3413  }
3414
3415  @Override
3416  public long getLastMajorCompactionTimestampForRegion(byte[] regionName) throws IOException {
3417    return getClusterMetrics(EnumSet.of(Option.LIVE_SERVERS))
3418        .getLastMajorCompactionTimestamp(regionName);
3419  }
3420
3421  /**
3422   * Gets the mob file compaction state for a specific table.
3423   * Whether all the mob files are selected is known during the compaction execution, but
3424   * the statistic is done just before compaction starts, it is hard to know the compaction
3425   * type at that time, so the rough statistics are chosen for the mob file compaction. Only two
3426   * compaction states are available, CompactionState.MAJOR_AND_MINOR and CompactionState.NONE.
3427   * @param tableName The current table name.
3428   * @return If a given table is in mob file compaction now.
3429   */
3430  public CompactionState getMobCompactionState(TableName tableName) {
3431    AtomicInteger compactionsCount = mobCompactionStates.get(tableName);
3432    if (compactionsCount != null && compactionsCount.get() != 0) {
3433      return CompactionState.MAJOR_AND_MINOR;
3434    }
3435    return CompactionState.NONE;
3436  }
3437
3438  public void reportMobCompactionStart(TableName tableName) throws IOException {
3439    IdLock.Entry lockEntry = null;
3440    try {
3441      lockEntry = mobCompactionLock.getLockEntry(tableName.hashCode());
3442      AtomicInteger compactionsCount = mobCompactionStates.get(tableName);
3443      if (compactionsCount == null) {
3444        compactionsCount = new AtomicInteger(0);
3445        mobCompactionStates.put(tableName, compactionsCount);
3446      }
3447      compactionsCount.incrementAndGet();
3448    } finally {
3449      if (lockEntry != null) {
3450        mobCompactionLock.releaseLockEntry(lockEntry);
3451      }
3452    }
3453  }
3454
3455  public void reportMobCompactionEnd(TableName tableName) throws IOException {
3456    IdLock.Entry lockEntry = null;
3457    try {
3458      lockEntry = mobCompactionLock.getLockEntry(tableName.hashCode());
3459      AtomicInteger compactionsCount = mobCompactionStates.get(tableName);
3460      if (compactionsCount != null) {
3461        int count = compactionsCount.decrementAndGet();
3462        // remove the entry if the count is 0.
3463        if (count == 0) {
3464          mobCompactionStates.remove(tableName);
3465        }
3466      }
3467    } finally {
3468      if (lockEntry != null) {
3469        mobCompactionLock.releaseLockEntry(lockEntry);
3470      }
3471    }
3472  }
3473
3474  /**
3475   * Requests mob compaction.
3476   * @param tableName The table the compact.
3477   * @param columns The compacted columns.
3478   * @param allFiles Whether add all mob files into the compaction.
3479   */
3480  public void requestMobCompaction(TableName tableName,
3481                                   List<ColumnFamilyDescriptor> columns, boolean allFiles) throws IOException {
3482    mobCompactThread.requestMobCompaction(conf, getFileSystem(), tableName, columns, allFiles);
3483  }
3484
3485  /**
3486   * Queries the state of the {@link LoadBalancerTracker}. If the balancer is not initialized,
3487   * false is returned.
3488   *
3489   * @return The state of the load balancer, or false if the load balancer isn't defined.
3490   */
3491  public boolean isBalancerOn() {
3492    return !isInMaintenanceMode()
3493        && loadBalancerTracker != null
3494        && loadBalancerTracker.isBalancerOn();
3495  }
3496
3497  /**
3498   * Queries the state of the {@link RegionNormalizerTracker}. If it's not initialized,
3499   * false is returned.
3500   */
3501  public boolean isNormalizerOn() {
3502    return !isInMaintenanceMode()
3503        && regionNormalizerTracker != null
3504        && regionNormalizerTracker.isNormalizerOn();
3505  }
3506
3507  /**
3508   * Queries the state of the {@link SplitOrMergeTracker}. If it is not initialized,
3509   * false is returned. If switchType is illegal, false will return.
3510   * @param switchType see {@link org.apache.hadoop.hbase.client.MasterSwitchType}
3511   * @return The state of the switch
3512   */
3513  @Override
3514  public boolean isSplitOrMergeEnabled(MasterSwitchType switchType) {
3515    return !isInMaintenanceMode()
3516        && splitOrMergeTracker != null
3517        && splitOrMergeTracker.isSplitOrMergeEnabled(switchType);
3518  }
3519
3520  /**
3521   * Fetch the configured {@link LoadBalancer} class name. If none is set, a default is returned.
3522   *
3523   * @return The name of the {@link LoadBalancer} in use.
3524   */
3525  public String getLoadBalancerClassName() {
3526    return conf.get(HConstants.HBASE_MASTER_LOADBALANCER_CLASS, LoadBalancerFactory
3527        .getDefaultLoadBalancerClass().getName());
3528  }
3529
3530  /**
3531   * @return RegionNormalizerTracker instance
3532   */
3533  public RegionNormalizerTracker getRegionNormalizerTracker() {
3534    return regionNormalizerTracker;
3535  }
3536
3537  public SplitOrMergeTracker getSplitOrMergeTracker() {
3538    return splitOrMergeTracker;
3539  }
3540
3541  @Override
3542  public LoadBalancer getLoadBalancer() {
3543    return balancer;
3544  }
3545
3546  @Override
3547  public FavoredNodesManager getFavoredNodesManager() {
3548    return favoredNodesManager;
3549  }
3550
3551  private long executePeerProcedure(AbstractPeerProcedure<?> procedure) throws IOException {
3552    long procId = procedureExecutor.submitProcedure(procedure);
3553    procedure.getLatch().await();
3554    return procId;
3555  }
3556
3557  @Override
3558  public long addReplicationPeer(String peerId, ReplicationPeerConfig peerConfig, boolean enabled)
3559      throws ReplicationException, IOException {
3560    LOG.info(getClientIdAuditPrefix() + " creating replication peer, id=" + peerId + ", config=" +
3561      peerConfig + ", state=" + (enabled ? "ENABLED" : "DISABLED"));
3562    return executePeerProcedure(new AddPeerProcedure(peerId, peerConfig, enabled));
3563  }
3564
3565  @Override
3566  public long removeReplicationPeer(String peerId) throws ReplicationException, IOException {
3567    LOG.info(getClientIdAuditPrefix() + " removing replication peer, id=" + peerId);
3568    return executePeerProcedure(new RemovePeerProcedure(peerId));
3569  }
3570
3571  @Override
3572  public long enableReplicationPeer(String peerId) throws ReplicationException, IOException {
3573    LOG.info(getClientIdAuditPrefix() + " enable replication peer, id=" + peerId);
3574    return executePeerProcedure(new EnablePeerProcedure(peerId));
3575  }
3576
3577  @Override
3578  public long disableReplicationPeer(String peerId) throws ReplicationException, IOException {
3579    LOG.info(getClientIdAuditPrefix() + " disable replication peer, id=" + peerId);
3580    return executePeerProcedure(new DisablePeerProcedure(peerId));
3581  }
3582
3583  @Override
3584  public ReplicationPeerConfig getReplicationPeerConfig(String peerId)
3585      throws ReplicationException, IOException {
3586    if (cpHost != null) {
3587      cpHost.preGetReplicationPeerConfig(peerId);
3588    }
3589    LOG.info(getClientIdAuditPrefix() + " get replication peer config, id=" + peerId);
3590    ReplicationPeerConfig peerConfig = this.replicationPeerManager.getPeerConfig(peerId)
3591        .orElseThrow(() -> new ReplicationPeerNotFoundException(peerId));
3592    if (cpHost != null) {
3593      cpHost.postGetReplicationPeerConfig(peerId);
3594    }
3595    return peerConfig;
3596  }
3597
3598  @Override
3599  public long updateReplicationPeerConfig(String peerId, ReplicationPeerConfig peerConfig)
3600      throws ReplicationException, IOException {
3601    LOG.info(getClientIdAuditPrefix() + " update replication peer config, id=" + peerId +
3602      ", config=" + peerConfig);
3603    return executePeerProcedure(new UpdatePeerConfigProcedure(peerId, peerConfig));
3604  }
3605
3606  @Override
3607  public List<ReplicationPeerDescription> listReplicationPeers(String regex)
3608      throws ReplicationException, IOException {
3609    if (cpHost != null) {
3610      cpHost.preListReplicationPeers(regex);
3611    }
3612    LOG.debug("{} list replication peers, regex={}", getClientIdAuditPrefix(), regex);
3613    Pattern pattern = regex == null ? null : Pattern.compile(regex);
3614    List<ReplicationPeerDescription> peers =
3615      this.replicationPeerManager.listPeers(pattern);
3616    if (cpHost != null) {
3617      cpHost.postListReplicationPeers(regex);
3618    }
3619    return peers;
3620  }
3621
3622  @Override
3623  public long transitReplicationPeerSyncReplicationState(String peerId, SyncReplicationState state)
3624    throws ReplicationException, IOException {
3625    LOG.info(
3626      getClientIdAuditPrefix() +
3627        " transit current cluster state to {} in a synchronous replication peer id={}",
3628      state, peerId);
3629    return executePeerProcedure(new TransitPeerSyncReplicationStateProcedure(peerId, state));
3630  }
3631
3632  /**
3633   * Mark region server(s) as decommissioned (previously called 'draining') to prevent additional
3634   * regions from getting assigned to them. Also unload the regions on the servers asynchronously.0
3635   * @param servers Region servers to decommission.
3636   */
3637  public void decommissionRegionServers(final List<ServerName> servers, final boolean offload)
3638      throws HBaseIOException {
3639    List<ServerName> serversAdded = new ArrayList<>(servers.size());
3640    // Place the decommission marker first.
3641    String parentZnode = getZooKeeper().getZNodePaths().drainingZNode;
3642    for (ServerName server : servers) {
3643      try {
3644        String node = ZNodePaths.joinZNode(parentZnode, server.getServerName());
3645        ZKUtil.createAndFailSilent(getZooKeeper(), node);
3646      } catch (KeeperException ke) {
3647        throw new HBaseIOException(
3648          this.zooKeeper.prefix("Unable to decommission '" + server.getServerName() + "'."), ke);
3649      }
3650      if (this.serverManager.addServerToDrainList(server)) {
3651        serversAdded.add(server);
3652      }
3653    }
3654    // Move the regions off the decommissioned servers.
3655    if (offload) {
3656      final List<ServerName> destServers = this.serverManager.createDestinationServersList();
3657      for (ServerName server : serversAdded) {
3658        final List<RegionInfo> regionsOnServer = this.assignmentManager.getRegionsOnServer(server);
3659        for (RegionInfo hri : regionsOnServer) {
3660          ServerName dest = balancer.randomAssignment(hri, destServers);
3661          if (dest == null) {
3662            throw new HBaseIOException("Unable to determine a plan to move " + hri);
3663          }
3664          RegionPlan rp = new RegionPlan(hri, server, dest);
3665          this.assignmentManager.moveAsync(rp);
3666        }
3667      }
3668    }
3669  }
3670
3671  /**
3672   * List region servers marked as decommissioned (previously called 'draining') to not get regions
3673   * assigned to them.
3674   * @return List of decommissioned servers.
3675   */
3676  public List<ServerName> listDecommissionedRegionServers() {
3677    return this.serverManager.getDrainingServersList();
3678  }
3679
3680  /**
3681   * Remove decommission marker (previously called 'draining') from a region server to allow regions
3682   * assignments. Load regions onto the server asynchronously if a list of regions is given
3683   * @param server Region server to remove decommission marker from.
3684   */
3685  public void recommissionRegionServer(final ServerName server,
3686      final List<byte[]> encodedRegionNames) throws IOException {
3687    // Remove the server from decommissioned (draining) server list.
3688    String parentZnode = getZooKeeper().getZNodePaths().drainingZNode;
3689    String node = ZNodePaths.joinZNode(parentZnode, server.getServerName());
3690    try {
3691      ZKUtil.deleteNodeFailSilent(getZooKeeper(), node);
3692    } catch (KeeperException ke) {
3693      throw new HBaseIOException(
3694        this.zooKeeper.prefix("Unable to recommission '" + server.getServerName() + "'."), ke);
3695    }
3696    this.serverManager.removeServerFromDrainList(server);
3697
3698    // Load the regions onto the server if we are given a list of regions.
3699    if (encodedRegionNames == null || encodedRegionNames.isEmpty()) {
3700      return;
3701    }
3702    if (!this.serverManager.isServerOnline(server)) {
3703      return;
3704    }
3705    for (byte[] encodedRegionName : encodedRegionNames) {
3706      RegionState regionState =
3707        assignmentManager.getRegionStates().getRegionState(Bytes.toString(encodedRegionName));
3708      if (regionState == null) {
3709        LOG.warn("Unknown region " + Bytes.toStringBinary(encodedRegionName));
3710        continue;
3711      }
3712      RegionInfo hri = regionState.getRegion();
3713      if (server.equals(regionState.getServerName())) {
3714        LOG.info("Skipping move of region " + hri.getRegionNameAsString() +
3715          " because region already assigned to the same server " + server + ".");
3716        continue;
3717      }
3718      RegionPlan rp = new RegionPlan(hri, regionState.getServerName(), server);
3719      this.assignmentManager.moveAsync(rp);
3720    }
3721  }
3722
3723  @Override
3724  public LockManager getLockManager() {
3725    return lockManager;
3726  }
3727
3728  public QuotaObserverChore getQuotaObserverChore() {
3729    return this.quotaObserverChore;
3730  }
3731
3732  public SpaceQuotaSnapshotNotifier getSpaceQuotaSnapshotNotifier() {
3733    return this.spaceQuotaSnapshotNotifier;
3734  }
3735
3736  @SuppressWarnings("unchecked")
3737  private RemoteProcedure<MasterProcedureEnv, ?> getRemoteProcedure(long procId) {
3738    Procedure<?> procedure = procedureExecutor.getProcedure(procId);
3739    if (procedure == null) {
3740      return null;
3741    }
3742    assert procedure instanceof RemoteProcedure;
3743    return (RemoteProcedure<MasterProcedureEnv, ?>) procedure;
3744  }
3745
3746  public void remoteProcedureCompleted(long procId) {
3747    LOG.debug("Remote procedure done, pid={}", procId);
3748    RemoteProcedure<MasterProcedureEnv, ?> procedure = getRemoteProcedure(procId);
3749    if (procedure != null) {
3750      procedure.remoteOperationCompleted(procedureExecutor.getEnvironment());
3751    }
3752  }
3753
3754  public void remoteProcedureFailed(long procId, RemoteProcedureException error) {
3755    LOG.debug("Remote procedure failed, pid={}", procId, error);
3756    RemoteProcedure<MasterProcedureEnv, ?> procedure = getRemoteProcedure(procId);
3757    if (procedure != null) {
3758      procedure.remoteOperationFailed(procedureExecutor.getEnvironment(), error);
3759    }
3760  }
3761
3762  /**
3763   * Reopen regions provided in the argument
3764   *
3765   * @param tableName The current table name
3766   * @param regionNames The region names of the regions to reopen
3767   * @param nonceGroup Identifier for the source of the request, a client or process
3768   * @param nonce A unique identifier for this operation from the client or process identified by
3769   *   <code>nonceGroup</code> (the source must ensure each operation gets a unique id).
3770   * @return procedure Id
3771   * @throws IOException if reopening region fails while running procedure
3772   */
3773  long reopenRegions(final TableName tableName, final List<byte[]> regionNames,
3774      final long nonceGroup, final long nonce)
3775      throws IOException {
3776
3777    return MasterProcedureUtil
3778      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
3779
3780        @Override
3781        protected void run() throws IOException {
3782          submitProcedure(new ReopenTableRegionsProcedure(tableName, regionNames));
3783        }
3784
3785        @Override
3786        protected String getDescription() {
3787          return "ReopenTableRegionsProcedure";
3788        }
3789
3790      });
3791
3792  }
3793
3794  @Override
3795  public ReplicationPeerManager getReplicationPeerManager() {
3796    return replicationPeerManager;
3797  }
3798
3799  public HashMap<String, List<Pair<ServerName, ReplicationLoadSource>>>
3800      getReplicationLoad(ServerName[] serverNames) {
3801    List<ReplicationPeerDescription> peerList = this.getReplicationPeerManager().listPeers(null);
3802    if (peerList == null) {
3803      return null;
3804    }
3805    HashMap<String, List<Pair<ServerName, ReplicationLoadSource>>> replicationLoadSourceMap =
3806        new HashMap<>(peerList.size());
3807    peerList.stream()
3808        .forEach(peer -> replicationLoadSourceMap.put(peer.getPeerId(), new ArrayList<>()));
3809    for (ServerName serverName : serverNames) {
3810      List<ReplicationLoadSource> replicationLoadSources =
3811          getServerManager().getLoad(serverName).getReplicationLoadSourceList();
3812      for (ReplicationLoadSource replicationLoadSource : replicationLoadSources) {
3813        replicationLoadSourceMap.get(replicationLoadSource.getPeerID())
3814            .add(new Pair<>(serverName, replicationLoadSource));
3815      }
3816    }
3817    for (List<Pair<ServerName, ReplicationLoadSource>> loads : replicationLoadSourceMap.values()) {
3818      if (loads.size() > 0) {
3819        loads.sort(Comparator.comparingLong(load -> (-1) * load.getSecond().getReplicationLag()));
3820      }
3821    }
3822    return replicationLoadSourceMap;
3823  }
3824
3825  /**
3826   * This method modifies the master's configuration in order to inject replication-related features
3827   */
3828  @VisibleForTesting
3829  public static void decorateMasterConfiguration(Configuration conf) {
3830    String plugins = conf.get(HBASE_MASTER_LOGCLEANER_PLUGINS);
3831    String cleanerClass = ReplicationLogCleaner.class.getCanonicalName();
3832    if (!plugins.contains(cleanerClass)) {
3833      conf.set(HBASE_MASTER_LOGCLEANER_PLUGINS, plugins + "," + cleanerClass);
3834    }
3835    if (ReplicationUtils.isReplicationForBulkLoadDataEnabled(conf)) {
3836      plugins = conf.get(HFileCleaner.MASTER_HFILE_CLEANER_PLUGINS);
3837      cleanerClass = ReplicationHFileCleaner.class.getCanonicalName();
3838      if (!plugins.contains(cleanerClass)) {
3839        conf.set(HFileCleaner.MASTER_HFILE_CLEANER_PLUGINS, plugins + "," + cleanerClass);
3840      }
3841    }
3842  }
3843
3844  public SnapshotQuotaObserverChore getSnapshotQuotaObserverChore() {
3845    return this.snapshotQuotaChore;
3846  }
3847
3848  @Override
3849  public SyncReplicationReplayWALManager getSyncReplicationReplayWALManager() {
3850    return this.syncReplicationReplayWALManager;
3851  }
3852
3853  @Override
3854  public Map<String, ReplicationStatus> getWalGroupsReplicationStatus() {
3855    if (!this.isOnline() || !LoadBalancer.isMasterCanHostUserRegions(conf)) {
3856      return new HashMap<>();
3857    }
3858    return super.getWalGroupsReplicationStatus();
3859  }
3860
3861  public HbckChore getHbckChore() {
3862    return this.hbckChore;
3863  }
3864
3865  @Override
3866  public String getClusterId() {
3867    if (activeMaster) {
3868      return super.getClusterId();
3869    }
3870    return cachedClusterId.getFromCacheOrFetch();
3871  }
3872
3873  @Override
3874  public void runReplicationBarrierCleaner() {
3875    ReplicationBarrierCleaner rbc = this.replicationBarrierCleaner;
3876    if (rbc != null) {
3877      rbc.chore();
3878    }
3879  }
3880}