/*
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.hadoop.hbase.master;

import static org.apache.hadoop.hbase.HConstants.DEFAULT_HBASE_SPLIT_COORDINATED_BY_ZK;
import static org.apache.hadoop.hbase.HConstants.HBASE_MASTER_LOGCLEANER_PLUGINS;
import static org.apache.hadoop.hbase.HConstants.HBASE_SPLIT_WAL_COORDINATED_BY_ZK;
import static org.apache.hadoop.hbase.master.cleaner.HFileCleaner.CUSTOM_POOL_SIZE;
import static org.apache.hadoop.hbase.util.DNS.MASTER_HOSTNAME_KEY;

import com.google.errorprone.annotations.RestrictedApi;
import io.opentelemetry.api.trace.Span;
import io.opentelemetry.api.trace.StatusCode;
import io.opentelemetry.context.Scope;
import java.io.IOException;
import java.io.InterruptedIOException;
import java.lang.reflect.Constructor;
import java.lang.reflect.InvocationTargetException;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.UnknownHostException;
import java.time.Instant;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.regex.Pattern;
import java.util.stream.Collectors;
import javax.servlet.http.HttpServlet;
import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.CatalogFamilyFormat;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellBuilderFactory;
import org.apache.hadoop.hbase.CellBuilderType;
import org.apache.hadoop.hbase.ClusterId;
import org.apache.hadoop.hbase.ClusterMetrics;
import org.apache.hadoop.hbase.ClusterMetrics.Option;
import org.apache.hadoop.hbase.ClusterMetricsBuilder;
import org.apache.hadoop.hbase.DoNotRetryIOException;
import org.apache.hadoop.hbase.HBaseIOException;
import org.apache.hadoop.hbase.HBaseInterfaceAudience;
import org.apache.hadoop.hbase.HBaseServerBase;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionLocation;
import org.apache.hadoop.hbase.InvalidFamilyOperationException;
import org.apache.hadoop.hbase.MasterNotRunningException;
import org.apache.hadoop.hbase.MetaTableAccessor;
import org.apache.hadoop.hbase.NamespaceDescriptor;
import org.apache.hadoop.hbase.PleaseHoldException;
import org.apache.hadoop.hbase.PleaseRestartMasterException;
import org.apache.hadoop.hbase.RegionMetrics;
import org.apache.hadoop.hbase.ReplicationPeerNotFoundException;
import org.apache.hadoop.hbase.ScheduledChore;
import org.apache.hadoop.hbase.ServerMetrics;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.ServerTask;
import org.apache.hadoop.hbase.ServerTaskBuilder;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.TableNotDisabledException;
import org.apache.hadoop.hbase.TableNotFoundException;
import org.apache.hadoop.hbase.UnknownRegionException;
import org.apache.hadoop.hbase.client.BalanceRequest;
import org.apache.hadoop.hbase.client.BalanceResponse;
import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
import org.apache.hadoop.hbase.client.CompactionState;
import org.apache.hadoop.hbase.client.MasterSwitchType;
import org.apache.hadoop.hbase.client.NormalizeTableFilterParams;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.RegionInfo;
import org.apache.hadoop.hbase.client.RegionInfoBuilder;
import org.apache.hadoop.hbase.client.RegionStatesCount;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.TableDescriptor;
import org.apache.hadoop.hbase.client.TableDescriptorBuilder;
import org.apache.hadoop.hbase.client.TableState;
import org.apache.hadoop.hbase.coprocessor.CoprocessorHost;
import org.apache.hadoop.hbase.exceptions.DeserializationException;
import org.apache.hadoop.hbase.exceptions.MasterStoppedException;
import org.apache.hadoop.hbase.executor.ExecutorType;
import org.apache.hadoop.hbase.favored.FavoredNodesManager;
import org.apache.hadoop.hbase.http.HttpServer;
import org.apache.hadoop.hbase.http.InfoServer;
import org.apache.hadoop.hbase.ipc.CoprocessorRpcUtils;
import org.apache.hadoop.hbase.ipc.RpcServer;
import org.apache.hadoop.hbase.ipc.ServerNotRunningYetException;
import org.apache.hadoop.hbase.log.HBaseMarkers;
import org.apache.hadoop.hbase.master.MasterRpcServices.BalanceSwitchMode;
import org.apache.hadoop.hbase.master.assignment.AssignmentManager;
import org.apache.hadoop.hbase.master.assignment.MergeTableRegionsProcedure;
import org.apache.hadoop.hbase.master.assignment.RegionStateNode;
import org.apache.hadoop.hbase.master.assignment.RegionStateStore;
import org.apache.hadoop.hbase.master.assignment.RegionStates;
import org.apache.hadoop.hbase.master.assignment.TransitRegionStateProcedure;
import org.apache.hadoop.hbase.master.balancer.BalancerChore;
import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer;
import org.apache.hadoop.hbase.master.balancer.ClusterStatusChore;
import org.apache.hadoop.hbase.master.balancer.LoadBalancerFactory;
import org.apache.hadoop.hbase.master.balancer.LoadBalancerStateStore;
import org.apache.hadoop.hbase.master.balancer.MaintenanceLoadBalancer;
import org.apache.hadoop.hbase.master.cleaner.DirScanPool;
import org.apache.hadoop.hbase.master.cleaner.HFileCleaner;
import org.apache.hadoop.hbase.master.cleaner.LogCleaner;
import org.apache.hadoop.hbase.master.cleaner.ReplicationBarrierCleaner;
import org.apache.hadoop.hbase.master.cleaner.SnapshotCleanerChore;
import org.apache.hadoop.hbase.master.hbck.HbckChore;
import org.apache.hadoop.hbase.master.http.MasterDumpServlet;
import org.apache.hadoop.hbase.master.http.MasterRedirectServlet;
import org.apache.hadoop.hbase.master.http.MasterStatusServlet;
import org.apache.hadoop.hbase.master.http.api_v1.ResourceConfigFactory;
import org.apache.hadoop.hbase.master.http.hbck.HbckConfigFactory;
import org.apache.hadoop.hbase.master.janitor.CatalogJanitor;
import org.apache.hadoop.hbase.master.locking.LockManager;
import org.apache.hadoop.hbase.master.migrate.RollingUpgradeChore;
import org.apache.hadoop.hbase.master.normalizer.RegionNormalizerFactory;
import org.apache.hadoop.hbase.master.normalizer.RegionNormalizerManager;
import org.apache.hadoop.hbase.master.normalizer.RegionNormalizerStateStore;
import org.apache.hadoop.hbase.master.procedure.CreateTableProcedure;
import org.apache.hadoop.hbase.master.procedure.DeleteNamespaceProcedure;
import org.apache.hadoop.hbase.master.procedure.DeleteTableProcedure;
import org.apache.hadoop.hbase.master.procedure.DisableTableProcedure;
import org.apache.hadoop.hbase.master.procedure.EnableTableProcedure;
import org.apache.hadoop.hbase.master.procedure.FlushTableProcedure;
import org.apache.hadoop.hbase.master.procedure.InitMetaProcedure;
import org.apache.hadoop.hbase.master.procedure.MasterProcedureConstants;
import org.apache.hadoop.hbase.master.procedure.MasterProcedureEnv;
import org.apache.hadoop.hbase.master.procedure.MasterProcedureScheduler;
import org.apache.hadoop.hbase.master.procedure.MasterProcedureUtil;
import org.apache.hadoop.hbase.master.procedure.MasterProcedureUtil.NonceProcedureRunnable;
import org.apache.hadoop.hbase.master.procedure.ModifyTableProcedure;
import org.apache.hadoop.hbase.master.procedure.ProcedurePrepareLatch;
import org.apache.hadoop.hbase.master.procedure.ProcedureSyncWait;
import org.apache.hadoop.hbase.master.procedure.ReopenTableRegionsProcedure;
import org.apache.hadoop.hbase.master.procedure.ServerCrashProcedure;
import org.apache.hadoop.hbase.master.procedure.TruncateRegionProcedure;
import org.apache.hadoop.hbase.master.procedure.TruncateTableProcedure;
import org.apache.hadoop.hbase.master.region.MasterRegion;
import org.apache.hadoop.hbase.master.region.MasterRegionFactory;
import org.apache.hadoop.hbase.master.replication.AbstractPeerProcedure;
import org.apache.hadoop.hbase.master.replication.AddPeerProcedure;
import org.apache.hadoop.hbase.master.replication.DisablePeerProcedure;
import org.apache.hadoop.hbase.master.replication.EnablePeerProcedure;
import org.apache.hadoop.hbase.master.replication.MigrateReplicationQueueFromZkToTableProcedure;
import org.apache.hadoop.hbase.master.replication.RemovePeerProcedure;
import org.apache.hadoop.hbase.master.replication.ReplicationPeerManager;
import org.apache.hadoop.hbase.master.replication.ReplicationPeerModificationStateStore;
import org.apache.hadoop.hbase.master.replication.SyncReplicationReplayWALManager;
import org.apache.hadoop.hbase.master.replication.TransitPeerSyncReplicationStateProcedure;
import org.apache.hadoop.hbase.master.replication.UpdatePeerConfigProcedure;
import org.apache.hadoop.hbase.master.slowlog.SlowLogMasterService;
import org.apache.hadoop.hbase.master.snapshot.SnapshotCleanupStateStore;
import org.apache.hadoop.hbase.master.snapshot.SnapshotManager;
import org.apache.hadoop.hbase.master.waleventtracker.WALEventTrackerTableCreator;
import org.apache.hadoop.hbase.master.zksyncer.MasterAddressSyncer;
import org.apache.hadoop.hbase.master.zksyncer.MetaLocationSyncer;
import org.apache.hadoop.hbase.mob.MobFileCleanerChore;
import org.apache.hadoop.hbase.mob.MobFileCompactionChore;
import org.apache.hadoop.hbase.monitoring.MemoryBoundedLogMessageBuffer;
import org.apache.hadoop.hbase.monitoring.MonitoredTask;
import org.apache.hadoop.hbase.monitoring.TaskGroup;
import org.apache.hadoop.hbase.monitoring.TaskMonitor;
import org.apache.hadoop.hbase.namequeues.NamedQueueRecorder;
import org.apache.hadoop.hbase.procedure.MasterProcedureManagerHost;
import org.apache.hadoop.hbase.procedure.flush.MasterFlushTableProcedureManager;
import org.apache.hadoop.hbase.procedure2.LockedResource;
import org.apache.hadoop.hbase.procedure2.Procedure;
import org.apache.hadoop.hbase.procedure2.ProcedureEvent;
import org.apache.hadoop.hbase.procedure2.ProcedureExecutor;
import org.apache.hadoop.hbase.procedure2.RemoteProcedureDispatcher.RemoteProcedure;
import org.apache.hadoop.hbase.procedure2.RemoteProcedureException;
import org.apache.hadoop.hbase.procedure2.store.ProcedureStore;
import org.apache.hadoop.hbase.procedure2.store.ProcedureStore.ProcedureStoreListener;
import org.apache.hadoop.hbase.procedure2.store.region.RegionProcedureStore;
import org.apache.hadoop.hbase.quotas.MasterQuotaManager;
import org.apache.hadoop.hbase.quotas.MasterQuotasObserver;
import org.apache.hadoop.hbase.quotas.QuotaObserverChore;
import org.apache.hadoop.hbase.quotas.QuotaTableUtil;
import org.apache.hadoop.hbase.quotas.QuotaUtil;
import org.apache.hadoop.hbase.quotas.SnapshotQuotaObserverChore;
import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshot;
import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshot.SpaceQuotaStatus;
import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshotNotifier;
import org.apache.hadoop.hbase.quotas.SpaceQuotaSnapshotNotifierFactory;
import org.apache.hadoop.hbase.quotas.SpaceViolationPolicy;
import org.apache.hadoop.hbase.regionserver.HRegionServer;
import org.apache.hadoop.hbase.regionserver.NoSuchColumnFamilyException;
import org.apache.hadoop.hbase.regionserver.storefiletracker.ModifyColumnFamilyStoreFileTrackerProcedure;
import org.apache.hadoop.hbase.regionserver.storefiletracker.ModifyTableStoreFileTrackerProcedure;
import org.apache.hadoop.hbase.replication.ReplicationException;
import org.apache.hadoop.hbase.replication.ReplicationLoadSource;
import org.apache.hadoop.hbase.replication.ReplicationPeerConfig;
import org.apache.hadoop.hbase.replication.ReplicationPeerDescription;
import org.apache.hadoop.hbase.replication.ReplicationUtils;
import org.apache.hadoop.hbase.replication.SyncReplicationState;
import org.apache.hadoop.hbase.replication.ZKReplicationQueueStorageForMigration;
import org.apache.hadoop.hbase.replication.master.ReplicationHFileCleaner;
import org.apache.hadoop.hbase.replication.master.ReplicationLogCleaner;
import org.apache.hadoop.hbase.replication.master.ReplicationLogCleanerBarrier;
import org.apache.hadoop.hbase.replication.master.ReplicationSinkTrackerTableCreator;
import org.apache.hadoop.hbase.replication.regionserver.ReplicationSyncUp;
import org.apache.hadoop.hbase.replication.regionserver.ReplicationSyncUp.ReplicationSyncUpToolInfo;
import org.apache.hadoop.hbase.rsgroup.RSGroupAdminEndpoint;
import org.apache.hadoop.hbase.rsgroup.RSGroupBasedLoadBalancer;
import org.apache.hadoop.hbase.rsgroup.RSGroupInfoManager;
import org.apache.hadoop.hbase.rsgroup.RSGroupUtil;
import org.apache.hadoop.hbase.security.AccessDeniedException;
import org.apache.hadoop.hbase.security.SecurityConstants;
import org.apache.hadoop.hbase.security.Superusers;
import org.apache.hadoop.hbase.security.UserProvider;
import org.apache.hadoop.hbase.trace.TraceUtil;
import org.apache.hadoop.hbase.util.Addressing;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.CommonFSUtils;
import org.apache.hadoop.hbase.util.CoprocessorConfigurationUtil;
import org.apache.hadoop.hbase.util.EnvironmentEdgeManager;
import org.apache.hadoop.hbase.util.FSTableDescriptors;
import org.apache.hadoop.hbase.util.FutureUtils;
import org.apache.hadoop.hbase.util.HBaseFsck;
import org.apache.hadoop.hbase.util.HFileArchiveUtil;
import org.apache.hadoop.hbase.util.IdLock;
import org.apache.hadoop.hbase.util.JVMClusterUtil;
import org.apache.hadoop.hbase.util.JsonMapper;
import org.apache.hadoop.hbase.util.ModifyRegionUtils;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.RetryCounter;
import org.apache.hadoop.hbase.util.RetryCounterFactory;
import org.apache.hadoop.hbase.util.TableDescriptorChecker;
import org.apache.hadoop.hbase.util.Threads;
import org.apache.hadoop.hbase.util.VersionInfo;
import org.apache.hadoop.hbase.zookeeper.MasterAddressTracker;
import org.apache.hadoop.hbase.zookeeper.MetaTableLocator;
import org.apache.hadoop.hbase.zookeeper.ZKClusterId;
import org.apache.hadoop.hbase.zookeeper.ZKUtil;
import org.apache.hadoop.hbase.zookeeper.ZKWatcher;
import org.apache.hadoop.hbase.zookeeper.ZNodePaths;
import org.apache.yetus.audience.InterfaceAudience;
import org.apache.zookeeper.KeeperException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import org.apache.hbase.thirdparty.com.google.common.collect.Lists;
import org.apache.hbase.thirdparty.com.google.common.collect.Maps;
import org.apache.hbase.thirdparty.com.google.common.collect.Sets;
import org.apache.hbase.thirdparty.com.google.common.io.ByteStreams;
import org.apache.hbase.thirdparty.com.google.common.io.Closeables;
import org.apache.hbase.thirdparty.com.google.gson.JsonParseException;
import org.apache.hbase.thirdparty.com.google.protobuf.Descriptors;
import org.apache.hbase.thirdparty.com.google.protobuf.Service;
import org.apache.hbase.thirdparty.org.eclipse.jetty.server.Server;
import org.apache.hbase.thirdparty.org.eclipse.jetty.server.ServerConnector;
import org.apache.hbase.thirdparty.org.eclipse.jetty.servlet.ServletHolder;
import org.apache.hbase.thirdparty.org.eclipse.jetty.webapp.WebAppContext;
import org.apache.hbase.thirdparty.org.glassfish.jersey.server.ResourceConfig;
import org.apache.hbase.thirdparty.org.glassfish.jersey.servlet.ServletContainer;

import org.apache.hadoop.hbase.shaded.protobuf.RequestConverter;
import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.GetRegionInfoResponse;
import org.apache.hadoop.hbase.shaded.protobuf.generated.SnapshotProtos.SnapshotDescription;

/**
 * HMaster is the "master server" for HBase. An HBase cluster has one active master. If many masters
 * are started, all compete. Whichever wins goes on to run the cluster. All others park themselves
 * in their constructor until master or cluster shutdown or until the active master loses its lease
 * in zookeeper. Thereafter, all running master jostle to take over master role.
 * <p/>
 * The Master can be asked shutdown the cluster. See {@link #shutdown()}. In this case it will tell
 * all regionservers to go down and then wait on them all reporting in that they are down. This
 * master will then shut itself down.
 * <p/>
 * You can also shutdown just this master. Call {@link #stopMaster()}.
 * @see org.apache.zookeeper.Watcher
 */
@InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.TOOLS)
public class HMaster extends HBaseServerBase<MasterRpcServices> implements MasterServices {

  private static final Logger LOG = LoggerFactory.getLogger(HMaster.class);

  // MASTER is name of the webapp and the attribute name used stuffing this
  // instance into a web context !! AND OTHER PLACES !!
  public static final String MASTER = "master";

  // Manager and zk listener for master election
  private final ActiveMasterManager activeMasterManager;
  // Region server tracker
  private final RegionServerTracker regionServerTracker;
  // Draining region server tracker
  private DrainingServerTracker drainingServerTracker;
  // Tracker for load balancer state
  LoadBalancerStateStore loadBalancerStateStore;
  // Tracker for meta location, if any client ZK quorum specified
  private MetaLocationSyncer metaLocationSyncer;
  // Tracker for active master location, if any client ZK quorum specified
  @InterfaceAudience.Private
  MasterAddressSyncer masterAddressSyncer;
  // Tracker for auto snapshot cleanup state
  SnapshotCleanupStateStore snapshotCleanupStateStore;

  // Tracker for split and merge state
  private SplitOrMergeStateStore splitOrMergeStateStore;

  private ClusterSchemaService clusterSchemaService;

  public static final String HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS =
    "hbase.master.wait.on.service.seconds";
  public static final int DEFAULT_HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS = 5 * 60;

  public static final String HBASE_MASTER_CLEANER_INTERVAL = "hbase.master.cleaner.interval";

  public static final int DEFAULT_HBASE_MASTER_CLEANER_INTERVAL = 600 * 1000;

  private String clusterId;

  // Metrics for the HMaster
  final MetricsMaster metricsMaster;
  // file system manager for the master FS operations
  private MasterFileSystem fileSystemManager;
  private MasterWalManager walManager;

  // manager to manage procedure-based WAL splitting, can be null if current
  // is zk-based WAL splitting. SplitWALManager will replace SplitLogManager
  // and MasterWalManager, which means zk-based WAL splitting code will be
  // useless after we switch to the procedure-based one. our eventual goal
  // is to remove all the zk-based WAL splitting code.
  private SplitWALManager splitWALManager;

  // server manager to deal with region server info
  private volatile ServerManager serverManager;

  // manager of assignment nodes in zookeeper
  private AssignmentManager assignmentManager;

  private RSGroupInfoManager rsGroupInfoManager;

  private final ReplicationLogCleanerBarrier replicationLogCleanerBarrier =
    new ReplicationLogCleanerBarrier();

  // Only allow to add one sync replication peer concurrently
  private final Semaphore syncReplicationPeerLock = new Semaphore(1);

  // manager of replication
  private ReplicationPeerManager replicationPeerManager;

  private SyncReplicationReplayWALManager syncReplicationReplayWALManager;

  // buffer for "fatal error" notices from region servers
  // in the cluster. This is only used for assisting
  // operations/debugging.
  MemoryBoundedLogMessageBuffer rsFatals;

  // flag set after we become the active master (used for testing)
  private volatile boolean activeMaster = false;

  // flag set after we complete initialization once active
  private final ProcedureEvent<?> initialized = new ProcedureEvent<>("master initialized");

  // flag set after master services are started,
  // initialization may have not completed yet.
  volatile boolean serviceStarted = false;

  // Maximum time we should run balancer for
  private final int maxBalancingTime;
  // Maximum percent of regions in transition when balancing
  private final double maxRitPercent;

  private final LockManager lockManager = new LockManager(this);

  private RSGroupBasedLoadBalancer balancer;
  private BalancerChore balancerChore;
  private static boolean disableBalancerChoreForTest = false;
  private RegionNormalizerManager regionNormalizerManager;
  private ClusterStatusChore clusterStatusChore;
  private ClusterStatusPublisher clusterStatusPublisherChore = null;
  private SnapshotCleanerChore snapshotCleanerChore = null;

  private HbckChore hbckChore;
  CatalogJanitor catalogJanitorChore;
  // Threadpool for scanning the Old logs directory, used by the LogCleaner
  private DirScanPool logCleanerPool;
  private LogCleaner logCleaner;
  // HFile cleaners for the custom hfile archive paths and the default archive path
  // The archive path cleaner is the first element
  private List<HFileCleaner> hfileCleaners = new ArrayList<>();
  // The hfile cleaner paths, including custom paths and the default archive path
  private List<Path> hfileCleanerPaths = new ArrayList<>();
  // The shared hfile cleaner pool for the custom archive paths
  private DirScanPool sharedHFileCleanerPool;
  // The exclusive hfile cleaner pool for scanning the archive directory
  private DirScanPool exclusiveHFileCleanerPool;
  private ReplicationBarrierCleaner replicationBarrierCleaner;
  private MobFileCleanerChore mobFileCleanerChore;
  private MobFileCompactionChore mobFileCompactionChore;
  private RollingUpgradeChore rollingUpgradeChore;
  // used to synchronize the mobCompactionStates
  private final IdLock mobCompactionLock = new IdLock();
  // save the information of mob compactions in tables.
  // the key is table name, the value is the number of compactions in that table.
  private Map<TableName, AtomicInteger> mobCompactionStates = Maps.newConcurrentMap();

  volatile MasterCoprocessorHost cpHost;

  private final boolean preLoadTableDescriptors;

  // Time stamps for when a hmaster became active
  private long masterActiveTime;

  // Time stamp for when HMaster finishes becoming Active Master
  private long masterFinishedInitializationTime;

  Map<String, Service> coprocessorServiceHandlers = Maps.newHashMap();

  // monitor for snapshot of hbase tables
  SnapshotManager snapshotManager;
  // monitor for distributed procedures
  private MasterProcedureManagerHost mpmHost;

  private RegionsRecoveryChore regionsRecoveryChore = null;

  private RegionsRecoveryConfigManager regionsRecoveryConfigManager = null;
  // it is assigned after 'initialized' guard set to true, so should be volatile
  private volatile MasterQuotaManager quotaManager;
  private SpaceQuotaSnapshotNotifier spaceQuotaSnapshotNotifier;
  private QuotaObserverChore quotaObserverChore;
  private SnapshotQuotaObserverChore snapshotQuotaChore;
  private OldWALsDirSizeChore oldWALsDirSizeChore;

  private ProcedureExecutor<MasterProcedureEnv> procedureExecutor;
  private ProcedureStore procedureStore;

  // the master local storage to store procedure data, meta region locations, etc.
  private MasterRegion masterRegion;

  private RegionServerList rsListStorage;

  // handle table states
  private TableStateManager tableStateManager;

  /** jetty server for master to redirect requests to regionserver infoServer */
  private Server masterJettyServer;

  // Determine if we should do normal startup or minimal "single-user" mode with no region
  // servers and no user tables. Useful for repair and recovery of hbase:meta
  private final boolean maintenanceMode;
  static final String MAINTENANCE_MODE = "hbase.master.maintenance_mode";

  // the in process region server for carry system regions in maintenanceMode
  private JVMClusterUtil.RegionServerThread maintenanceRegionServer;

  // Cached clusterId on stand by masters to serve clusterID requests from clients.
  private final CachedClusterId cachedClusterId;

  public static final String WARMUP_BEFORE_MOVE = "hbase.master.warmup.before.move";
  private static final boolean DEFAULT_WARMUP_BEFORE_MOVE = true;

  private TaskGroup startupTaskGroup;

  /**
   * Store whether we allow replication peer modification operations.
   */
  private ReplicationPeerModificationStateStore replicationPeerModificationStateStore;

  /**
   * Initializes the HMaster. The steps are as follows:
   * <p>
   * <ol>
   * <li>Initialize the local HRegionServer
   * <li>Start the ActiveMasterManager.
   * </ol>
   * <p>
   * Remaining steps of initialization occur in {@link #finishActiveMasterInitialization()} after
   * the master becomes the active one.
   */
  public HMaster(final Configuration conf) throws IOException {
    super(conf, "Master");
    final Span span = TraceUtil.createSpan("HMaster.cxtor");
    try (Scope ignored = span.makeCurrent()) {
      if (conf.getBoolean(MAINTENANCE_MODE, false)) {
        LOG.info("Detected {}=true via configuration.", MAINTENANCE_MODE);
        maintenanceMode = true;
      } else if (Boolean.getBoolean(MAINTENANCE_MODE)) {
        LOG.info("Detected {}=true via environment variables.", MAINTENANCE_MODE);
        maintenanceMode = true;
      } else {
        maintenanceMode = false;
      }
      this.rsFatals = new MemoryBoundedLogMessageBuffer(
        conf.getLong("hbase.master.buffer.for.rs.fatals", 1 * 1024 * 1024));
      LOG.info("hbase.rootdir={}, hbase.cluster.distributed={}",
        CommonFSUtils.getRootDir(this.conf),
        this.conf.getBoolean(HConstants.CLUSTER_DISTRIBUTED, false));

      // Disable usage of meta replicas in the master
      this.conf.setBoolean(HConstants.USE_META_REPLICAS, false);

      decorateMasterConfiguration(this.conf);

      // Hack! Maps DFSClient => Master for logs. HDFS made this
      // config param for task trackers, but we can piggyback off of it.
      if (this.conf.get("mapreduce.task.attempt.id") == null) {
        this.conf.set("mapreduce.task.attempt.id", "hb_m_" + this.serverName.toString());
      }

      this.metricsMaster = new MetricsMaster(new MetricsMasterWrapperImpl(this));

      // preload table descriptor at startup
      this.preLoadTableDescriptors = conf.getBoolean("hbase.master.preload.tabledescriptors", true);

      this.maxBalancingTime = getMaxBalancingTime();
      this.maxRitPercent = conf.getDouble(HConstants.HBASE_MASTER_BALANCER_MAX_RIT_PERCENT,
        HConstants.DEFAULT_HBASE_MASTER_BALANCER_MAX_RIT_PERCENT);

      // Do we publish the status?
      boolean shouldPublish =
        conf.getBoolean(HConstants.STATUS_PUBLISHED, HConstants.STATUS_PUBLISHED_DEFAULT);
      Class<? extends ClusterStatusPublisher.Publisher> publisherClass =
        conf.getClass(ClusterStatusPublisher.STATUS_PUBLISHER_CLASS,
          ClusterStatusPublisher.DEFAULT_STATUS_PUBLISHER_CLASS,
          ClusterStatusPublisher.Publisher.class);

      if (shouldPublish) {
        if (publisherClass == null) {
          LOG.warn(HConstants.STATUS_PUBLISHED + " is true, but "
            + ClusterStatusPublisher.DEFAULT_STATUS_PUBLISHER_CLASS
            + " is not set - not publishing status");
        } else {
          clusterStatusPublisherChore = new ClusterStatusPublisher(this, conf, publisherClass);
          LOG.debug("Created {}", this.clusterStatusPublisherChore);
          getChoreService().scheduleChore(clusterStatusPublisherChore);
        }
      }
      this.activeMasterManager = createActiveMasterManager(zooKeeper, serverName, this);
      cachedClusterId = new CachedClusterId(this, conf);
      this.regionServerTracker = new RegionServerTracker(zooKeeper, this);
      this.rpcServices.start(zooKeeper);
      span.setStatus(StatusCode.OK);
    } catch (Throwable t) {
      // Make sure we log the exception. HMaster is often started via reflection and the
      // cause of failed startup is lost.
      TraceUtil.setError(span, t);
      LOG.error("Failed construction of Master", t);
      throw t;
    } finally {
      span.end();
    }
  }

  /**
   * Protected to have custom implementations in tests override the default ActiveMaster
   * implementation.
   */
  protected ActiveMasterManager createActiveMasterManager(ZKWatcher zk, ServerName sn,
    org.apache.hadoop.hbase.Server server) throws InterruptedIOException {
    return new ActiveMasterManager(zk, sn, server);
  }

  @Override
  protected String getUseThisHostnameInstead(Configuration conf) {
    return conf.get(MASTER_HOSTNAME_KEY);
  }

  private void registerConfigurationObservers() {
    configurationManager.registerObserver(this.rpcServices);
    configurationManager.registerObserver(this);
  }

  // Main run loop. Calls through to the regionserver run loop AFTER becoming active Master; will
  // block in here until then.
  @Override
  public void run() {
    try {
      installShutdownHook();
      registerConfigurationObservers();
      Threads.setDaemonThreadRunning(new Thread(TraceUtil.tracedRunnable(() -> {
        try {
          int infoPort = putUpJettyServer();
          startActiveMasterManager(infoPort);
        } catch (Throwable t) {
          // Make sure we log the exception.
          String error = "Failed to become Active Master";
          LOG.error(error, t);
          // Abort should have been called already.
          if (!isAborted()) {
            abort(error, t);
          }
        }
      }, "HMaster.becomeActiveMaster")), getName() + ":becomeActiveMaster");
      while (!isStopped() && !isAborted()) {
        sleeper.sleep();
      }
      final Span span = TraceUtil.createSpan("HMaster exiting main loop");
      try (Scope ignored = span.makeCurrent()) {
        stopInfoServer();
        closeClusterConnection();
        stopServiceThreads();
        if (this.rpcServices != null) {
          this.rpcServices.stop();
        }
        closeZooKeeper();
        closeTableDescriptors();
        span.setStatus(StatusCode.OK);
      } finally {
        span.end();
      }
    } finally {
      if (this.clusterSchemaService != null) {
        // If on way out, then we are no longer active master.
        this.clusterSchemaService.stopAsync();
        try {
          this.clusterSchemaService
            .awaitTerminated(getConfiguration().getInt(HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS,
              DEFAULT_HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS), TimeUnit.SECONDS);
        } catch (TimeoutException te) {
          LOG.warn("Failed shutdown of clusterSchemaService", te);
        }
      }
      this.activeMaster = false;
    }
  }

  // return the actual infoPort, -1 means disable info server.
  private int putUpJettyServer() throws IOException {
    if (!conf.getBoolean("hbase.master.infoserver.redirect", true)) {
      return -1;
    }
    final int infoPort =
      conf.getInt("hbase.master.info.port.orig", HConstants.DEFAULT_MASTER_INFOPORT);
    // -1 is for disabling info server, so no redirecting
    if (infoPort < 0 || infoServer == null) {
      return -1;
    }
    if (infoPort == infoServer.getPort()) {
      // server is already running
      return infoPort;
    }
    final String addr = conf.get("hbase.master.info.bindAddress", "0.0.0.0");
    if (!Addressing.isLocalAddress(InetAddress.getByName(addr))) {
      String msg = "Failed to start redirecting jetty server. Address " + addr
        + " does not belong to this host. Correct configuration parameter: "
        + "hbase.master.info.bindAddress";
      LOG.error(msg);
      throw new IOException(msg);
    }

    // TODO I'm pretty sure we could just add another binding to the InfoServer run by
    // the RegionServer and have it run the RedirectServlet instead of standing up
    // a second entire stack here.
    masterJettyServer = new Server();
    final ServerConnector connector = new ServerConnector(masterJettyServer);
    connector.setHost(addr);
    connector.setPort(infoPort);
    masterJettyServer.addConnector(connector);
    masterJettyServer.setStopAtShutdown(true);
    masterJettyServer.setHandler(HttpServer.buildGzipHandler(masterJettyServer.getHandler()));

    final String redirectHostname =
      StringUtils.isBlank(useThisHostnameInstead) ? null : useThisHostnameInstead;

    final MasterRedirectServlet redirect = new MasterRedirectServlet(infoServer, redirectHostname);
    final WebAppContext context =
      new WebAppContext(null, "/", null, null, null, null, WebAppContext.NO_SESSIONS);
    context.addServlet(new ServletHolder(redirect), "/*");
    context.setServer(masterJettyServer);

    try {
      masterJettyServer.start();
    } catch (Exception e) {
      throw new IOException("Failed to start redirecting jetty server", e);
    }
    return connector.getLocalPort();
  }

  /**
   * For compatibility, if failed with regionserver credentials, try the master one
   */
  @Override
  protected void login(UserProvider user, String host) throws IOException {
    try {
      user.login(SecurityConstants.REGIONSERVER_KRB_KEYTAB_FILE,
        SecurityConstants.REGIONSERVER_KRB_PRINCIPAL, host);
    } catch (IOException ie) {
      user.login(SecurityConstants.MASTER_KRB_KEYTAB_FILE, SecurityConstants.MASTER_KRB_PRINCIPAL,
        host);
    }
  }

  public MasterRpcServices getMasterRpcServices() {
    return rpcServices;
  }

  @Override
  protected MasterCoprocessorHost getCoprocessorHost() {
    return getMasterCoprocessorHost();
  }

  public boolean balanceSwitch(final boolean b) throws IOException {
    return getMasterRpcServices().switchBalancer(b, BalanceSwitchMode.ASYNC);
  }

  @Override
  protected String getProcessName() {
    return MASTER;
  }

  @Override
  protected boolean canCreateBaseZNode() {
    return true;
  }

  @Override
  protected boolean canUpdateTableDescriptor() {
    return true;
  }

  @Override
  protected boolean cacheTableDescriptor() {
    return true;
  }

  protected MasterRpcServices createRpcServices() throws IOException {
    return new MasterRpcServices(this);
  }

  @Override
  protected void configureInfoServer(InfoServer infoServer) {
    infoServer.addUnprivilegedServlet("master-status", "/master-status", MasterStatusServlet.class);
    infoServer.addUnprivilegedServlet("api_v1", "/api/v1/*", buildApiV1Servlet());
    infoServer.addUnprivilegedServlet("hbck", "/hbck/*", buildHbckServlet());

    infoServer.setAttribute(MASTER, this);
  }

  private ServletHolder buildApiV1Servlet() {
    final ResourceConfig config = ResourceConfigFactory.createResourceConfig(conf, this);
    return new ServletHolder(new ServletContainer(config));
  }

  private ServletHolder buildHbckServlet() {
    final ResourceConfig config = HbckConfigFactory.createResourceConfig(conf, this);
    return new ServletHolder(new ServletContainer(config));
  }

  @Override
  protected Class<? extends HttpServlet> getDumpServlet() {
    return MasterDumpServlet.class;
  }

  @Override
  public MetricsMaster getMasterMetrics() {
    return metricsMaster;
  }

  /**
   * Initialize all ZK based system trackers. But do not include {@link RegionServerTracker}, it
   * should have already been initialized along with {@link ServerManager}.
   */
  private void initializeZKBasedSystemTrackers()
    throws IOException, KeeperException, ReplicationException, DeserializationException {
    if (maintenanceMode) {
      // in maintenance mode, always use MaintenanceLoadBalancer.
      conf.unset(LoadBalancer.HBASE_RSGROUP_LOADBALANCER_CLASS);
      conf.setClass(HConstants.HBASE_MASTER_LOADBALANCER_CLASS, MaintenanceLoadBalancer.class,
        LoadBalancer.class);
    }
    this.balancer = new RSGroupBasedLoadBalancer();
    this.loadBalancerStateStore = new LoadBalancerStateStore(masterRegion, zooKeeper);

    this.regionNormalizerManager =
      RegionNormalizerFactory.createNormalizerManager(conf, masterRegion, zooKeeper, this);
    this.configurationManager.registerObserver(regionNormalizerManager);
    this.regionNormalizerManager.start();

    this.splitOrMergeStateStore = new SplitOrMergeStateStore(masterRegion, zooKeeper, conf);

    // This is for backwards compatible. We do not need the CP for rs group now but if user want to
    // load it, we need to enable rs group.
    String[] cpClasses = conf.getStrings(MasterCoprocessorHost.MASTER_COPROCESSOR_CONF_KEY);
    if (cpClasses != null) {
      for (String cpClass : cpClasses) {
        if (RSGroupAdminEndpoint.class.getName().equals(cpClass)) {
          RSGroupUtil.enableRSGroup(conf);
          break;
        }
      }
    }
    this.rsGroupInfoManager = RSGroupInfoManager.create(this);

    this.replicationPeerManager = ReplicationPeerManager.create(this, clusterId);
    this.configurationManager.registerObserver(replicationPeerManager);
    this.replicationPeerModificationStateStore =
      new ReplicationPeerModificationStateStore(masterRegion);

    this.drainingServerTracker = new DrainingServerTracker(zooKeeper, this, this.serverManager);
    this.drainingServerTracker.start();

    this.snapshotCleanupStateStore = new SnapshotCleanupStateStore(masterRegion, zooKeeper);

    String clientQuorumServers = conf.get(HConstants.CLIENT_ZOOKEEPER_QUORUM);
    boolean clientZkObserverMode = conf.getBoolean(HConstants.CLIENT_ZOOKEEPER_OBSERVER_MODE,
      HConstants.DEFAULT_CLIENT_ZOOKEEPER_OBSERVER_MODE);
    if (clientQuorumServers != null && !clientZkObserverMode) {
      // we need to take care of the ZK information synchronization
      // if given client ZK are not observer nodes
      ZKWatcher clientZkWatcher = new ZKWatcher(conf,
        getProcessName() + ":" + rpcServices.getSocketAddress().getPort() + "-clientZK", this,
        false, true);
      this.metaLocationSyncer = new MetaLocationSyncer(zooKeeper, clientZkWatcher, this);
      this.metaLocationSyncer.start();
      this.masterAddressSyncer = new MasterAddressSyncer(zooKeeper, clientZkWatcher, this);
      this.masterAddressSyncer.start();
      // set cluster id is a one-go effort
      ZKClusterId.setClusterId(clientZkWatcher, fileSystemManager.getClusterId());
    }

    // Set the cluster as up. If new RSs, they'll be waiting on this before
    // going ahead with their startup.
    boolean wasUp = this.clusterStatusTracker.isClusterUp();
    if (!wasUp) this.clusterStatusTracker.setClusterUp();

    LOG.info("Active/primary master=" + this.serverName + ", sessionid=0x"
      + Long.toHexString(this.zooKeeper.getRecoverableZooKeeper().getSessionId())
      + ", setting cluster-up flag (Was=" + wasUp + ")");

    // create/initialize the snapshot manager and other procedure managers
    this.snapshotManager = new SnapshotManager();
    this.mpmHost = new MasterProcedureManagerHost();
    this.mpmHost.register(this.snapshotManager);
    this.mpmHost.register(new MasterFlushTableProcedureManager());
    this.mpmHost.loadProcedures(conf);
    this.mpmHost.initialize(this, this.metricsMaster);
  }

  // Will be overriden in test to inject customized AssignmentManager
  @InterfaceAudience.Private
  protected AssignmentManager createAssignmentManager(MasterServices master,
    MasterRegion masterRegion) {
    return new AssignmentManager(master, masterRegion);
  }

  private void tryMigrateMetaLocationsFromZooKeeper() throws IOException, KeeperException {
    // try migrate data from zookeeper
    try (ResultScanner scanner =
      masterRegion.getScanner(new Scan().addFamily(HConstants.CATALOG_FAMILY))) {
      if (scanner.next() != null) {
        // notice that all replicas for a region are in the same row, so the migration can be
        // done with in a one row put, which means if we have data in catalog family then we can
        // make sure that the migration is done.
        LOG.info("The {} family in master local region already has data in it, skip migrating...",
          HConstants.CATALOG_FAMILY_STR);
        return;
      }
    }
    // start migrating
    byte[] row = CatalogFamilyFormat.getMetaKeyForRegion(RegionInfoBuilder.FIRST_META_REGIONINFO);
    Put put = new Put(row);
    List<String> metaReplicaNodes = zooKeeper.getMetaReplicaNodes();
    StringBuilder info = new StringBuilder("Migrating meta locations:");
    for (String metaReplicaNode : metaReplicaNodes) {
      int replicaId = zooKeeper.getZNodePaths().getMetaReplicaIdFromZNode(metaReplicaNode);
      RegionState state = MetaTableLocator.getMetaRegionState(zooKeeper, replicaId);
      info.append(" ").append(state);
      put.setTimestamp(state.getStamp());
      MetaTableAccessor.addRegionInfo(put, state.getRegion());
      if (state.getServerName() != null) {
        MetaTableAccessor.addLocation(put, state.getServerName(), HConstants.NO_SEQNUM, replicaId);
      }
      put.add(CellBuilderFactory.create(CellBuilderType.SHALLOW_COPY).setRow(put.getRow())
        .setFamily(HConstants.CATALOG_FAMILY)
        .setQualifier(RegionStateStore.getStateColumn(replicaId)).setTimestamp(put.getTimestamp())
        .setType(Cell.Type.Put).setValue(Bytes.toBytes(state.getState().name())).build());
    }
    if (!put.isEmpty()) {
      LOG.info(info.toString());
      masterRegion.update(r -> r.put(put));
    } else {
      LOG.info("No meta location available on zookeeper, skip migrating...");
    }
  }

  /**
   * Finish initialization of HMaster after becoming the primary master.
   * <p/>
   * The startup order is a bit complicated but very important, do not change it unless you know
   * what you are doing.
   * <ol>
   * <li>Initialize file system based components - file system manager, wal manager, table
   * descriptors, etc</li>
   * <li>Publish cluster id</li>
   * <li>Here comes the most complicated part - initialize server manager, assignment manager and
   * region server tracker
   * <ol type='i'>
   * <li>Create server manager</li>
   * <li>Create master local region</li>
   * <li>Create procedure executor, load the procedures, but do not start workers. We will start it
   * later after we finish scheduling SCPs to avoid scheduling duplicated SCPs for the same
   * server</li>
   * <li>Create assignment manager and start it, load the meta region state, but do not load data
   * from meta region</li>
   * <li>Start region server tracker, construct the online servers set and find out dead servers and
   * schedule SCP for them. The online servers will be constructed by scanning zk, and we will also
   * scan the wal directory and load from master local region to find out possible live region
   * servers, and the differences between these two sets are the dead servers</li>
   * </ol>
   * </li>
   * <li>If this is a new deploy, schedule a InitMetaProcedure to initialize meta</li>
   * <li>Start necessary service threads - balancer, catalog janitor, executor services, and also
   * the procedure executor, etc. Notice that the balancer must be created first as assignment
   * manager may use it when assigning regions.</li>
   * <li>Wait for meta to be initialized if necessary, start table state manager.</li>
   * <li>Wait for enough region servers to check-in</li>
   * <li>Let assignment manager load data from meta and construct region states</li>
   * <li>Start all other things such as chore services, etc</li>
   * </ol>
   * <p/>
   * Notice that now we will not schedule a special procedure to make meta online(unless the first
   * time where meta has not been created yet), we will rely on SCP to bring meta online.
   */
  private void finishActiveMasterInitialization() throws IOException, InterruptedException,
    KeeperException, ReplicationException, DeserializationException {
    /*
     * We are active master now... go initialize components we need to run.
     */
    startupTaskGroup.addTask("Initializing Master file system");

    this.masterActiveTime = EnvironmentEdgeManager.currentTime();
    // TODO: Do this using Dependency Injection, using PicoContainer, Guice or Spring.

    // always initialize the MemStoreLAB as we use a region to store data in master now, see
    // localStore.
    initializeMemStoreChunkCreator(null);
    this.fileSystemManager = new MasterFileSystem(conf);
    this.walManager = new MasterWalManager(this);

    // warm-up HTDs cache on master initialization
    if (preLoadTableDescriptors) {
      startupTaskGroup.addTask("Pre-loading table descriptors");
      this.tableDescriptors.getAll();
    }

    // Publish cluster ID; set it in Master too. The superclass RegionServer does this later but
    // only after it has checked in with the Master. At least a few tests ask Master for clusterId
    // before it has called its run method and before RegionServer has done the reportForDuty.
    ClusterId clusterId = fileSystemManager.getClusterId();
    startupTaskGroup.addTask("Publishing Cluster ID " + clusterId + " in ZooKeeper");
    ZKClusterId.setClusterId(this.zooKeeper, fileSystemManager.getClusterId());
    this.clusterId = clusterId.toString();

    // Precaution. Put in place the old hbck1 lock file to fence out old hbase1s running their
    // hbck1s against an hbase2 cluster; it could do damage. To skip this behavior, set
    // hbase.write.hbck1.lock.file to false.
    if (this.conf.getBoolean("hbase.write.hbck1.lock.file", true)) {
      Pair<Path, FSDataOutputStream> result = null;
      try {
        result = HBaseFsck.checkAndMarkRunningHbck(this.conf,
          HBaseFsck.createLockRetryCounterFactory(this.conf).create());
      } finally {
        if (result != null) {
          Closeables.close(result.getSecond(), true);
        }
      }
    }

    startupTaskGroup.addTask("Initialize ServerManager and schedule SCP for crash servers");
    // The below two managers must be created before loading procedures, as they will be used during
    // loading.
    // initialize master local region
    masterRegion = MasterRegionFactory.create(this);
    rsListStorage = new MasterRegionServerList(masterRegion, this);

    // Initialize the ServerManager and register it as a configuration observer
    this.serverManager = createServerManager(this, rsListStorage);
    this.configurationManager.registerObserver(this.serverManager);

    this.syncReplicationReplayWALManager = new SyncReplicationReplayWALManager(this);
    if (
      !conf.getBoolean(HBASE_SPLIT_WAL_COORDINATED_BY_ZK, DEFAULT_HBASE_SPLIT_COORDINATED_BY_ZK)
    ) {
      this.splitWALManager = new SplitWALManager(this);
    }

    tryMigrateMetaLocationsFromZooKeeper();

    createProcedureExecutor();
    Map<Class<?>, List<Procedure<MasterProcedureEnv>>> procsByType = procedureExecutor
      .getActiveProceduresNoCopy().stream().collect(Collectors.groupingBy(p -> p.getClass()));

    // Create Assignment Manager
    this.assignmentManager = createAssignmentManager(this, masterRegion);
    this.assignmentManager.start();
    // TODO: TRSP can perform as the sub procedure for other procedures, so even if it is marked as
    // completed, it could still be in the procedure list. This is a bit strange but is another
    // story, need to verify the implementation for ProcedureExecutor and ProcedureStore.
    List<TransitRegionStateProcedure> ritList =
      procsByType.getOrDefault(TransitRegionStateProcedure.class, Collections.emptyList()).stream()
        .filter(p -> !p.isFinished()).map(p -> (TransitRegionStateProcedure) p)
        .collect(Collectors.toList());
    this.assignmentManager.setupRIT(ritList);

    // Start RegionServerTracker with listing of servers found with exiting SCPs -- these should
    // be registered in the deadServers set -- and the servernames loaded from the WAL directory
    // and master local region that COULD BE 'alive'(we'll schedule SCPs for each and let SCP figure
    // it out).
    // We also pass dirs that are already 'splitting'... so we can do some checks down in tracker.
    // TODO: Generate the splitting and live Set in one pass instead of two as we currently do.
    this.regionServerTracker.upgrade(
      procsByType.getOrDefault(ServerCrashProcedure.class, Collections.emptyList()).stream()
        .map(p -> (ServerCrashProcedure) p).map(p -> p.getServerName()).collect(Collectors.toSet()),
      Sets.union(rsListStorage.getAll(), walManager.getLiveServersFromWALDir()),
      walManager.getSplittingServersFromWALDir());
    // This manager must be accessed AFTER hbase:meta is confirmed on line..
    this.tableStateManager = new TableStateManager(this);

    startupTaskGroup.addTask("Initializing ZK system trackers");
    initializeZKBasedSystemTrackers();
    startupTaskGroup.addTask("Loading last flushed sequence id of regions");
    try {
      this.serverManager.loadLastFlushedSequenceIds();
    } catch (IOException e) {
      LOG.info("Failed to load last flushed sequence id of regions" + " from file system", e);
    }
    // Set ourselves as active Master now our claim has succeeded up in zk.
    this.activeMaster = true;

    // Start the Zombie master detector after setting master as active, see HBASE-21535
    Thread zombieDetector = new Thread(new MasterInitializationMonitor(this),
      "ActiveMasterInitializationMonitor-" + EnvironmentEdgeManager.currentTime());
    zombieDetector.setDaemon(true);
    zombieDetector.start();

    if (!maintenanceMode) {
      startupTaskGroup.addTask("Initializing master coprocessors");
      setQuotasObserver(conf);
      initializeCoprocessorHost(conf);
    } else {
      // start an in process region server for carrying system regions
      maintenanceRegionServer =
        JVMClusterUtil.createRegionServerThread(getConfiguration(), HRegionServer.class, 0);
      maintenanceRegionServer.start();
    }

    // Checking if meta needs initializing.
    startupTaskGroup.addTask("Initializing meta table if this is a new deploy");
    InitMetaProcedure initMetaProc = null;
    // Print out state of hbase:meta on startup; helps debugging.
    if (!this.assignmentManager.getRegionStates().hasTableRegionStates(TableName.META_TABLE_NAME)) {
      Optional<InitMetaProcedure> optProc = procedureExecutor.getProcedures().stream()
        .filter(p -> p instanceof InitMetaProcedure).map(o -> (InitMetaProcedure) o).findAny();
      initMetaProc = optProc.orElseGet(() -> {
        // schedule an init meta procedure if meta has not been deployed yet
        InitMetaProcedure temp = new InitMetaProcedure();
        procedureExecutor.submitProcedure(temp);
        return temp;
      });
    }

    // initialize load balancer
    this.balancer.setMasterServices(this);
    this.balancer.initialize();
    this.balancer.updateClusterMetrics(getClusterMetricsWithoutCoprocessor());

    // try migrate replication data
    ZKReplicationQueueStorageForMigration oldReplicationQueueStorage =
      new ZKReplicationQueueStorageForMigration(zooKeeper, conf);
    // check whether there are something to migrate and we haven't scheduled a migration procedure
    // yet
    if (
      oldReplicationQueueStorage.hasData() && procedureExecutor.getProcedures().stream()
        .allMatch(p -> !(p instanceof MigrateReplicationQueueFromZkToTableProcedure))
    ) {
      procedureExecutor.submitProcedure(new MigrateReplicationQueueFromZkToTableProcedure());
    }
    // start up all service threads.
    startupTaskGroup.addTask("Initializing master service threads");
    startServiceThreads();
    // wait meta to be initialized after we start procedure executor
    if (initMetaProc != null) {
      initMetaProc.await();
    }
    // Wake up this server to check in
    sleeper.skipSleepCycle();

    // Wait for region servers to report in.
    // With this as part of master initialization, it precludes our being able to start a single
    // server that is both Master and RegionServer. Needs more thought. TODO.
    String statusStr = "Wait for region servers to report in";
    MonitoredTask waitRegionServer = startupTaskGroup.addTask(statusStr);
    LOG.info(Objects.toString(waitRegionServer));
    waitForRegionServers(waitRegionServer);

    // Check if master is shutting down because issue initializing regionservers or balancer.
    if (isStopped()) {
      return;
    }

    startupTaskGroup.addTask("Starting assignment manager");
    // FIRST HBASE:META READ!!!!
    // The below cannot make progress w/o hbase:meta being online.
    // This is the FIRST attempt at going to hbase:meta. Meta on-lining is going on in background
    // as procedures run -- in particular SCPs for crashed servers... One should put up hbase:meta
    // if it is down. It may take a while to come online. So, wait here until meta if for sure
    // available. That's what waitForMetaOnline does.
    if (!waitForMetaOnline()) {
      return;
    }

    TableDescriptor metaDescriptor = tableDescriptors.get(TableName.META_TABLE_NAME);
    final ColumnFamilyDescriptor tableFamilyDesc =
      metaDescriptor.getColumnFamily(HConstants.TABLE_FAMILY);
    final ColumnFamilyDescriptor replBarrierFamilyDesc =
      metaDescriptor.getColumnFamily(HConstants.REPLICATION_BARRIER_FAMILY);

    this.assignmentManager.joinCluster();
    // The below depends on hbase:meta being online.
    this.assignmentManager.processOfflineRegions();
    // this must be called after the above processOfflineRegions to prevent race
    this.assignmentManager.wakeMetaLoadedEvent();

    // for migrating from a version without HBASE-25099, and also for honoring the configuration
    // first.
    if (conf.get(HConstants.META_REPLICAS_NUM) != null) {
      int replicasNumInConf =
        conf.getInt(HConstants.META_REPLICAS_NUM, HConstants.DEFAULT_META_REPLICA_NUM);
      TableDescriptor metaDesc = tableDescriptors.get(TableName.META_TABLE_NAME);
      if (metaDesc.getRegionReplication() != replicasNumInConf) {
        // it is possible that we already have some replicas before upgrading, so we must set the
        // region replication number in meta TableDescriptor directly first, without creating a
        // ModifyTableProcedure, otherwise it may cause a double assign for the meta replicas.
        int existingReplicasCount =
          assignmentManager.getRegionStates().getRegionsOfTable(TableName.META_TABLE_NAME).size();
        if (existingReplicasCount > metaDesc.getRegionReplication()) {
          LOG.info("Update replica count of hbase:meta from {}(in TableDescriptor)"
            + " to {}(existing ZNodes)", metaDesc.getRegionReplication(), existingReplicasCount);
          metaDesc = TableDescriptorBuilder.newBuilder(metaDesc)
            .setRegionReplication(existingReplicasCount).build();
          tableDescriptors.update(metaDesc);
        }
        // check again, and issue a ModifyTableProcedure if needed
        if (metaDesc.getRegionReplication() != replicasNumInConf) {
          LOG.info(
            "The {} config is {} while the replica count in TableDescriptor is {}"
              + " for hbase:meta, altering...",
            HConstants.META_REPLICAS_NUM, replicasNumInConf, metaDesc.getRegionReplication());
          procedureExecutor.submitProcedure(new ModifyTableProcedure(
            procedureExecutor.getEnvironment(), TableDescriptorBuilder.newBuilder(metaDesc)
              .setRegionReplication(replicasNumInConf).build(),
            null, metaDesc, false, true));
        }
      }
    }
    // Initialize after meta is up as below scans meta
    FavoredNodesManager fnm = getFavoredNodesManager();
    if (fnm != null) {
      fnm.initializeFromMeta();
    }

    // set cluster status again after user regions are assigned
    this.balancer.updateClusterMetrics(getClusterMetricsWithoutCoprocessor());

    // Start balancer and meta catalog janitor after meta and regions have been assigned.
    startupTaskGroup.addTask("Starting balancer and catalog janitor");
    this.clusterStatusChore = new ClusterStatusChore(this, balancer);
    getChoreService().scheduleChore(clusterStatusChore);
    this.balancerChore = new BalancerChore(this);
    if (!disableBalancerChoreForTest) {
      getChoreService().scheduleChore(balancerChore);
    }
    if (regionNormalizerManager != null) {
      getChoreService().scheduleChore(regionNormalizerManager.getRegionNormalizerChore());
    }
    this.catalogJanitorChore = new CatalogJanitor(this);
    getChoreService().scheduleChore(catalogJanitorChore);
    this.hbckChore = new HbckChore(this);
    getChoreService().scheduleChore(hbckChore);
    this.serverManager.startChore();

    // Only for rolling upgrade, where we need to migrate the data in namespace table to meta table.
    if (!waitForNamespaceOnline()) {
      return;
    }
    startupTaskGroup.addTask("Starting cluster schema service");
    try {
      initClusterSchemaService();
    } catch (IllegalStateException e) {
      if (
        e.getCause() != null && e.getCause() instanceof NoSuchColumnFamilyException
          && tableFamilyDesc == null && replBarrierFamilyDesc == null
      ) {
        LOG.info("ClusterSchema service could not be initialized. This is "
          + "expected during HBase 1 to 2 upgrade", e);
      } else {
        throw e;
      }
    }

    if (this.cpHost != null) {
      try {
        this.cpHost.preMasterInitialization();
      } catch (IOException e) {
        LOG.error("Coprocessor preMasterInitialization() hook failed", e);
      }
    }

    LOG.info(String.format("Master has completed initialization %.3fsec",
      (EnvironmentEdgeManager.currentTime() - masterActiveTime) / 1000.0f));
    this.masterFinishedInitializationTime = EnvironmentEdgeManager.currentTime();
    configurationManager.registerObserver(this.balancer);
    configurationManager.registerObserver(this.logCleanerPool);
    configurationManager.registerObserver(this.logCleaner);
    configurationManager.registerObserver(this.regionsRecoveryConfigManager);
    configurationManager.registerObserver(this.exclusiveHFileCleanerPool);
    if (this.sharedHFileCleanerPool != null) {
      configurationManager.registerObserver(this.sharedHFileCleanerPool);
    }
    if (this.hfileCleaners != null) {
      for (HFileCleaner cleaner : hfileCleaners) {
        configurationManager.registerObserver(cleaner);
      }
    }
    // Set master as 'initialized'.
    setInitialized(true);
    startupTaskGroup.markComplete("Initialization successful");
    MonitoredTask status =
      TaskMonitor.get().createStatus("Progress after master initialized", false, true);

    if (tableFamilyDesc == null && replBarrierFamilyDesc == null) {
      // create missing CFs in meta table after master is set to 'initialized'.
      createMissingCFsInMetaDuringUpgrade(metaDescriptor);

      // Throwing this Exception to abort active master is painful but this
      // seems the only way to add missing CFs in meta while upgrading from
      // HBase 1 to 2 (where HBase 2 has HBASE-23055 & HBASE-23782 checked-in).
      // So, why do we abort active master after adding missing CFs in meta?
      // When we reach here, we would have already bypassed NoSuchColumnFamilyException
      // in initClusterSchemaService(), meaning ClusterSchemaService is not
      // correctly initialized but we bypassed it. Similarly, we bypassed
      // tableStateManager.start() as well. Hence, we should better abort
      // current active master because our main task - adding missing CFs
      // in meta table is done (possible only after master state is set as
      // initialized) at the expense of bypassing few important tasks as part
      // of active master init routine. So now we abort active master so that
      // next active master init will not face any issues and all mandatory
      // services will be started during master init phase.
      throw new PleaseRestartMasterException("Aborting active master after missing"
        + " CFs are successfully added in meta. Subsequent active master "
        + "initialization should be uninterrupted");
    }

    if (maintenanceMode) {
      LOG.info("Detected repair mode, skipping final initialization steps.");
      return;
    }

    assignmentManager.checkIfShouldMoveSystemRegionAsync();
    status.setStatus("Starting quota manager");
    initQuotaManager();
    if (QuotaUtil.isQuotaEnabled(conf)) {
      // Create the quota snapshot notifier
      spaceQuotaSnapshotNotifier = createQuotaSnapshotNotifier();
      spaceQuotaSnapshotNotifier.initialize(getConnection());
      this.quotaObserverChore = new QuotaObserverChore(this, getMasterMetrics());
      // Start the chore to read the region FS space reports and act on them
      getChoreService().scheduleChore(quotaObserverChore);

      this.snapshotQuotaChore = new SnapshotQuotaObserverChore(this, getMasterMetrics());
      // Start the chore to read snapshots and add their usage to table/NS quotas
      getChoreService().scheduleChore(snapshotQuotaChore);
    }
    final SlowLogMasterService slowLogMasterService = new SlowLogMasterService(conf, this);
    slowLogMasterService.init();

    WALEventTrackerTableCreator.createIfNeededAndNotExists(conf, this);
    // Create REPLICATION.SINK_TRACKER table if needed.
    ReplicationSinkTrackerTableCreator.createIfNeededAndNotExists(conf, this);

    // clear the dead servers with same host name and port of online server because we are not
    // removing dead server with same hostname and port of rs which is trying to check in before
    // master initialization. See HBASE-5916.
    this.serverManager.clearDeadServersWithSameHostNameAndPortOfOnlineServer();

    // Check and set the znode ACLs if needed in case we are overtaking a non-secure configuration
    status.setStatus("Checking ZNode ACLs");
    zooKeeper.checkAndSetZNodeAcls();

    status.setStatus("Initializing MOB Cleaner");
    initMobCleaner();

    // delete the stale data for replication sync up tool if necessary
    status.setStatus("Cleanup ReplicationSyncUp status if necessary");
    Path replicationSyncUpInfoFile =
      new Path(new Path(dataRootDir, ReplicationSyncUp.INFO_DIR), ReplicationSyncUp.INFO_FILE);
    if (dataFs.exists(replicationSyncUpInfoFile)) {
      // info file is available, load the timestamp and use it to clean up stale data in replication
      // queue storage.
      byte[] data;
      try (FSDataInputStream in = dataFs.open(replicationSyncUpInfoFile)) {
        data = ByteStreams.toByteArray(in);
      }
      ReplicationSyncUpToolInfo info = null;
      try {
        info = JsonMapper.fromJson(Bytes.toString(data), ReplicationSyncUpToolInfo.class);
      } catch (JsonParseException e) {
        // usually this should be a partial file, which means the ReplicationSyncUp tool did not
        // finish properly, so not a problem. Here we do not clean up the status as we do not know
        // the reason why the tool did not finish properly, so let users clean the status up
        // manually
        LOG.warn("failed to parse replication sync up info file, ignore and continue...", e);
      }
      if (info != null) {
        LOG.info("Remove last sequence ids and hfile references which are written before {}({})",
          info.getStartTimeMs(), DateTimeFormatter.ISO_DATE_TIME.withZone(ZoneId.systemDefault())
            .format(Instant.ofEpochMilli(info.getStartTimeMs())));
        replicationPeerManager.getQueueStorage()
          .removeLastSequenceIdsAndHFileRefsBefore(info.getStartTimeMs());
        // delete the file after removing the stale data, so next time we do not need to do this
        // again.
        dataFs.delete(replicationSyncUpInfoFile, false);
      }
    }
    status.setStatus("Calling postStartMaster coprocessors");
    if (this.cpHost != null) {
      // don't let cp initialization errors kill the master
      try {
        this.cpHost.postStartMaster();
      } catch (IOException ioe) {
        LOG.error("Coprocessor postStartMaster() hook failed", ioe);
      }
    }

    zombieDetector.interrupt();

    /*
     * After master has started up, lets do balancer post startup initialization. Since this runs in
     * activeMasterManager thread, it should be fine.
     */
    long start = EnvironmentEdgeManager.currentTime();
    this.balancer.postMasterStartupInitialize();
    if (LOG.isDebugEnabled()) {
      LOG.debug("Balancer post startup initialization complete, took "
        + ((EnvironmentEdgeManager.currentTime() - start) / 1000) + " seconds");
    }

    this.rollingUpgradeChore = new RollingUpgradeChore(this);
    getChoreService().scheduleChore(rollingUpgradeChore);

    this.oldWALsDirSizeChore = new OldWALsDirSizeChore(this);
    getChoreService().scheduleChore(this.oldWALsDirSizeChore);

    status.markComplete("Progress after master initialized complete");
  }

  /**
   * Used for testing only to set Mock objects.
   * @param hbckChore hbckChore
   */
  public void setHbckChoreForTesting(HbckChore hbckChore) {
    this.hbckChore = hbckChore;
  }

  /**
   * Used for testing only to set Mock objects.
   * @param catalogJanitorChore catalogJanitorChore
   */
  public void setCatalogJanitorChoreForTesting(CatalogJanitor catalogJanitorChore) {
    this.catalogJanitorChore = catalogJanitorChore;
  }

  private void createMissingCFsInMetaDuringUpgrade(TableDescriptor metaDescriptor)
    throws IOException {
    TableDescriptor newMetaDesc = TableDescriptorBuilder.newBuilder(metaDescriptor)
      .setColumnFamily(FSTableDescriptors.getTableFamilyDescForMeta(conf))
      .setColumnFamily(FSTableDescriptors.getReplBarrierFamilyDescForMeta()).build();
    long pid = this.modifyTable(TableName.META_TABLE_NAME, () -> newMetaDesc, 0, 0, false);
    int tries = 30;
    while (
      !(getMasterProcedureExecutor().isFinished(pid)) && getMasterProcedureExecutor().isRunning()
        && tries > 0
    ) {
      try {
        Thread.sleep(1000);
      } catch (InterruptedException e) {
        throw new IOException("Wait interrupted", e);
      }
      tries--;
    }
    if (tries <= 0) {
      throw new HBaseIOException(
        "Failed to add table and rep_barrier CFs to meta in a given time.");
    } else {
      Procedure<?> result = getMasterProcedureExecutor().getResult(pid);
      if (result != null && result.isFailed()) {
        throw new IOException("Failed to add table and rep_barrier CFs to meta. "
          + MasterProcedureUtil.unwrapRemoteIOException(result));
      }
    }
  }

  /**
   * Check hbase:meta is up and ready for reading. For use during Master startup only.
   * @return True if meta is UP and online and startup can progress. Otherwise, meta is not online
   *         and we will hold here until operator intervention.
   */
  @InterfaceAudience.Private
  public boolean waitForMetaOnline() {
    return isRegionOnline(RegionInfoBuilder.FIRST_META_REGIONINFO);
  }

  /**
   * @return True if region is online and scannable else false if an error or shutdown (Otherwise we
   *         just block in here holding up all forward-progess).
   */
  private boolean isRegionOnline(RegionInfo ri) {
    RetryCounter rc = null;
    while (!isStopped()) {
      RegionState rs = this.assignmentManager.getRegionStates().getRegionState(ri);
      if (rs != null && rs.isOpened()) {
        if (this.getServerManager().isServerOnline(rs.getServerName())) {
          return true;
        }
      }
      // Region is not OPEN.
      Optional<Procedure<MasterProcedureEnv>> optProc = this.procedureExecutor.getProcedures()
        .stream().filter(p -> p instanceof ServerCrashProcedure).findAny();
      // TODO: Add a page to refguide on how to do repair. Have this log message point to it.
      // Page will talk about loss of edits, how to schedule at least the meta WAL recovery, and
      // then how to assign including how to break region lock if one held.
      LOG.warn(
        "{} is NOT online; state={}; ServerCrashProcedures={}. Master startup cannot "
          + "progress, in holding-pattern until region onlined.",
        ri.getRegionNameAsString(), rs, optProc.isPresent());
      // Check once-a-minute.
      if (rc == null) {
        rc = new RetryCounterFactory(Integer.MAX_VALUE, 1000, 60_000).create();
      }
      Threads.sleep(rc.getBackoffTimeAndIncrementAttempts());
    }
    return false;
  }

  /**
   * Check hbase:namespace table is assigned. If not, startup will hang looking for the ns table
   * <p/>
   * This is for rolling upgrading, later we will migrate the data in ns table to the ns family of
   * meta table. And if this is a new cluster, this method will return immediately as there will be
   * no namespace table/region.
   * @return True if namespace table is up/online.
   */
  private boolean waitForNamespaceOnline() throws IOException {
    TableState nsTableState =
      MetaTableAccessor.getTableState(getConnection(), TableName.NAMESPACE_TABLE_NAME);
    if (nsTableState == null || nsTableState.isDisabled()) {
      // this means we have already migrated the data and disabled or deleted the namespace table,
      // or this is a new deploy which does not have a namespace table from the beginning.
      return true;
    }
    List<RegionInfo> ris =
      this.assignmentManager.getRegionStates().getRegionsOfTable(TableName.NAMESPACE_TABLE_NAME);
    if (ris.isEmpty()) {
      // maybe this will not happen any more, but anyway, no harm to add a check here...
      return true;
    }
    // Else there are namespace regions up in meta. Ensure they are assigned before we go on.
    for (RegionInfo ri : ris) {
      if (!isRegionOnline(ri)) {
        return false;
      }
    }
    return true;
  }

  /**
   * Adds the {@code MasterQuotasObserver} to the list of configured Master observers to
   * automatically remove quotas for a table when that table is deleted.
   */
  @InterfaceAudience.Private
  public void updateConfigurationForQuotasObserver(Configuration conf) {
    // We're configured to not delete quotas on table deletion, so we don't need to add the obs.
    if (
      !conf.getBoolean(MasterQuotasObserver.REMOVE_QUOTA_ON_TABLE_DELETE,
        MasterQuotasObserver.REMOVE_QUOTA_ON_TABLE_DELETE_DEFAULT)
    ) {
      return;
    }
    String[] masterCoprocs = conf.getStrings(CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY);
    final int length = null == masterCoprocs ? 0 : masterCoprocs.length;
    String[] updatedCoprocs = new String[length + 1];
    if (length > 0) {
      System.arraycopy(masterCoprocs, 0, updatedCoprocs, 0, masterCoprocs.length);
    }
    updatedCoprocs[length] = MasterQuotasObserver.class.getName();
    conf.setStrings(CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY, updatedCoprocs);
  }

  private void initMobCleaner() {
    this.mobFileCleanerChore = new MobFileCleanerChore(this);
    getChoreService().scheduleChore(mobFileCleanerChore);
    this.mobFileCompactionChore = new MobFileCompactionChore(this);
    getChoreService().scheduleChore(mobFileCompactionChore);
  }

  /**
   * <p>
   * Create a {@link ServerManager} instance.
   * </p>
   * <p>
   * Will be overridden in tests.
   * </p>
   */
  @InterfaceAudience.Private
  protected ServerManager createServerManager(MasterServices master, RegionServerList storage)
    throws IOException {
    // We put this out here in a method so can do a Mockito.spy and stub it out
    // w/ a mocked up ServerManager.
    setupClusterConnection();
    return new ServerManager(master, storage);
  }

  private void waitForRegionServers(final MonitoredTask status)
    throws IOException, InterruptedException {
    this.serverManager.waitForRegionServers(status);
  }

  // Will be overridden in tests
  @InterfaceAudience.Private
  protected void initClusterSchemaService() throws IOException, InterruptedException {
    this.clusterSchemaService = new ClusterSchemaServiceImpl(this);
    this.clusterSchemaService.startAsync();
    try {
      this.clusterSchemaService
        .awaitRunning(getConfiguration().getInt(HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS,
          DEFAULT_HBASE_MASTER_WAIT_ON_SERVICE_IN_SECONDS), TimeUnit.SECONDS);
    } catch (TimeoutException toe) {
      throw new IOException("Timedout starting ClusterSchemaService", toe);
    }
  }

  private void initQuotaManager() throws IOException {
    MasterQuotaManager quotaManager = new MasterQuotaManager(this);
    quotaManager.start();
    this.quotaManager = quotaManager;
  }

  private SpaceQuotaSnapshotNotifier createQuotaSnapshotNotifier() {
    SpaceQuotaSnapshotNotifier notifier =
      SpaceQuotaSnapshotNotifierFactory.getInstance().create(getConfiguration());
    return notifier;
  }

  public boolean isCatalogJanitorEnabled() {
    return catalogJanitorChore != null ? catalogJanitorChore.getEnabled() : false;
  }

  boolean isCleanerChoreEnabled() {
    boolean hfileCleanerFlag = true, logCleanerFlag = true;

    if (getHFileCleaner() != null) {
      hfileCleanerFlag = getHFileCleaner().getEnabled();
    }

    if (logCleaner != null) {
      logCleanerFlag = logCleaner.getEnabled();
    }

    return (hfileCleanerFlag && logCleanerFlag);
  }

  @Override
  public ServerManager getServerManager() {
    return this.serverManager;
  }

  @Override
  public MasterFileSystem getMasterFileSystem() {
    return this.fileSystemManager;
  }

  @Override
  public MasterWalManager getMasterWalManager() {
    return this.walManager;
  }

  @Override
  public SplitWALManager getSplitWALManager() {
    return splitWALManager;
  }

  @Override
  public TableStateManager getTableStateManager() {
    return tableStateManager;
  }

  /*
   * Start up all services. If any of these threads gets an unhandled exception then they just die
   * with a logged message. This should be fine because in general, we do not expect the master to
   * get such unhandled exceptions as OOMEs; it should be lightly loaded. See what HRegionServer
   * does if need to install an unexpected exception handler.
   */
  private void startServiceThreads() throws IOException {
    // Start the executor service pools
    final int masterOpenRegionPoolSize = conf.getInt(HConstants.MASTER_OPEN_REGION_THREADS,
      HConstants.MASTER_OPEN_REGION_THREADS_DEFAULT);
    executorService.startExecutorService(executorService.new ExecutorConfig()
      .setExecutorType(ExecutorType.MASTER_OPEN_REGION).setCorePoolSize(masterOpenRegionPoolSize));
    final int masterCloseRegionPoolSize = conf.getInt(HConstants.MASTER_CLOSE_REGION_THREADS,
      HConstants.MASTER_CLOSE_REGION_THREADS_DEFAULT);
    executorService.startExecutorService(
      executorService.new ExecutorConfig().setExecutorType(ExecutorType.MASTER_CLOSE_REGION)
        .setCorePoolSize(masterCloseRegionPoolSize));
    final int masterServerOpThreads = conf.getInt(HConstants.MASTER_SERVER_OPERATIONS_THREADS,
      HConstants.MASTER_SERVER_OPERATIONS_THREADS_DEFAULT);
    executorService.startExecutorService(
      executorService.new ExecutorConfig().setExecutorType(ExecutorType.MASTER_SERVER_OPERATIONS)
        .setCorePoolSize(masterServerOpThreads));
    final int masterServerMetaOpsThreads =
      conf.getInt(HConstants.MASTER_META_SERVER_OPERATIONS_THREADS,
        HConstants.MASTER_META_SERVER_OPERATIONS_THREADS_DEFAULT);
    executorService.startExecutorService(executorService.new ExecutorConfig()
      .setExecutorType(ExecutorType.MASTER_META_SERVER_OPERATIONS)
      .setCorePoolSize(masterServerMetaOpsThreads));
    final int masterLogReplayThreads = conf.getInt(HConstants.MASTER_LOG_REPLAY_OPS_THREADS,
      HConstants.MASTER_LOG_REPLAY_OPS_THREADS_DEFAULT);
    executorService.startExecutorService(executorService.new ExecutorConfig()
      .setExecutorType(ExecutorType.M_LOG_REPLAY_OPS).setCorePoolSize(masterLogReplayThreads));
    final int masterSnapshotThreads = conf.getInt(SnapshotManager.SNAPSHOT_POOL_THREADS_KEY,
      SnapshotManager.SNAPSHOT_POOL_THREADS_DEFAULT);
    executorService.startExecutorService(
      executorService.new ExecutorConfig().setExecutorType(ExecutorType.MASTER_SNAPSHOT_OPERATIONS)
        .setCorePoolSize(masterSnapshotThreads).setAllowCoreThreadTimeout(true));
    final int masterMergeDispatchThreads = conf.getInt(HConstants.MASTER_MERGE_DISPATCH_THREADS,
      HConstants.MASTER_MERGE_DISPATCH_THREADS_DEFAULT);
    executorService.startExecutorService(
      executorService.new ExecutorConfig().setExecutorType(ExecutorType.MASTER_MERGE_OPERATIONS)
        .setCorePoolSize(masterMergeDispatchThreads).setAllowCoreThreadTimeout(true));

    // We depend on there being only one instance of this executor running
    // at a time. To do concurrency, would need fencing of enable/disable of
    // tables.
    // Any time changing this maxThreads to > 1, pls see the comment at
    // AccessController#postCompletedCreateTableAction
    executorService.startExecutorService(executorService.new ExecutorConfig()
      .setExecutorType(ExecutorType.MASTER_TABLE_OPERATIONS).setCorePoolSize(1));
    startProcedureExecutor();

    // Create log cleaner thread pool
    logCleanerPool = DirScanPool.getLogCleanerScanPool(conf);
    Map<String, Object> params = new HashMap<>();
    params.put(MASTER, this);
    // Start log cleaner thread
    int cleanerInterval =
      conf.getInt(HBASE_MASTER_CLEANER_INTERVAL, DEFAULT_HBASE_MASTER_CLEANER_INTERVAL);
    this.logCleaner =
      new LogCleaner(cleanerInterval, this, conf, getMasterWalManager().getFileSystem(),
        getMasterWalManager().getOldLogDir(), logCleanerPool, params);
    getChoreService().scheduleChore(logCleaner);

    Path archiveDir = HFileArchiveUtil.getArchivePath(conf);

    // Create custom archive hfile cleaners
    String[] paths = conf.getStrings(HFileCleaner.HFILE_CLEANER_CUSTOM_PATHS);
    // todo: handle the overlap issues for the custom paths

    if (paths != null && paths.length > 0) {
      if (conf.getStrings(HFileCleaner.HFILE_CLEANER_CUSTOM_PATHS_PLUGINS) == null) {
        Set<String> cleanerClasses = new HashSet<>();
        String[] cleaners = conf.getStrings(HFileCleaner.MASTER_HFILE_CLEANER_PLUGINS);
        if (cleaners != null) {
          Collections.addAll(cleanerClasses, cleaners);
        }
        conf.setStrings(HFileCleaner.HFILE_CLEANER_CUSTOM_PATHS_PLUGINS,
          cleanerClasses.toArray(new String[cleanerClasses.size()]));
        LOG.info("Archive custom cleaner paths: {}, plugins: {}", Arrays.asList(paths),
          cleanerClasses);
      }
      // share the hfile cleaner pool in custom paths
      sharedHFileCleanerPool = DirScanPool.getHFileCleanerScanPool(conf.get(CUSTOM_POOL_SIZE, "6"));
      for (int i = 0; i < paths.length; i++) {
        Path path = new Path(paths[i].trim());
        HFileCleaner cleaner =
          new HFileCleaner("ArchiveCustomHFileCleaner-" + path.getName(), cleanerInterval, this,
            conf, getMasterFileSystem().getFileSystem(), new Path(archiveDir, path),
            HFileCleaner.HFILE_CLEANER_CUSTOM_PATHS_PLUGINS, sharedHFileCleanerPool, params, null);
        hfileCleaners.add(cleaner);
        hfileCleanerPaths.add(path);
      }
    }

    // Create the whole archive dir cleaner thread pool
    exclusiveHFileCleanerPool = DirScanPool.getHFileCleanerScanPool(conf);
    hfileCleaners.add(0,
      new HFileCleaner(cleanerInterval, this, conf, getMasterFileSystem().getFileSystem(),
        archiveDir, exclusiveHFileCleanerPool, params, hfileCleanerPaths));
    hfileCleanerPaths.add(0, archiveDir);
    // Schedule all the hfile cleaners
    for (HFileCleaner hFileCleaner : hfileCleaners) {
      getChoreService().scheduleChore(hFileCleaner);
    }

    // Regions Reopen based on very high storeFileRefCount is considered enabled
    // only if hbase.regions.recovery.store.file.ref.count has value > 0
    final int maxStoreFileRefCount = conf.getInt(HConstants.STORE_FILE_REF_COUNT_THRESHOLD,
      HConstants.DEFAULT_STORE_FILE_REF_COUNT_THRESHOLD);
    if (maxStoreFileRefCount > 0) {
      this.regionsRecoveryChore = new RegionsRecoveryChore(this, conf, this);
      getChoreService().scheduleChore(this.regionsRecoveryChore);
    } else {
      LOG.info(
        "Reopening regions with very high storeFileRefCount is disabled. "
          + "Provide threshold value > 0 for {} to enable it.",
        HConstants.STORE_FILE_REF_COUNT_THRESHOLD);
    }

    this.regionsRecoveryConfigManager = new RegionsRecoveryConfigManager(this);

    replicationBarrierCleaner =
      new ReplicationBarrierCleaner(conf, this, getConnection(), replicationPeerManager);
    getChoreService().scheduleChore(replicationBarrierCleaner);

    final boolean isSnapshotChoreEnabled = this.snapshotCleanupStateStore.get();
    this.snapshotCleanerChore = new SnapshotCleanerChore(this, conf, getSnapshotManager());
    if (isSnapshotChoreEnabled) {
      getChoreService().scheduleChore(this.snapshotCleanerChore);
    } else {
      if (LOG.isTraceEnabled()) {
        LOG.trace("Snapshot Cleaner Chore is disabled. Not starting up the chore..");
      }
    }
    serviceStarted = true;
    if (LOG.isTraceEnabled()) {
      LOG.trace("Started service threads");
    }
  }

  protected void stopServiceThreads() {
    if (masterJettyServer != null) {
      LOG.info("Stopping master jetty server");
      try {
        masterJettyServer.stop();
      } catch (Exception e) {
        LOG.error("Failed to stop master jetty server", e);
      }
    }
    stopChoreService();
    stopExecutorService();
    if (exclusiveHFileCleanerPool != null) {
      exclusiveHFileCleanerPool.shutdownNow();
      exclusiveHFileCleanerPool = null;
    }
    if (logCleanerPool != null) {
      logCleanerPool.shutdownNow();
      logCleanerPool = null;
    }
    if (sharedHFileCleanerPool != null) {
      sharedHFileCleanerPool.shutdownNow();
      sharedHFileCleanerPool = null;
    }
    if (maintenanceRegionServer != null) {
      maintenanceRegionServer.getRegionServer().stop(HBASE_MASTER_CLEANER_INTERVAL);
    }

    LOG.debug("Stopping service threads");
    // stop procedure executor prior to other services such as server manager and assignment
    // manager, as these services are important for some running procedures. See HBASE-24117 for
    // example.
    stopProcedureExecutor();

    if (regionNormalizerManager != null) {
      regionNormalizerManager.stop();
    }
    if (this.quotaManager != null) {
      this.quotaManager.stop();
    }

    if (this.activeMasterManager != null) {
      this.activeMasterManager.stop();
    }
    if (this.serverManager != null) {
      this.serverManager.stop();
    }
    if (this.assignmentManager != null) {
      this.assignmentManager.stop();
    }

    if (masterRegion != null) {
      masterRegion.close(isAborted());
    }
    if (this.walManager != null) {
      this.walManager.stop();
    }
    if (this.fileSystemManager != null) {
      this.fileSystemManager.stop();
    }
    if (this.mpmHost != null) {
      this.mpmHost.stop("server shutting down.");
    }
    if (this.regionServerTracker != null) {
      this.regionServerTracker.stop();
    }
  }

  private void createProcedureExecutor() throws IOException {
    MasterProcedureEnv procEnv = new MasterProcedureEnv(this);
    procedureStore = new RegionProcedureStore(this, masterRegion,
      new MasterProcedureEnv.FsUtilsLeaseRecovery(this));
    procedureStore.registerListener(new ProcedureStoreListener() {

      @Override
      public void abortProcess() {
        abort("The Procedure Store lost the lease", null);
      }
    });
    MasterProcedureScheduler procedureScheduler = procEnv.getProcedureScheduler();
    procedureExecutor = new ProcedureExecutor<>(conf, procEnv, procedureStore, procedureScheduler);
    configurationManager.registerObserver(procEnv);

    int cpus = Runtime.getRuntime().availableProcessors();
    final int numThreads = conf.getInt(MasterProcedureConstants.MASTER_PROCEDURE_THREADS, Math.max(
      (cpus > 0 ? cpus / 4 : 0), MasterProcedureConstants.DEFAULT_MIN_MASTER_PROCEDURE_THREADS));
    final boolean abortOnCorruption =
      conf.getBoolean(MasterProcedureConstants.EXECUTOR_ABORT_ON_CORRUPTION,
        MasterProcedureConstants.DEFAULT_EXECUTOR_ABORT_ON_CORRUPTION);
    procedureStore.start(numThreads);
    // Just initialize it but do not start the workers, we will start the workers later by calling
    // startProcedureExecutor. See the javadoc for finishActiveMasterInitialization for more
    // details.
    procedureExecutor.init(numThreads, abortOnCorruption);
    if (!procEnv.getRemoteDispatcher().start()) {
      throw new HBaseIOException("Failed start of remote dispatcher");
    }
  }

  // will be override in UT
  protected void startProcedureExecutor() throws IOException {
    procedureExecutor.startWorkers();
  }

  /**
   * Turn on/off Snapshot Cleanup Chore
   * @param on indicates whether Snapshot Cleanup Chore is to be run
   */
  void switchSnapshotCleanup(final boolean on, final boolean synchronous) throws IOException {
    if (synchronous) {
      synchronized (this.snapshotCleanerChore) {
        switchSnapshotCleanup(on);
      }
    } else {
      switchSnapshotCleanup(on);
    }
  }

  private void switchSnapshotCleanup(final boolean on) throws IOException {
    snapshotCleanupStateStore.set(on);
    if (on) {
      getChoreService().scheduleChore(this.snapshotCleanerChore);
    } else {
      this.snapshotCleanerChore.cancel();
    }
  }

  private void stopProcedureExecutor() {
    if (procedureExecutor != null) {
      configurationManager.deregisterObserver(procedureExecutor.getEnvironment());
      procedureExecutor.getEnvironment().getRemoteDispatcher().stop();
      procedureExecutor.stop();
      procedureExecutor.join();
      procedureExecutor = null;
    }

    if (procedureStore != null) {
      procedureStore.stop(isAborted());
      procedureStore = null;
    }
  }

  protected void stopChores() {
    shutdownChore(mobFileCleanerChore);
    shutdownChore(mobFileCompactionChore);
    shutdownChore(balancerChore);
    if (regionNormalizerManager != null) {
      shutdownChore(regionNormalizerManager.getRegionNormalizerChore());
    }
    shutdownChore(clusterStatusChore);
    shutdownChore(catalogJanitorChore);
    shutdownChore(clusterStatusPublisherChore);
    shutdownChore(snapshotQuotaChore);
    shutdownChore(logCleaner);
    if (hfileCleaners != null) {
      for (ScheduledChore chore : hfileCleaners) {
        chore.shutdown();
      }
      hfileCleaners = null;
    }
    shutdownChore(replicationBarrierCleaner);
    shutdownChore(snapshotCleanerChore);
    shutdownChore(hbckChore);
    shutdownChore(regionsRecoveryChore);
    shutdownChore(rollingUpgradeChore);
    shutdownChore(oldWALsDirSizeChore);
  }

  /** Returns Get remote side's InetAddress */
  InetAddress getRemoteInetAddress(final int port, final long serverStartCode)
    throws UnknownHostException {
    // Do it out here in its own little method so can fake an address when
    // mocking up in tests.
    InetAddress ia = RpcServer.getRemoteIp();

    // The call could be from the local regionserver,
    // in which case, there is no remote address.
    if (ia == null && serverStartCode == startcode) {
      InetSocketAddress isa = rpcServices.getSocketAddress();
      if (isa != null && isa.getPort() == port) {
        ia = isa.getAddress();
      }
    }
    return ia;
  }

  /** Returns Maximum time we should run balancer for */
  private int getMaxBalancingTime() {
    // if max balancing time isn't set, defaulting it to period time
    int maxBalancingTime =
      getConfiguration().getInt(HConstants.HBASE_BALANCER_MAX_BALANCING, getConfiguration()
        .getInt(HConstants.HBASE_BALANCER_PERIOD, HConstants.DEFAULT_HBASE_BALANCER_PERIOD));
    return maxBalancingTime;
  }

  /** Returns Maximum number of regions in transition */
  private int getMaxRegionsInTransition() {
    int numRegions = this.assignmentManager.getRegionStates().getRegionAssignments().size();
    return Math.max((int) Math.floor(numRegions * this.maxRitPercent), 1);
  }

  /**
   * It first sleep to the next balance plan start time. Meanwhile, throttling by the max number
   * regions in transition to protect availability.
   * @param nextBalanceStartTime   The next balance plan start time
   * @param maxRegionsInTransition max number of regions in transition
   * @param cutoffTime             when to exit balancer
   */
  private void balanceThrottling(long nextBalanceStartTime, int maxRegionsInTransition,
    long cutoffTime) {
    boolean interrupted = false;

    // Sleep to next balance plan start time
    // But if there are zero regions in transition, it can skip sleep to speed up.
    while (
      !interrupted && EnvironmentEdgeManager.currentTime() < nextBalanceStartTime
        && this.assignmentManager.getRegionStates().hasRegionsInTransition()
    ) {
      try {
        Thread.sleep(100);
      } catch (InterruptedException ie) {
        interrupted = true;
      }
    }

    // Throttling by max number regions in transition
    while (
      !interrupted && maxRegionsInTransition > 0
        && this.assignmentManager.getRegionStates().getRegionsInTransitionCount()
            >= maxRegionsInTransition
        && EnvironmentEdgeManager.currentTime() <= cutoffTime
    ) {
      try {
        // sleep if the number of regions in transition exceeds the limit
        Thread.sleep(100);
      } catch (InterruptedException ie) {
        interrupted = true;
      }
    }

    if (interrupted) Thread.currentThread().interrupt();
  }

  public BalanceResponse balance() throws IOException {
    return balance(BalanceRequest.defaultInstance());
  }

  /**
   * Trigger a normal balance, see {@link HMaster#balance()} . If the balance is not executed this
   * time, the metrics related to the balance will be updated. When balance is running, related
   * metrics will be updated at the same time. But if some checking logic failed and cause the
   * balancer exit early, we lost the chance to update balancer metrics. This will lead to user
   * missing the latest balancer info.
   */
  public BalanceResponse balanceOrUpdateMetrics() throws IOException {
    synchronized (this.balancer) {
      BalanceResponse response = balance();
      if (!response.isBalancerRan()) {
        Map<TableName, Map<ServerName, List<RegionInfo>>> assignments =
          this.assignmentManager.getRegionStates().getAssignmentsForBalancer(this.tableStateManager,
            this.serverManager.getOnlineServersList());
        for (Map<ServerName, List<RegionInfo>> serverMap : assignments.values()) {
          serverMap.keySet().removeAll(this.serverManager.getDrainingServersList());
        }
        this.balancer.updateBalancerLoadInfo(assignments);
      }
      return response;
    }
  }

  /**
   * Checks master state before initiating action over region topology.
   * @param action the name of the action under consideration, for logging.
   * @return {@code true} when the caller should exit early, {@code false} otherwise.
   */
  @Override
  public boolean skipRegionManagementAction(final String action) {
    // Note: this method could be `default` on MasterServices if but for logging.
    if (!isInitialized()) {
      LOG.debug("Master has not been initialized, don't run {}.", action);
      return true;
    }
    if (this.getServerManager().isClusterShutdown()) {
      LOG.info("Cluster is shutting down, don't run {}.", action);
      return true;
    }
    if (isInMaintenanceMode()) {
      LOG.info("Master is in maintenance mode, don't run {}.", action);
      return true;
    }
    return false;
  }

  public BalanceResponse balance(BalanceRequest request) throws IOException {
    checkInitialized();

    BalanceResponse.Builder responseBuilder = BalanceResponse.newBuilder();

    if (loadBalancerStateStore == null || !(loadBalancerStateStore.get() || request.isDryRun())) {
      return responseBuilder.build();
    }

    if (skipRegionManagementAction("balancer")) {
      return responseBuilder.build();
    }

    synchronized (this.balancer) {
      // Only allow one balance run at at time.
      if (this.assignmentManager.hasRegionsInTransition()) {
        List<RegionStateNode> regionsInTransition = assignmentManager.getRegionsInTransition();
        // if hbase:meta region is in transition, result of assignment cannot be recorded
        // ignore the force flag in that case
        boolean metaInTransition = assignmentManager.isMetaRegionInTransition();
        List<RegionStateNode> toPrint = regionsInTransition;
        int max = 5;
        boolean truncated = false;
        if (regionsInTransition.size() > max) {
          toPrint = regionsInTransition.subList(0, max);
          truncated = true;
        }

        if (!request.isIgnoreRegionsInTransition() || metaInTransition) {
          LOG.info("Not running balancer (ignoreRIT=false" + ", metaRIT=" + metaInTransition
            + ") because " + regionsInTransition.size() + " region(s) in transition: " + toPrint
            + (truncated ? "(truncated list)" : ""));
          return responseBuilder.build();
        }
      }
      if (this.serverManager.areDeadServersInProgress()) {
        LOG.info("Not running balancer because processing dead regionserver(s): "
          + this.serverManager.getDeadServers());
        return responseBuilder.build();
      }

      if (this.cpHost != null) {
        try {
          if (this.cpHost.preBalance(request)) {
            LOG.debug("Coprocessor bypassing balancer request");
            return responseBuilder.build();
          }
        } catch (IOException ioe) {
          LOG.error("Error invoking master coprocessor preBalance()", ioe);
          return responseBuilder.build();
        }
      }

      Map<TableName, Map<ServerName, List<RegionInfo>>> assignments =
        this.assignmentManager.getRegionStates().getAssignmentsForBalancer(tableStateManager,
          this.serverManager.getOnlineServersList());
      for (Map<ServerName, List<RegionInfo>> serverMap : assignments.values()) {
        serverMap.keySet().removeAll(this.serverManager.getDrainingServersList());
      }

      // Give the balancer the current cluster state.
      this.balancer.updateClusterMetrics(getClusterMetricsWithoutCoprocessor());

      List<RegionPlan> plans = this.balancer.balanceCluster(assignments);

      responseBuilder.setBalancerRan(true).setMovesCalculated(plans == null ? 0 : plans.size());

      if (skipRegionManagementAction("balancer")) {
        // make one last check that the cluster isn't shutting down before proceeding.
        return responseBuilder.build();
      }

      // For dry run we don't actually want to execute the moves, but we do want
      // to execute the coprocessor below
      List<RegionPlan> sucRPs =
        request.isDryRun() ? Collections.emptyList() : executeRegionPlansWithThrottling(plans);

      if (this.cpHost != null) {
        try {
          this.cpHost.postBalance(request, sucRPs);
        } catch (IOException ioe) {
          // balancing already succeeded so don't change the result
          LOG.error("Error invoking master coprocessor postBalance()", ioe);
        }
      }

      responseBuilder.setMovesExecuted(sucRPs.size());
    }

    // If LoadBalancer did not generate any plans, it means the cluster is already balanced.
    // Return true indicating a success.
    return responseBuilder.build();
  }

  /**
   * Execute region plans with throttling
   * @param plans to execute
   * @return succeeded plans
   */
  public List<RegionPlan> executeRegionPlansWithThrottling(List<RegionPlan> plans) {
    List<RegionPlan> successRegionPlans = new ArrayList<>();
    int maxRegionsInTransition = getMaxRegionsInTransition();
    long balanceStartTime = EnvironmentEdgeManager.currentTime();
    long cutoffTime = balanceStartTime + this.maxBalancingTime;
    int rpCount = 0; // number of RegionPlans balanced so far
    if (plans != null && !plans.isEmpty()) {
      int balanceInterval = this.maxBalancingTime / plans.size();
      LOG.info(
        "Balancer plans size is " + plans.size() + ", the balance interval is " + balanceInterval
          + " ms, and the max number regions in transition is " + maxRegionsInTransition);

      for (RegionPlan plan : plans) {
        LOG.info("balance " + plan);
        // TODO: bulk assign
        try {
          this.assignmentManager.balance(plan);
        } catch (HBaseIOException hioe) {
          // should ignore failed plans here, avoiding the whole balance plans be aborted
          // later calls of balance() can fetch up the failed and skipped plans
          LOG.warn("Failed balance plan {}, skipping...", plan, hioe);
        }
        // rpCount records balance plans processed, does not care if a plan succeeds
        rpCount++;
        successRegionPlans.add(plan);

        if (this.maxBalancingTime > 0) {
          balanceThrottling(balanceStartTime + rpCount * balanceInterval, maxRegionsInTransition,
            cutoffTime);
        }

        // if performing next balance exceeds cutoff time, exit the loop
        if (
          this.maxBalancingTime > 0 && rpCount < plans.size()
            && EnvironmentEdgeManager.currentTime() > cutoffTime
        ) {
          // TODO: After balance, there should not be a cutoff time (keeping it as
          // a security net for now)
          LOG.debug(
            "No more balancing till next balance run; maxBalanceTime=" + this.maxBalancingTime);
          break;
        }
      }
    }
    LOG.debug("Balancer is going into sleep until next period in {}ms", getConfiguration()
      .getInt(HConstants.HBASE_BALANCER_PERIOD, HConstants.DEFAULT_HBASE_BALANCER_PERIOD));
    return successRegionPlans;
  }

  @Override
  public RegionNormalizerManager getRegionNormalizerManager() {
    return regionNormalizerManager;
  }

  @Override
  public boolean normalizeRegions(final NormalizeTableFilterParams ntfp,
    final boolean isHighPriority) throws IOException {
    if (regionNormalizerManager == null || !regionNormalizerManager.isNormalizerOn()) {
      LOG.debug("Region normalization is disabled, don't run region normalizer.");
      return false;
    }
    if (skipRegionManagementAction("region normalizer")) {
      return false;
    }
    if (assignmentManager.hasRegionsInTransition()) {
      return false;
    }

    final Set<TableName> matchingTables = getTableDescriptors(new LinkedList<>(),
      ntfp.getNamespace(), ntfp.getRegex(), ntfp.getTableNames(), false).stream()
        .map(TableDescriptor::getTableName).collect(Collectors.toSet());
    final Set<TableName> allEnabledTables =
      tableStateManager.getTablesInStates(TableState.State.ENABLED);
    final List<TableName> targetTables =
      new ArrayList<>(Sets.intersection(matchingTables, allEnabledTables));
    Collections.shuffle(targetTables);
    return regionNormalizerManager.normalizeRegions(targetTables, isHighPriority);
  }

  /** Returns Client info for use as prefix on an audit log string; who did an action */
  @Override
  public String getClientIdAuditPrefix() {
    return "Client=" + RpcServer.getRequestUserName().orElse(null) + "/"
      + RpcServer.getRemoteAddress().orElse(null);
  }

  /**
   * Switch for the background CatalogJanitor thread. Used for testing. The thread will continue to
   * run. It will just be a noop if disabled.
   * @param b If false, the catalog janitor won't do anything.
   */
  public void setCatalogJanitorEnabled(final boolean b) {
    this.catalogJanitorChore.setEnabled(b);
  }

  @Override
  public long mergeRegions(final RegionInfo[] regionsToMerge, final boolean forcible, final long ng,
    final long nonce) throws IOException {
    checkInitialized();

    if (!isSplitOrMergeEnabled(MasterSwitchType.MERGE)) {
      String regionsStr = Arrays.deepToString(regionsToMerge);
      LOG.warn("Merge switch is off! skip merge of " + regionsStr);
      throw new DoNotRetryIOException(
        "Merge of " + regionsStr + " failed because merge switch is off");
    }

    final String mergeRegionsStr = Arrays.stream(regionsToMerge).map(RegionInfo::getEncodedName)
      .collect(Collectors.joining(", "));
    return MasterProcedureUtil.submitProcedure(new NonceProcedureRunnable(this, ng, nonce) {
      @Override
      protected void run() throws IOException {
        getMaster().getMasterCoprocessorHost().preMergeRegions(regionsToMerge);
        String aid = getClientIdAuditPrefix();
        LOG.info("{} merge regions {}", aid, mergeRegionsStr);
        submitProcedure(new MergeTableRegionsProcedure(procedureExecutor.getEnvironment(),
          regionsToMerge, forcible));
        getMaster().getMasterCoprocessorHost().postMergeRegions(regionsToMerge);
      }

      @Override
      protected String getDescription() {
        return "MergeTableProcedure";
      }
    });
  }

  @Override
  public long splitRegion(final RegionInfo regionInfo, final byte[] splitRow, final long nonceGroup,
    final long nonce) throws IOException {
    checkInitialized();

    if (!isSplitOrMergeEnabled(MasterSwitchType.SPLIT)) {
      LOG.warn("Split switch is off! skip split of " + regionInfo);
      throw new DoNotRetryIOException(
        "Split region " + regionInfo.getRegionNameAsString() + " failed due to split switch off");
    }

    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
        @Override
        protected void run() throws IOException {
          getMaster().getMasterCoprocessorHost().preSplitRegion(regionInfo.getTable(), splitRow);
          LOG.info(getClientIdAuditPrefix() + " split " + regionInfo.getRegionNameAsString());

          // Execute the operation asynchronously
          submitProcedure(getAssignmentManager().createSplitProcedure(regionInfo, splitRow));
        }

        @Override
        protected String getDescription() {
          return "SplitTableProcedure";
        }
      });
  }

  private void warmUpRegion(ServerName server, RegionInfo region) {
    FutureUtils.addListener(asyncClusterConnection.getRegionServerAdmin(server)
      .warmupRegion(RequestConverter.buildWarmupRegionRequest(region)), (r, e) -> {
        if (e != null) {
          LOG.warn("Failed to warm up region {} on server {}", region, server, e);
        }
      });
  }

  // Public so can be accessed by tests. Blocks until move is done.
  // Replace with an async implementation from which you can get
  // a success/failure result.
  @InterfaceAudience.Private
  public void move(final byte[] encodedRegionName, byte[] destServerName) throws IOException {
    RegionState regionState =
      assignmentManager.getRegionStates().getRegionState(Bytes.toString(encodedRegionName));

    RegionInfo hri;
    if (regionState != null) {
      hri = regionState.getRegion();
    } else {
      throw new UnknownRegionException(Bytes.toStringBinary(encodedRegionName));
    }

    ServerName dest;
    List<ServerName> exclude = hri.getTable().isSystemTable()
      ? assignmentManager.getExcludedServersForSystemTable()
      : new ArrayList<>(1);
    if (
      destServerName != null && exclude.contains(ServerName.valueOf(Bytes.toString(destServerName)))
    ) {
      LOG.info(Bytes.toString(encodedRegionName) + " can not move to "
        + Bytes.toString(destServerName) + " because the server is in exclude list");
      destServerName = null;
    }
    if (destServerName == null || destServerName.length == 0) {
      LOG.info("Passed destination servername is null/empty so " + "choosing a server at random");
      exclude.add(regionState.getServerName());
      final List<ServerName> destServers = this.serverManager.createDestinationServersList(exclude);
      dest = balancer.randomAssignment(hri, destServers);
      if (dest == null) {
        LOG.debug("Unable to determine a plan to assign " + hri);
        return;
      }
    } else {
      ServerName candidate = ServerName.valueOf(Bytes.toString(destServerName));
      dest = balancer.randomAssignment(hri, Lists.newArrayList(candidate));
      if (dest == null) {
        LOG.debug("Unable to determine a plan to assign " + hri);
        return;
      }
      // TODO: deal with table on master for rs group.
      if (dest.equals(serverName)) {
        // To avoid unnecessary region moving later by balancer. Don't put user
        // regions on master.
        LOG.debug("Skipping move of region " + hri.getRegionNameAsString()
          + " to avoid unnecessary region moving later by load balancer,"
          + " because it should not be on master");
        return;
      }
    }

    if (dest.equals(regionState.getServerName())) {
      LOG.debug("Skipping move of region " + hri.getRegionNameAsString()
        + " because region already assigned to the same server " + dest + ".");
      return;
    }

    // Now we can do the move
    RegionPlan rp = new RegionPlan(hri, regionState.getServerName(), dest);
    assert rp.getDestination() != null : rp.toString() + " " + dest;

    try {
      checkInitialized();
      if (this.cpHost != null) {
        this.cpHost.preMove(hri, rp.getSource(), rp.getDestination());
      }

      TransitRegionStateProcedure proc =
        this.assignmentManager.createMoveRegionProcedure(rp.getRegionInfo(), rp.getDestination());
      if (conf.getBoolean(WARMUP_BEFORE_MOVE, DEFAULT_WARMUP_BEFORE_MOVE)) {
        // Warmup the region on the destination before initiating the move.
        // A region server could reject the close request because it either does not
        // have the specified region or the region is being split.
        LOG.info(getClientIdAuditPrefix() + " move " + rp + ", warming up region on "
          + rp.getDestination());
        warmUpRegion(rp.getDestination(), hri);
      }
      LOG.info(getClientIdAuditPrefix() + " move " + rp + ", running balancer");
      Future<byte[]> future = ProcedureSyncWait.submitProcedure(this.procedureExecutor, proc);
      try {
        // Is this going to work? Will we throw exception on error?
        // TODO: CompletableFuture rather than this stunted Future.
        future.get();
      } catch (InterruptedException | ExecutionException e) {
        throw new HBaseIOException(e);
      }
      if (this.cpHost != null) {
        this.cpHost.postMove(hri, rp.getSource(), rp.getDestination());
      }
    } catch (IOException ioe) {
      if (ioe instanceof HBaseIOException) {
        throw (HBaseIOException) ioe;
      }
      throw new HBaseIOException(ioe);
    }
  }

  @Override
  public long createTable(final TableDescriptor tableDescriptor, final byte[][] splitKeys,
    final long nonceGroup, final long nonce) throws IOException {
    checkInitialized();
    TableDescriptor desc = getMasterCoprocessorHost().preCreateTableRegionsInfos(tableDescriptor);
    if (desc == null) {
      throw new IOException("Creation for " + tableDescriptor + " is canceled by CP");
    }
    String namespace = desc.getTableName().getNamespaceAsString();
    this.clusterSchemaService.getNamespace(namespace);

    RegionInfo[] newRegions = ModifyRegionUtils.createRegionInfos(desc, splitKeys);
    TableDescriptorChecker.sanityCheck(conf, desc);

    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
        @Override
        protected void run() throws IOException {
          getMaster().getMasterCoprocessorHost().preCreateTable(desc, newRegions);

          LOG.info(getClientIdAuditPrefix() + " create " + desc);

          // TODO: We can handle/merge duplicate requests, and differentiate the case of
          // TableExistsException by saying if the schema is the same or not.
          //
          // We need to wait for the procedure to potentially fail due to "prepare" sanity
          // checks. This will block only the beginning of the procedure. See HBASE-19953.
          ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch();
          submitProcedure(
            new CreateTableProcedure(procedureExecutor.getEnvironment(), desc, newRegions, latch));
          latch.await();

          getMaster().getMasterCoprocessorHost().postCreateTable(desc, newRegions);
        }

        @Override
        protected String getDescription() {
          return "CreateTableProcedure";
        }
      });
  }

  @Override
  public long createSystemTable(final TableDescriptor tableDescriptor) throws IOException {
    if (isStopped()) {
      throw new MasterNotRunningException();
    }

    TableName tableName = tableDescriptor.getTableName();
    if (!(tableName.isSystemTable())) {
      throw new IllegalArgumentException(
        "Only system table creation can use this createSystemTable API");
    }

    RegionInfo[] newRegions = ModifyRegionUtils.createRegionInfos(tableDescriptor, null);

    LOG.info(getClientIdAuditPrefix() + " create " + tableDescriptor);

    // This special create table is called locally to master. Therefore, no RPC means no need
    // to use nonce to detect duplicated RPC call.
    long procId = this.procedureExecutor.submitProcedure(
      new CreateTableProcedure(procedureExecutor.getEnvironment(), tableDescriptor, newRegions));

    return procId;
  }

  private void startActiveMasterManager(int infoPort) throws KeeperException {
    String backupZNode = ZNodePaths.joinZNode(zooKeeper.getZNodePaths().backupMasterAddressesZNode,
      serverName.toString());
    /*
     * Add a ZNode for ourselves in the backup master directory since we may not become the active
     * master. If so, we want the actual active master to know we are backup masters, so that it
     * won't assign regions to us if so configured. If we become the active master later,
     * ActiveMasterManager will delete this node explicitly. If we crash before then, ZooKeeper will
     * delete this node for us since it is ephemeral.
     */
    LOG.info("Adding backup master ZNode " + backupZNode);
    if (!MasterAddressTracker.setMasterAddress(zooKeeper, backupZNode, serverName, infoPort)) {
      LOG.warn("Failed create of " + backupZNode + " by " + serverName);
    }
    this.activeMasterManager.setInfoPort(infoPort);
    int timeout = conf.getInt(HConstants.ZK_SESSION_TIMEOUT, HConstants.DEFAULT_ZK_SESSION_TIMEOUT);
    // If we're a backup master, stall until a primary to write this address
    if (conf.getBoolean(HConstants.MASTER_TYPE_BACKUP, HConstants.DEFAULT_MASTER_TYPE_BACKUP)) {
      LOG.debug("HMaster started in backup mode. Stalling until master znode is written.");
      // This will only be a minute or so while the cluster starts up,
      // so don't worry about setting watches on the parent znode
      while (!activeMasterManager.hasActiveMaster()) {
        LOG.debug("Waiting for master address and cluster state znode to be written.");
        Threads.sleep(timeout);
      }
    }

    // Here for the master startup process, we use TaskGroup to monitor the whole progress.
    // The UI is similar to how Hadoop designed the startup page for the NameNode.
    // See HBASE-21521 for more details.
    // We do not cleanup the startupTaskGroup, let the startup progress information
    // be permanent in the MEM.
    startupTaskGroup = TaskMonitor.createTaskGroup(true, "Master startup");
    try {
      if (activeMasterManager.blockUntilBecomingActiveMaster(timeout, startupTaskGroup)) {
        finishActiveMasterInitialization();
      }
    } catch (Throwable t) {
      startupTaskGroup.abort("Failed to become active master due to:" + t.getMessage());
      LOG.error(HBaseMarkers.FATAL, "Failed to become active master", t);
      // HBASE-5680: Likely hadoop23 vs hadoop 20.x/1.x incompatibility
      if (
        t instanceof NoClassDefFoundError
          && t.getMessage().contains("org/apache/hadoop/hdfs/protocol/HdfsConstants$SafeModeAction")
      ) {
        // improved error message for this special case
        abort("HBase is having a problem with its Hadoop jars.  You may need to recompile "
          + "HBase against Hadoop version " + org.apache.hadoop.util.VersionInfo.getVersion()
          + " or change your hadoop jars to start properly", t);
      } else {
        abort("Unhandled exception. Starting shutdown.", t);
      }
    }
  }

  private static boolean isCatalogTable(final TableName tableName) {
    return tableName.equals(TableName.META_TABLE_NAME);
  }

  @Override
  public long deleteTable(final TableName tableName, final long nonceGroup, final long nonce)
    throws IOException {
    checkInitialized();

    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
        @Override
        protected void run() throws IOException {
          getMaster().getMasterCoprocessorHost().preDeleteTable(tableName);

          LOG.info(getClientIdAuditPrefix() + " delete " + tableName);

          // TODO: We can handle/merge duplicate request
          //
          // We need to wait for the procedure to potentially fail due to "prepare" sanity
          // checks. This will block only the beginning of the procedure. See HBASE-19953.
          ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch();
          submitProcedure(
            new DeleteTableProcedure(procedureExecutor.getEnvironment(), tableName, latch));
          latch.await();

          getMaster().getMasterCoprocessorHost().postDeleteTable(tableName);
        }

        @Override
        protected String getDescription() {
          return "DeleteTableProcedure";
        }
      });
  }

  @Override
  public long truncateTable(final TableName tableName, final boolean preserveSplits,
    final long nonceGroup, final long nonce) throws IOException {
    checkInitialized();

    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
        @Override
        protected void run() throws IOException {
          getMaster().getMasterCoprocessorHost().preTruncateTable(tableName);

          LOG.info(getClientIdAuditPrefix() + " truncate " + tableName);
          ProcedurePrepareLatch latch = ProcedurePrepareLatch.createLatch(2, 0);
          submitProcedure(new TruncateTableProcedure(procedureExecutor.getEnvironment(), tableName,
            preserveSplits, latch));
          latch.await();

          getMaster().getMasterCoprocessorHost().postTruncateTable(tableName);
        }

        @Override
        protected String getDescription() {
          return "TruncateTableProcedure";
        }
      });
  }

  @Override
  public long truncateRegion(final RegionInfo regionInfo, final long nonceGroup, final long nonce)
    throws IOException {
    checkInitialized();

    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
        @Override
        protected void run() throws IOException {
          getMaster().getMasterCoprocessorHost().preTruncateRegion(regionInfo);

          LOG.info(
            getClientIdAuditPrefix() + " truncate region " + regionInfo.getRegionNameAsString());

          // Execute the operation asynchronously
          ProcedurePrepareLatch latch = ProcedurePrepareLatch.createLatch(2, 0);
          submitProcedure(
            new TruncateRegionProcedure(procedureExecutor.getEnvironment(), regionInfo, latch));
          latch.await();

          getMaster().getMasterCoprocessorHost().postTruncateRegion(regionInfo);
        }

        @Override
        protected String getDescription() {
          return "TruncateRegionProcedure";
        }
      });
  }

  @Override
  public long addColumn(final TableName tableName, final ColumnFamilyDescriptor column,
    final long nonceGroup, final long nonce) throws IOException {
    checkInitialized();
    checkTableExists(tableName);

    return modifyTable(tableName, new TableDescriptorGetter() {

      @Override
      public TableDescriptor get() throws IOException {
        TableDescriptor old = getTableDescriptors().get(tableName);
        if (old.hasColumnFamily(column.getName())) {
          throw new InvalidFamilyOperationException("Column family '" + column.getNameAsString()
            + "' in table '" + tableName + "' already exists so cannot be added");
        }

        return TableDescriptorBuilder.newBuilder(old).setColumnFamily(column).build();
      }
    }, nonceGroup, nonce, true);
  }

  /**
   * Implement to return TableDescriptor after pre-checks
   */
  protected interface TableDescriptorGetter {
    TableDescriptor get() throws IOException;
  }

  @Override
  public long modifyColumn(final TableName tableName, final ColumnFamilyDescriptor descriptor,
    final long nonceGroup, final long nonce) throws IOException {
    checkInitialized();
    checkTableExists(tableName);
    return modifyTable(tableName, new TableDescriptorGetter() {

      @Override
      public TableDescriptor get() throws IOException {
        TableDescriptor old = getTableDescriptors().get(tableName);
        if (!old.hasColumnFamily(descriptor.getName())) {
          throw new InvalidFamilyOperationException("Family '" + descriptor.getNameAsString()
            + "' does not exist, so it cannot be modified");
        }

        return TableDescriptorBuilder.newBuilder(old).modifyColumnFamily(descriptor).build();
      }
    }, nonceGroup, nonce, true);
  }

  @Override
  public long modifyColumnStoreFileTracker(TableName tableName, byte[] family, String dstSFT,
    long nonceGroup, long nonce) throws IOException {
    checkInitialized();
    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {

        @Override
        protected void run() throws IOException {
          String sft = getMaster().getMasterCoprocessorHost()
            .preModifyColumnFamilyStoreFileTracker(tableName, family, dstSFT);
          LOG.info("{} modify column {} store file tracker of table {} to {}",
            getClientIdAuditPrefix(), Bytes.toStringBinary(family), tableName, sft);
          submitProcedure(new ModifyColumnFamilyStoreFileTrackerProcedure(
            procedureExecutor.getEnvironment(), tableName, family, sft));
          getMaster().getMasterCoprocessorHost().postModifyColumnFamilyStoreFileTracker(tableName,
            family, dstSFT);
        }

        @Override
        protected String getDescription() {
          return "ModifyColumnFamilyStoreFileTrackerProcedure";
        }
      });
  }

  @Override
  public long deleteColumn(final TableName tableName, final byte[] columnName,
    final long nonceGroup, final long nonce) throws IOException {
    checkInitialized();
    checkTableExists(tableName);

    return modifyTable(tableName, new TableDescriptorGetter() {

      @Override
      public TableDescriptor get() throws IOException {
        TableDescriptor old = getTableDescriptors().get(tableName);

        if (!old.hasColumnFamily(columnName)) {
          throw new InvalidFamilyOperationException(
            "Family '" + Bytes.toString(columnName) + "' does not exist, so it cannot be deleted");
        }
        if (old.getColumnFamilyCount() == 1) {
          throw new InvalidFamilyOperationException("Family '" + Bytes.toString(columnName)
            + "' is the only column family in the table, so it cannot be deleted");
        }
        return TableDescriptorBuilder.newBuilder(old).removeColumnFamily(columnName).build();
      }
    }, nonceGroup, nonce, true);
  }

  @Override
  public long enableTable(final TableName tableName, final long nonceGroup, final long nonce)
    throws IOException {
    checkInitialized();

    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
        @Override
        protected void run() throws IOException {
          getMaster().getMasterCoprocessorHost().preEnableTable(tableName);

          // Normally, it would make sense for this authorization check to exist inside
          // AccessController, but because the authorization check is done based on internal state
          // (rather than explicit permissions) we'll do the check here instead of in the
          // coprocessor.
          MasterQuotaManager quotaManager = getMasterQuotaManager();
          if (quotaManager != null) {
            if (quotaManager.isQuotaInitialized()) {
              // skip checking quotas for system tables, see:
              // https://issues.apache.org/jira/browse/HBASE-28183
              if (!tableName.isSystemTable()) {
                SpaceQuotaSnapshot currSnapshotOfTable =
                  QuotaTableUtil.getCurrentSnapshotFromQuotaTable(getConnection(), tableName);
                if (currSnapshotOfTable != null) {
                  SpaceQuotaStatus quotaStatus = currSnapshotOfTable.getQuotaStatus();
                  if (
                    quotaStatus.isInViolation()
                      && SpaceViolationPolicy.DISABLE == quotaStatus.getPolicy().orElse(null)
                  ) {
                    throw new AccessDeniedException("Enabling the table '" + tableName
                      + "' is disallowed due to a violated space quota.");
                  }
                }
              }
            } else if (LOG.isTraceEnabled()) {
              LOG
                .trace("Unable to check for space quotas as the MasterQuotaManager is not enabled");
            }
          }

          LOG.info(getClientIdAuditPrefix() + " enable " + tableName);

          // Execute the operation asynchronously - client will check the progress of the operation
          // In case the request is from a <1.1 client before returning,
          // we want to make sure that the table is prepared to be
          // enabled (the table is locked and the table state is set).
          // Note: if the procedure throws exception, we will catch it and rethrow.
          final ProcedurePrepareLatch prepareLatch = ProcedurePrepareLatch.createLatch();
          submitProcedure(
            new EnableTableProcedure(procedureExecutor.getEnvironment(), tableName, prepareLatch));
          prepareLatch.await();

          getMaster().getMasterCoprocessorHost().postEnableTable(tableName);
        }

        @Override
        protected String getDescription() {
          return "EnableTableProcedure";
        }
      });
  }

  @Override
  public long disableTable(final TableName tableName, final long nonceGroup, final long nonce)
    throws IOException {
    checkInitialized();

    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
        @Override
        protected void run() throws IOException {
          getMaster().getMasterCoprocessorHost().preDisableTable(tableName);

          LOG.info(getClientIdAuditPrefix() + " disable " + tableName);

          // Execute the operation asynchronously - client will check the progress of the operation
          // In case the request is from a <1.1 client before returning,
          // we want to make sure that the table is prepared to be
          // enabled (the table is locked and the table state is set).
          // Note: if the procedure throws exception, we will catch it and rethrow.
          //
          // We need to wait for the procedure to potentially fail due to "prepare" sanity
          // checks. This will block only the beginning of the procedure. See HBASE-19953.
          final ProcedurePrepareLatch prepareLatch = ProcedurePrepareLatch.createBlockingLatch();
          submitProcedure(new DisableTableProcedure(procedureExecutor.getEnvironment(), tableName,
            false, prepareLatch));
          prepareLatch.await();

          getMaster().getMasterCoprocessorHost().postDisableTable(tableName);
        }

        @Override
        protected String getDescription() {
          return "DisableTableProcedure";
        }
      });
  }

  private long modifyTable(final TableName tableName,
    final TableDescriptorGetter newDescriptorGetter, final long nonceGroup, final long nonce,
    final boolean shouldCheckDescriptor) throws IOException {
    return modifyTable(tableName, newDescriptorGetter, nonceGroup, nonce, shouldCheckDescriptor,
      true);
  }

  private long modifyTable(final TableName tableName,
    final TableDescriptorGetter newDescriptorGetter, final long nonceGroup, final long nonce,
    final boolean shouldCheckDescriptor, final boolean reopenRegions) throws IOException {
    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
        @Override
        protected void run() throws IOException {
          TableDescriptor oldDescriptor = getMaster().getTableDescriptors().get(tableName);
          TableDescriptor newDescriptor = getMaster().getMasterCoprocessorHost()
            .preModifyTable(tableName, oldDescriptor, newDescriptorGetter.get());
          TableDescriptorChecker.sanityCheck(conf, newDescriptor);
          LOG.info("{} modify table {} from {} to {}", getClientIdAuditPrefix(), tableName,
            oldDescriptor, newDescriptor);

          // Execute the operation synchronously - wait for the operation completes before
          // continuing.
          //
          // We need to wait for the procedure to potentially fail due to "prepare" sanity
          // checks. This will block only the beginning of the procedure. See HBASE-19953.
          ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch();
          submitProcedure(new ModifyTableProcedure(procedureExecutor.getEnvironment(),
            newDescriptor, latch, oldDescriptor, shouldCheckDescriptor, reopenRegions));
          latch.await();

          getMaster().getMasterCoprocessorHost().postModifyTable(tableName, oldDescriptor,
            newDescriptor);
        }

        @Override
        protected String getDescription() {
          return "ModifyTableProcedure";
        }
      });

  }

  @Override
  public long modifyTable(final TableName tableName, final TableDescriptor newDescriptor,
    final long nonceGroup, final long nonce, final boolean reopenRegions) throws IOException {
    checkInitialized();
    return modifyTable(tableName, new TableDescriptorGetter() {
      @Override
      public TableDescriptor get() throws IOException {
        return newDescriptor;
      }
    }, nonceGroup, nonce, false, reopenRegions);

  }

  @Override
  public long modifyTableStoreFileTracker(TableName tableName, String dstSFT, long nonceGroup,
    long nonce) throws IOException {
    checkInitialized();
    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {

        @Override
        protected void run() throws IOException {
          String sft = getMaster().getMasterCoprocessorHost()
            .preModifyTableStoreFileTracker(tableName, dstSFT);
          LOG.info("{} modify table store file tracker of table {} to {}", getClientIdAuditPrefix(),
            tableName, sft);
          submitProcedure(new ModifyTableStoreFileTrackerProcedure(
            procedureExecutor.getEnvironment(), tableName, sft));
          getMaster().getMasterCoprocessorHost().postModifyTableStoreFileTracker(tableName, sft);
        }

        @Override
        protected String getDescription() {
          return "ModifyTableStoreFileTrackerProcedure";
        }
      });
  }

  public long restoreSnapshot(final SnapshotDescription snapshotDesc, final long nonceGroup,
    final long nonce, final boolean restoreAcl, final String customSFT) throws IOException {
    checkInitialized();
    getSnapshotManager().checkSnapshotSupport();

    // Ensure namespace exists. Will throw exception if non-known NS.
    final TableName dstTable = TableName.valueOf(snapshotDesc.getTable());
    getClusterSchema().getNamespace(dstTable.getNamespaceAsString());

    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
        @Override
        protected void run() throws IOException {
          setProcId(getSnapshotManager().restoreOrCloneSnapshot(snapshotDesc, getNonceKey(),
            restoreAcl, customSFT));
        }

        @Override
        protected String getDescription() {
          return "RestoreSnapshotProcedure";
        }
      });
  }

  private void checkTableExists(final TableName tableName)
    throws IOException, TableNotFoundException {
    if (!tableDescriptors.exists(tableName)) {
      throw new TableNotFoundException(tableName);
    }
  }

  @Override
  public void checkTableModifiable(final TableName tableName)
    throws IOException, TableNotFoundException, TableNotDisabledException {
    if (isCatalogTable(tableName)) {
      throw new IOException("Can't modify catalog tables");
    }
    checkTableExists(tableName);
    TableState ts = getTableStateManager().getTableState(tableName);
    if (!ts.isDisabled()) {
      throw new TableNotDisabledException("Not DISABLED; " + ts);
    }
  }

  public ClusterMetrics getClusterMetricsWithoutCoprocessor() throws InterruptedIOException {
    return getClusterMetricsWithoutCoprocessor(EnumSet.allOf(Option.class));
  }

  public ClusterMetrics getClusterMetricsWithoutCoprocessor(EnumSet<Option> options)
    throws InterruptedIOException {
    ClusterMetricsBuilder builder = ClusterMetricsBuilder.newBuilder();
    // given that hbase1 can't submit the request with Option,
    // we return all information to client if the list of Option is empty.
    if (options.isEmpty()) {
      options = EnumSet.allOf(Option.class);
    }

    // TASKS and/or LIVE_SERVERS will populate this map, which will be given to the builder if
    // not null after option processing completes.
    Map<ServerName, ServerMetrics> serverMetricsMap = null;

    for (Option opt : options) {
      switch (opt) {
        case HBASE_VERSION:
          builder.setHBaseVersion(VersionInfo.getVersion());
          break;
        case CLUSTER_ID:
          builder.setClusterId(getClusterId());
          break;
        case MASTER:
          builder.setMasterName(getServerName());
          break;
        case BACKUP_MASTERS:
          builder.setBackerMasterNames(getBackupMasters());
          break;
        case TASKS: {
          // Master tasks
          builder.setMasterTasks(TaskMonitor.get().getTasks().stream()
            .map(task -> ServerTaskBuilder.newBuilder().setDescription(task.getDescription())
              .setStatus(task.getStatus())
              .setState(ServerTask.State.valueOf(task.getState().name()))
              .setStartTime(task.getStartTime()).setCompletionTime(task.getCompletionTimestamp())
              .build())
            .collect(Collectors.toList()));
          // TASKS is also synonymous with LIVE_SERVERS for now because task information for
          // regionservers is carried in ServerLoad.
          // Add entries to serverMetricsMap for all live servers, if we haven't already done so
          if (serverMetricsMap == null) {
            serverMetricsMap = getOnlineServers();
          }
          break;
        }
        case LIVE_SERVERS: {
          // Add entries to serverMetricsMap for all live servers, if we haven't already done so
          if (serverMetricsMap == null) {
            serverMetricsMap = getOnlineServers();
          }
          break;
        }
        case DEAD_SERVERS: {
          if (serverManager != null) {
            builder.setDeadServerNames(
              new ArrayList<>(serverManager.getDeadServers().copyServerNames()));
          }
          break;
        }
        case UNKNOWN_SERVERS: {
          if (serverManager != null) {
            builder.setUnknownServerNames(getUnknownServers());
          }
          break;
        }
        case MASTER_COPROCESSORS: {
          if (cpHost != null) {
            builder.setMasterCoprocessorNames(Arrays.asList(getMasterCoprocessors()));
          }
          break;
        }
        case REGIONS_IN_TRANSITION: {
          if (assignmentManager != null) {
            builder.setRegionsInTransition(
              assignmentManager.getRegionStates().getRegionsStateInTransition());
          }
          break;
        }
        case BALANCER_ON: {
          if (loadBalancerStateStore != null) {
            builder.setBalancerOn(loadBalancerStateStore.get());
          }
          break;
        }
        case MASTER_INFO_PORT: {
          if (infoServer != null) {
            builder.setMasterInfoPort(infoServer.getPort());
          }
          break;
        }
        case SERVERS_NAME: {
          if (serverManager != null) {
            builder.setServerNames(serverManager.getOnlineServersList());
          }
          break;
        }
        case TABLE_TO_REGIONS_COUNT: {
          if (isActiveMaster() && isInitialized() && assignmentManager != null) {
            try {
              Map<TableName, RegionStatesCount> tableRegionStatesCountMap = new HashMap<>();
              Map<String, TableDescriptor> tableDescriptorMap = getTableDescriptors().getAll();
              for (TableDescriptor tableDescriptor : tableDescriptorMap.values()) {
                TableName tableName = tableDescriptor.getTableName();
                RegionStatesCount regionStatesCount =
                  assignmentManager.getRegionStatesCount(tableName);
                tableRegionStatesCountMap.put(tableName, regionStatesCount);
              }
              builder.setTableRegionStatesCount(tableRegionStatesCountMap);
            } catch (IOException e) {
              LOG.error("Error while populating TABLE_TO_REGIONS_COUNT for Cluster Metrics..", e);
            }
          }
          break;
        }
        case DECOMMISSIONED_SERVERS: {
          if (serverManager != null) {
            builder.setDecommissionedServerNames(serverManager.getDrainingServersList());
          }
          break;
        }
      }
    }

    if (serverMetricsMap != null) {
      builder.setLiveServerMetrics(serverMetricsMap);
    }

    return builder.build();
  }

  private List<ServerName> getUnknownServers() {
    if (serverManager != null) {
      final Set<ServerName> serverNames = getAssignmentManager().getRegionStates().getRegionStates()
        .stream().map(RegionState::getServerName).collect(Collectors.toSet());
      final List<ServerName> unknownServerNames = serverNames.stream()
        .filter(sn -> sn != null && serverManager.isServerUnknown(sn)).collect(Collectors.toList());
      return unknownServerNames;
    }
    return null;
  }

  private Map<ServerName, ServerMetrics> getOnlineServers() {
    if (serverManager != null) {
      final Map<ServerName, ServerMetrics> map = new HashMap<>();
      serverManager.getOnlineServers().entrySet().forEach(e -> map.put(e.getKey(), e.getValue()));
      return map;
    }
    return null;
  }

  /** Returns cluster status */
  public ClusterMetrics getClusterMetrics() throws IOException {
    return getClusterMetrics(EnumSet.allOf(Option.class));
  }

  public ClusterMetrics getClusterMetrics(EnumSet<Option> options) throws IOException {
    if (cpHost != null) {
      cpHost.preGetClusterMetrics();
    }
    ClusterMetrics status = getClusterMetricsWithoutCoprocessor(options);
    if (cpHost != null) {
      cpHost.postGetClusterMetrics(status);
    }
    return status;
  }

  /** Returns info port of active master or 0 if any exception occurs. */
  public int getActiveMasterInfoPort() {
    return activeMasterManager.getActiveMasterInfoPort();
  }

  /**
   * @param sn is ServerName of the backup master
   * @return info port of backup master or 0 if any exception occurs.
   */
  public int getBackupMasterInfoPort(final ServerName sn) {
    return activeMasterManager.getBackupMasterInfoPort(sn);
  }

  /**
   * The set of loaded coprocessors is stored in a static set. Since it's statically allocated, it
   * does not require that HMaster's cpHost be initialized prior to accessing it.
   * @return a String representation of the set of names of the loaded coprocessors.
   */
  public static String getLoadedCoprocessors() {
    return CoprocessorHost.getLoadedCoprocessors().toString();
  }

  /** Returns timestamp in millis when HMaster was started. */
  public long getMasterStartTime() {
    return startcode;
  }

  /** Returns timestamp in millis when HMaster became the active master. */
  public long getMasterActiveTime() {
    return masterActiveTime;
  }

  /** Returns timestamp in millis when HMaster finished becoming the active master */
  public long getMasterFinishedInitializationTime() {
    return masterFinishedInitializationTime;
  }

  public int getNumWALFiles() {
    return 0;
  }

  public ProcedureStore getProcedureStore() {
    return procedureStore;
  }

  public int getRegionServerInfoPort(final ServerName sn) {
    int port = this.serverManager.getInfoPort(sn);
    return port == 0
      ? conf.getInt(HConstants.REGIONSERVER_INFO_PORT, HConstants.DEFAULT_REGIONSERVER_INFOPORT)
      : port;
  }

  @Override
  public String getRegionServerVersion(ServerName sn) {
    // Will return "0.0.0" if the server is not online to prevent move system region to unknown
    // version RS.
    return this.serverManager.getVersion(sn);
  }

  @Override
  public void checkIfShouldMoveSystemRegionAsync() {
    assignmentManager.checkIfShouldMoveSystemRegionAsync();
  }

  /** Returns array of coprocessor SimpleNames. */
  public String[] getMasterCoprocessors() {
    Set<String> masterCoprocessors = getMasterCoprocessorHost().getCoprocessors();
    return masterCoprocessors.toArray(new String[masterCoprocessors.size()]);
  }

  @Override
  public void abort(String reason, Throwable cause) {
    if (!setAbortRequested() || isStopped()) {
      LOG.debug("Abort called but aborted={}, stopped={}", isAborted(), isStopped());
      return;
    }
    if (cpHost != null) {
      // HBASE-4014: dump a list of loaded coprocessors.
      LOG.error(HBaseMarkers.FATAL,
        "Master server abort: loaded coprocessors are: " + getLoadedCoprocessors());
    }
    String msg = "***** ABORTING master " + this + ": " + reason + " *****";
    if (cause != null) {
      LOG.error(HBaseMarkers.FATAL, msg, cause);
    } else {
      LOG.error(HBaseMarkers.FATAL, msg);
    }

    try {
      stopMaster();
    } catch (IOException e) {
      LOG.error("Exception occurred while stopping master", e);
    }
  }

  @Override
  public MasterCoprocessorHost getMasterCoprocessorHost() {
    return cpHost;
  }

  @Override
  public MasterQuotaManager getMasterQuotaManager() {
    return quotaManager;
  }

  @Override
  public ProcedureExecutor<MasterProcedureEnv> getMasterProcedureExecutor() {
    return procedureExecutor;
  }

  @Override
  public ServerName getServerName() {
    return this.serverName;
  }

  @Override
  public AssignmentManager getAssignmentManager() {
    return this.assignmentManager;
  }

  @Override
  public CatalogJanitor getCatalogJanitor() {
    return this.catalogJanitorChore;
  }

  public MemoryBoundedLogMessageBuffer getRegionServerFatalLogBuffer() {
    return rsFatals;
  }

  public TaskGroup getStartupProgress() {
    return startupTaskGroup;
  }

  /**
   * Shutdown the cluster. Master runs a coordinated stop of all RegionServers and then itself.
   */
  public void shutdown() throws IOException {
    TraceUtil.trace(() -> {
      if (cpHost != null) {
        cpHost.preShutdown();
      }

      // Tell the servermanager cluster shutdown has been called. This makes it so when Master is
      // last running server, it'll stop itself. Next, we broadcast the cluster shutdown by setting
      // the cluster status as down. RegionServers will notice this change in state and will start
      // shutting themselves down. When last has exited, Master can go down.
      if (this.serverManager != null) {
        this.serverManager.shutdownCluster();
      }
      if (this.clusterStatusTracker != null) {
        try {
          this.clusterStatusTracker.setClusterDown();
        } catch (KeeperException e) {
          LOG.error("ZooKeeper exception trying to set cluster as down in ZK", e);
        }
      }
      // Stop the procedure executor. Will stop any ongoing assign, unassign, server crash etc.,
      // processing so we can go down.
      if (this.procedureExecutor != null) {
        this.procedureExecutor.stop();
      }
      // Shutdown our cluster connection. This will kill any hosted RPCs that might be going on;
      // this is what we want especially if the Master is in startup phase doing call outs to
      // hbase:meta, etc. when cluster is down. Without ths connection close, we'd have to wait on
      // the rpc to timeout.
      if (this.asyncClusterConnection != null) {
        this.asyncClusterConnection.close();
      }
    }, "HMaster.shutdown");
  }

  public void stopMaster() throws IOException {
    if (cpHost != null) {
      cpHost.preStopMaster();
    }
    stop("Stopped by " + Thread.currentThread().getName());
  }

  @Override
  public void stop(String msg) {
    if (!this.stopped) {
      LOG.info("***** STOPPING master '" + this + "' *****");
      this.stopped = true;
      LOG.info("STOPPED: " + msg);
      // Wakes run() if it is sleeping
      sleeper.skipSleepCycle();
      if (this.activeMasterManager != null) {
        this.activeMasterManager.stop();
      }
    }
  }

  protected void checkServiceStarted() throws ServerNotRunningYetException {
    if (!serviceStarted) {
      throw new ServerNotRunningYetException("Server is not running yet");
    }
  }

  void checkInitialized() throws PleaseHoldException, ServerNotRunningYetException,
    MasterNotRunningException, MasterStoppedException {
    checkServiceStarted();
    if (!isInitialized()) {
      throw new PleaseHoldException("Master is initializing");
    }
    if (isStopped()) {
      throw new MasterStoppedException();
    }
  }

  /**
   * Report whether this master is currently the active master or not. If not active master, we are
   * parked on ZK waiting to become active. This method is used for testing.
   * @return true if active master, false if not.
   */
  @Override
  public boolean isActiveMaster() {
    return activeMaster;
  }

  /**
   * Report whether this master has completed with its initialization and is ready. If ready, the
   * master is also the active master. A standby master is never ready. This method is used for
   * testing.
   * @return true if master is ready to go, false if not.
   */
  @Override
  public boolean isInitialized() {
    return initialized.isReady();
  }

  /**
   * Report whether this master is started This method is used for testing.
   * @return true if master is ready to go, false if not.
   */
  public boolean isOnline() {
    return serviceStarted;
  }

  /**
   * Report whether this master is in maintenance mode.
   * @return true if master is in maintenanceMode
   */
  @Override
  public boolean isInMaintenanceMode() {
    return maintenanceMode;
  }

  public void setInitialized(boolean isInitialized) {
    procedureExecutor.getEnvironment().setEventReady(initialized, isInitialized);
  }

  /**
   * Mainly used in procedure related tests, where we will restart ProcedureExecutor and
   * AssignmentManager, but we do not want to restart master(to speed up the test), so we need to
   * disable rpc for a while otherwise some critical rpc requests such as
   * reportRegionStateTransition could fail and cause region server to abort.
   */
  @RestrictedApi(explanation = "Should only be called in tests", link = "",
      allowedOnPath = ".*/src/test/.*")
  public void setServiceStarted(boolean started) {
    this.serviceStarted = started;
  }

  @Override
  public ProcedureEvent<?> getInitializedEvent() {
    return initialized;
  }

  /**
   * Compute the average load across all region servers. Currently, this uses a very naive
   * computation - just uses the number of regions being served, ignoring stats about number of
   * requests.
   * @return the average load
   */
  public double getAverageLoad() {
    if (this.assignmentManager == null) {
      return 0;
    }

    RegionStates regionStates = this.assignmentManager.getRegionStates();
    if (regionStates == null) {
      return 0;
    }
    return regionStates.getAverageLoad();
  }

  @Override
  public boolean registerService(Service instance) {
    /*
     * No stacking of instances is allowed for a single service name
     */
    Descriptors.ServiceDescriptor serviceDesc = instance.getDescriptorForType();
    String serviceName = CoprocessorRpcUtils.getServiceName(serviceDesc);
    if (coprocessorServiceHandlers.containsKey(serviceName)) {
      LOG.error("Coprocessor service " + serviceName
        + " already registered, rejecting request from " + instance);
      return false;
    }

    coprocessorServiceHandlers.put(serviceName, instance);
    if (LOG.isDebugEnabled()) {
      LOG.debug("Registered master coprocessor service: service=" + serviceName);
    }
    return true;
  }

  /**
   * Utility for constructing an instance of the passed HMaster class.
   * @return HMaster instance.
   */
  public static HMaster constructMaster(Class<? extends HMaster> masterClass,
    final Configuration conf) {
    try {
      Constructor<? extends HMaster> c = masterClass.getConstructor(Configuration.class);
      return c.newInstance(conf);
    } catch (Exception e) {
      Throwable error = e;
      if (
        e instanceof InvocationTargetException
          && ((InvocationTargetException) e).getTargetException() != null
      ) {
        error = ((InvocationTargetException) e).getTargetException();
      }
      throw new RuntimeException("Failed construction of Master: " + masterClass.toString() + ". ",
        error);
    }
  }

  /**
   * @see org.apache.hadoop.hbase.master.HMasterCommandLine
   */
  public static void main(String[] args) {
    LOG.info("STARTING service " + HMaster.class.getSimpleName());
    VersionInfo.logVersion();
    new HMasterCommandLine(HMaster.class).doMain(args);
  }

  public HFileCleaner getHFileCleaner() {
    return this.hfileCleaners.get(0);
  }

  public List<HFileCleaner> getHFileCleaners() {
    return this.hfileCleaners;
  }

  public LogCleaner getLogCleaner() {
    return this.logCleaner;
  }

  /** Returns the underlying snapshot manager */
  @Override
  public SnapshotManager getSnapshotManager() {
    return this.snapshotManager;
  }

  /** Returns the underlying MasterProcedureManagerHost */
  @Override
  public MasterProcedureManagerHost getMasterProcedureManagerHost() {
    return mpmHost;
  }

  @Override
  public ClusterSchema getClusterSchema() {
    return this.clusterSchemaService;
  }

  /**
   * Create a new Namespace.
   * @param namespaceDescriptor descriptor for new Namespace
   * @param nonceGroup          Identifier for the source of the request, a client or process.
   * @param nonce               A unique identifier for this operation from the client or process
   *                            identified by <code>nonceGroup</code> (the source must ensure each
   *                            operation gets a unique id).
   * @return procedure id
   */
  long createNamespace(final NamespaceDescriptor namespaceDescriptor, final long nonceGroup,
    final long nonce) throws IOException {
    checkInitialized();

    TableName.isLegalNamespaceName(Bytes.toBytes(namespaceDescriptor.getName()));

    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
        @Override
        protected void run() throws IOException {
          getMaster().getMasterCoprocessorHost().preCreateNamespace(namespaceDescriptor);
          // We need to wait for the procedure to potentially fail due to "prepare" sanity
          // checks. This will block only the beginning of the procedure. See HBASE-19953.
          ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch();
          LOG.info(getClientIdAuditPrefix() + " creating " + namespaceDescriptor);
          // Execute the operation synchronously - wait for the operation to complete before
          // continuing.
          setProcId(getClusterSchema().createNamespace(namespaceDescriptor, getNonceKey(), latch));
          latch.await();
          getMaster().getMasterCoprocessorHost().postCreateNamespace(namespaceDescriptor);
        }

        @Override
        protected String getDescription() {
          return "CreateNamespaceProcedure";
        }
      });
  }

  /**
   * Modify an existing Namespace.
   * @param nonceGroup Identifier for the source of the request, a client or process.
   * @param nonce      A unique identifier for this operation from the client or process identified
   *                   by <code>nonceGroup</code> (the source must ensure each operation gets a
   *                   unique id).
   * @return procedure id
   */
  long modifyNamespace(final NamespaceDescriptor newNsDescriptor, final long nonceGroup,
    final long nonce) throws IOException {
    checkInitialized();

    TableName.isLegalNamespaceName(Bytes.toBytes(newNsDescriptor.getName()));

    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
        @Override
        protected void run() throws IOException {
          NamespaceDescriptor oldNsDescriptor = getNamespace(newNsDescriptor.getName());
          getMaster().getMasterCoprocessorHost().preModifyNamespace(oldNsDescriptor,
            newNsDescriptor);
          // We need to wait for the procedure to potentially fail due to "prepare" sanity
          // checks. This will block only the beginning of the procedure. See HBASE-19953.
          ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch();
          LOG.info(getClientIdAuditPrefix() + " modify " + newNsDescriptor);
          // Execute the operation synchronously - wait for the operation to complete before
          // continuing.
          setProcId(getClusterSchema().modifyNamespace(newNsDescriptor, getNonceKey(), latch));
          latch.await();
          getMaster().getMasterCoprocessorHost().postModifyNamespace(oldNsDescriptor,
            newNsDescriptor);
        }

        @Override
        protected String getDescription() {
          return "ModifyNamespaceProcedure";
        }
      });
  }

  /**
   * Delete an existing Namespace. Only empty Namespaces (no tables) can be removed.
   * @param nonceGroup Identifier for the source of the request, a client or process.
   * @param nonce      A unique identifier for this operation from the client or process identified
   *                   by <code>nonceGroup</code> (the source must ensure each operation gets a
   *                   unique id).
   * @return procedure id
   */
  long deleteNamespace(final String name, final long nonceGroup, final long nonce)
    throws IOException {
    checkInitialized();

    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
        @Override
        protected void run() throws IOException {
          getMaster().getMasterCoprocessorHost().preDeleteNamespace(name);
          LOG.info(getClientIdAuditPrefix() + " delete " + name);
          // Execute the operation synchronously - wait for the operation to complete before
          // continuing.
          //
          // We need to wait for the procedure to potentially fail due to "prepare" sanity
          // checks. This will block only the beginning of the procedure. See HBASE-19953.
          ProcedurePrepareLatch latch = ProcedurePrepareLatch.createBlockingLatch();
          setProcId(submitProcedure(
            new DeleteNamespaceProcedure(procedureExecutor.getEnvironment(), name, latch)));
          latch.await();
          // Will not be invoked in the face of Exception thrown by the Procedure's execution
          getMaster().getMasterCoprocessorHost().postDeleteNamespace(name);
        }

        @Override
        protected String getDescription() {
          return "DeleteNamespaceProcedure";
        }
      });
  }

  /**
   * Get a Namespace
   * @param name Name of the Namespace
   * @return Namespace descriptor for <code>name</code>
   */
  NamespaceDescriptor getNamespace(String name) throws IOException {
    checkInitialized();
    if (this.cpHost != null) this.cpHost.preGetNamespaceDescriptor(name);
    NamespaceDescriptor nsd = this.clusterSchemaService.getNamespace(name);
    if (this.cpHost != null) this.cpHost.postGetNamespaceDescriptor(nsd);
    return nsd;
  }

  /**
   * Get all Namespaces
   * @return All Namespace descriptors
   */
  List<NamespaceDescriptor> getNamespaces() throws IOException {
    checkInitialized();
    final List<NamespaceDescriptor> nsds = new ArrayList<>();
    if (cpHost != null) {
      cpHost.preListNamespaceDescriptors(nsds);
    }
    nsds.addAll(this.clusterSchemaService.getNamespaces());
    if (this.cpHost != null) {
      this.cpHost.postListNamespaceDescriptors(nsds);
    }
    return nsds;
  }

  /**
   * List namespace names
   * @return All namespace names
   */
  public List<String> listNamespaces() throws IOException {
    checkInitialized();
    List<String> namespaces = new ArrayList<>();
    if (cpHost != null) {
      cpHost.preListNamespaces(namespaces);
    }
    for (NamespaceDescriptor namespace : clusterSchemaService.getNamespaces()) {
      namespaces.add(namespace.getName());
    }
    if (cpHost != null) {
      cpHost.postListNamespaces(namespaces);
    }
    return namespaces;
  }

  @Override
  public List<TableName> listTableNamesByNamespace(String name) throws IOException {
    checkInitialized();
    return listTableNames(name, null, true);
  }

  @Override
  public List<TableDescriptor> listTableDescriptorsByNamespace(String name) throws IOException {
    checkInitialized();
    return listTableDescriptors(name, null, null, true);
  }

  @Override
  public boolean abortProcedure(final long procId, final boolean mayInterruptIfRunning)
    throws IOException {
    if (cpHost != null) {
      cpHost.preAbortProcedure(this.procedureExecutor, procId);
    }

    final boolean result = this.procedureExecutor.abort(procId, mayInterruptIfRunning);

    if (cpHost != null) {
      cpHost.postAbortProcedure();
    }

    return result;
  }

  @Override
  public List<Procedure<?>> getProcedures() throws IOException {
    if (cpHost != null) {
      cpHost.preGetProcedures();
    }

    @SuppressWarnings({ "unchecked", "rawtypes" })
    List<Procedure<?>> procList = (List) this.procedureExecutor.getProcedures();

    if (cpHost != null) {
      cpHost.postGetProcedures(procList);
    }

    return procList;
  }

  @Override
  public List<LockedResource> getLocks() throws IOException {
    if (cpHost != null) {
      cpHost.preGetLocks();
    }

    MasterProcedureScheduler procedureScheduler =
      procedureExecutor.getEnvironment().getProcedureScheduler();

    final List<LockedResource> lockedResources = procedureScheduler.getLocks();

    if (cpHost != null) {
      cpHost.postGetLocks(lockedResources);
    }

    return lockedResources;
  }

  /**
   * Returns the list of table descriptors that match the specified request
   * @param namespace        the namespace to query, or null if querying for all
   * @param regex            The regular expression to match against, or null if querying for all
   * @param tableNameList    the list of table names, or null if querying for all
   * @param includeSysTables False to match only against userspace tables
   * @return the list of table descriptors
   */
  public List<TableDescriptor> listTableDescriptors(final String namespace, final String regex,
    final List<TableName> tableNameList, final boolean includeSysTables) throws IOException {
    List<TableDescriptor> htds = new ArrayList<>();
    if (cpHost != null) {
      cpHost.preGetTableDescriptors(tableNameList, htds, regex);
    }
    htds = getTableDescriptors(htds, namespace, regex, tableNameList, includeSysTables);
    if (cpHost != null) {
      cpHost.postGetTableDescriptors(tableNameList, htds, regex);
    }
    return htds;
  }

  /**
   * Returns the list of table names that match the specified request
   * @param regex            The regular expression to match against, or null if querying for all
   * @param namespace        the namespace to query, or null if querying for all
   * @param includeSysTables False to match only against userspace tables
   * @return the list of table names
   */
  public List<TableName> listTableNames(final String namespace, final String regex,
    final boolean includeSysTables) throws IOException {
    List<TableDescriptor> htds = new ArrayList<>();
    if (cpHost != null) {
      cpHost.preGetTableNames(htds, regex);
    }
    htds = getTableDescriptors(htds, namespace, regex, null, includeSysTables);
    if (cpHost != null) {
      cpHost.postGetTableNames(htds, regex);
    }
    List<TableName> result = new ArrayList<>(htds.size());
    for (TableDescriptor htd : htds)
      result.add(htd.getTableName());
    return result;
  }

  /**
   * Return a list of table table descriptors after applying any provided filter parameters. Note
   * that the user-facing description of this filter logic is presented on the class-level javadoc
   * of {@link NormalizeTableFilterParams}.
   */
  private List<TableDescriptor> getTableDescriptors(final List<TableDescriptor> htds,
    final String namespace, final String regex, final List<TableName> tableNameList,
    final boolean includeSysTables) throws IOException {
    if (tableNameList == null || tableNameList.isEmpty()) {
      // request for all TableDescriptors
      Collection<TableDescriptor> allHtds;
      if (namespace != null && namespace.length() > 0) {
        // Do a check on the namespace existence. Will fail if does not exist.
        this.clusterSchemaService.getNamespace(namespace);
        allHtds = tableDescriptors.getByNamespace(namespace).values();
      } else {
        allHtds = tableDescriptors.getAll().values();
      }
      for (TableDescriptor desc : allHtds) {
        if (
          tableStateManager.isTablePresent(desc.getTableName())
            && (includeSysTables || !desc.getTableName().isSystemTable())
        ) {
          htds.add(desc);
        }
      }
    } else {
      for (TableName s : tableNameList) {
        if (tableStateManager.isTablePresent(s)) {
          TableDescriptor desc = tableDescriptors.get(s);
          if (desc != null) {
            htds.add(desc);
          }
        }
      }
    }

    // Retains only those matched by regular expression.
    if (regex != null) filterTablesByRegex(htds, Pattern.compile(regex));
    return htds;
  }

  /**
   * Removes the table descriptors that don't match the pattern.
   * @param descriptors list of table descriptors to filter
   * @param pattern     the regex to use
   */
  private static void filterTablesByRegex(final Collection<TableDescriptor> descriptors,
    final Pattern pattern) {
    final String defaultNS = NamespaceDescriptor.DEFAULT_NAMESPACE_NAME_STR;
    Iterator<TableDescriptor> itr = descriptors.iterator();
    while (itr.hasNext()) {
      TableDescriptor htd = itr.next();
      String tableName = htd.getTableName().getNameAsString();
      boolean matched = pattern.matcher(tableName).matches();
      if (!matched && htd.getTableName().getNamespaceAsString().equals(defaultNS)) {
        matched = pattern.matcher(defaultNS + TableName.NAMESPACE_DELIM + tableName).matches();
      }
      if (!matched) {
        itr.remove();
      }
    }
  }

  @Override
  public long getLastMajorCompactionTimestamp(TableName table) throws IOException {
    return getClusterMetrics(EnumSet.of(Option.LIVE_SERVERS))
      .getLastMajorCompactionTimestamp(table);
  }

  @Override
  public long getLastMajorCompactionTimestampForRegion(byte[] regionName) throws IOException {
    return getClusterMetrics(EnumSet.of(Option.LIVE_SERVERS))
      .getLastMajorCompactionTimestamp(regionName);
  }

  /**
   * Gets the mob file compaction state for a specific table. Whether all the mob files are selected
   * is known during the compaction execution, but the statistic is done just before compaction
   * starts, it is hard to know the compaction type at that time, so the rough statistics are chosen
   * for the mob file compaction. Only two compaction states are available,
   * CompactionState.MAJOR_AND_MINOR and CompactionState.NONE.
   * @param tableName The current table name.
   * @return If a given table is in mob file compaction now.
   */
  public GetRegionInfoResponse.CompactionState getMobCompactionState(TableName tableName) {
    AtomicInteger compactionsCount = mobCompactionStates.get(tableName);
    if (compactionsCount != null && compactionsCount.get() != 0) {
      return GetRegionInfoResponse.CompactionState.MAJOR_AND_MINOR;
    }
    return GetRegionInfoResponse.CompactionState.NONE;
  }

  public void reportMobCompactionStart(TableName tableName) throws IOException {
    IdLock.Entry lockEntry = null;
    try {
      lockEntry = mobCompactionLock.getLockEntry(tableName.hashCode());
      AtomicInteger compactionsCount = mobCompactionStates.get(tableName);
      if (compactionsCount == null) {
        compactionsCount = new AtomicInteger(0);
        mobCompactionStates.put(tableName, compactionsCount);
      }
      compactionsCount.incrementAndGet();
    } finally {
      if (lockEntry != null) {
        mobCompactionLock.releaseLockEntry(lockEntry);
      }
    }
  }

  public void reportMobCompactionEnd(TableName tableName) throws IOException {
    IdLock.Entry lockEntry = null;
    try {
      lockEntry = mobCompactionLock.getLockEntry(tableName.hashCode());
      AtomicInteger compactionsCount = mobCompactionStates.get(tableName);
      if (compactionsCount != null) {
        int count = compactionsCount.decrementAndGet();
        // remove the entry if the count is 0.
        if (count == 0) {
          mobCompactionStates.remove(tableName);
        }
      }
    } finally {
      if (lockEntry != null) {
        mobCompactionLock.releaseLockEntry(lockEntry);
      }
    }
  }

  /**
   * Queries the state of the {@link LoadBalancerStateStore}. If the balancer is not initialized,
   * false is returned.
   * @return The state of the load balancer, or false if the load balancer isn't defined.
   */
  public boolean isBalancerOn() {
    return !isInMaintenanceMode() && loadBalancerStateStore != null && loadBalancerStateStore.get();
  }

  /**
   * Queries the state of the {@link RegionNormalizerStateStore}. If it's not initialized, false is
   * returned.
   */
  public boolean isNormalizerOn() {
    return !isInMaintenanceMode() && getRegionNormalizerManager().isNormalizerOn();
  }

  /**
   * Queries the state of the {@link SplitOrMergeStateStore}. If it is not initialized, false is
   * returned. If switchType is illegal, false will return.
   * @param switchType see {@link org.apache.hadoop.hbase.client.MasterSwitchType}
   * @return The state of the switch
   */
  @Override
  public boolean isSplitOrMergeEnabled(MasterSwitchType switchType) {
    return !isInMaintenanceMode() && splitOrMergeStateStore != null
      && splitOrMergeStateStore.isSplitOrMergeEnabled(switchType);
  }

  /**
   * Fetch the configured {@link LoadBalancer} class name. If none is set, a default is returned.
   * <p/>
   * Notice that, the base load balancer will always be {@link RSGroupBasedLoadBalancer} now, so
   * this method will return the balancer used inside each rs group.
   * @return The name of the {@link LoadBalancer} in use.
   */
  public String getLoadBalancerClassName() {
    return conf.get(HConstants.HBASE_MASTER_LOADBALANCER_CLASS,
      LoadBalancerFactory.getDefaultLoadBalancerClass().getName());
  }

  public SplitOrMergeStateStore getSplitOrMergeStateStore() {
    return splitOrMergeStateStore;
  }

  @Override
  public RSGroupBasedLoadBalancer getLoadBalancer() {
    return balancer;
  }

  @Override
  public FavoredNodesManager getFavoredNodesManager() {
    return balancer.getFavoredNodesManager();
  }

  private long executePeerProcedure(AbstractPeerProcedure<?> procedure) throws IOException {
    if (!isReplicationPeerModificationEnabled()) {
      throw new IOException("Replication peer modification disabled");
    }
    long procId = procedureExecutor.submitProcedure(procedure);
    procedure.getLatch().await();
    return procId;
  }

  @Override
  public long addReplicationPeer(String peerId, ReplicationPeerConfig peerConfig, boolean enabled)
    throws ReplicationException, IOException {
    LOG.info(getClientIdAuditPrefix() + " creating replication peer, id=" + peerId + ", config="
      + peerConfig + ", state=" + (enabled ? "ENABLED" : "DISABLED"));
    return executePeerProcedure(new AddPeerProcedure(peerId, peerConfig, enabled));
  }

  @Override
  public long removeReplicationPeer(String peerId) throws ReplicationException, IOException {
    LOG.info(getClientIdAuditPrefix() + " removing replication peer, id=" + peerId);
    return executePeerProcedure(new RemovePeerProcedure(peerId));
  }

  @Override
  public long enableReplicationPeer(String peerId) throws ReplicationException, IOException {
    LOG.info(getClientIdAuditPrefix() + " enable replication peer, id=" + peerId);
    return executePeerProcedure(new EnablePeerProcedure(peerId));
  }

  @Override
  public long disableReplicationPeer(String peerId) throws ReplicationException, IOException {
    LOG.info(getClientIdAuditPrefix() + " disable replication peer, id=" + peerId);
    return executePeerProcedure(new DisablePeerProcedure(peerId));
  }

  @Override
  public ReplicationPeerConfig getReplicationPeerConfig(String peerId)
    throws ReplicationException, IOException {
    if (cpHost != null) {
      cpHost.preGetReplicationPeerConfig(peerId);
    }
    LOG.info(getClientIdAuditPrefix() + " get replication peer config, id=" + peerId);
    ReplicationPeerConfig peerConfig = this.replicationPeerManager.getPeerConfig(peerId)
      .orElseThrow(() -> new ReplicationPeerNotFoundException(peerId));
    if (cpHost != null) {
      cpHost.postGetReplicationPeerConfig(peerId);
    }
    return peerConfig;
  }

  @Override
  public long updateReplicationPeerConfig(String peerId, ReplicationPeerConfig peerConfig)
    throws ReplicationException, IOException {
    LOG.info(getClientIdAuditPrefix() + " update replication peer config, id=" + peerId
      + ", config=" + peerConfig);
    return executePeerProcedure(new UpdatePeerConfigProcedure(peerId, peerConfig));
  }

  @Override
  public List<ReplicationPeerDescription> listReplicationPeers(String regex)
    throws ReplicationException, IOException {
    if (cpHost != null) {
      cpHost.preListReplicationPeers(regex);
    }
    LOG.debug("{} list replication peers, regex={}", getClientIdAuditPrefix(), regex);
    Pattern pattern = regex == null ? null : Pattern.compile(regex);
    List<ReplicationPeerDescription> peers = this.replicationPeerManager.listPeers(pattern);
    if (cpHost != null) {
      cpHost.postListReplicationPeers(regex);
    }
    return peers;
  }

  @Override
  public long transitReplicationPeerSyncReplicationState(String peerId, SyncReplicationState state)
    throws ReplicationException, IOException {
    LOG.info(
      getClientIdAuditPrefix()
        + " transit current cluster state to {} in a synchronous replication peer id={}",
      state, peerId);
    return executePeerProcedure(new TransitPeerSyncReplicationStateProcedure(peerId, state));
  }

  @Override
  public boolean replicationPeerModificationSwitch(boolean on) throws IOException {
    return replicationPeerModificationStateStore.set(on);
  }

  @Override
  public boolean isReplicationPeerModificationEnabled() {
    return replicationPeerModificationStateStore.get();
  }

  /**
   * Mark region server(s) as decommissioned (previously called 'draining') to prevent additional
   * regions from getting assigned to them. Also unload the regions on the servers asynchronously.0
   * @param servers Region servers to decommission.
   */
  public void decommissionRegionServers(final List<ServerName> servers, final boolean offload)
    throws IOException {
    List<ServerName> serversAdded = new ArrayList<>(servers.size());
    // Place the decommission marker first.
    String parentZnode = getZooKeeper().getZNodePaths().drainingZNode;
    for (ServerName server : servers) {
      try {
        String node = ZNodePaths.joinZNode(parentZnode, server.getServerName());
        ZKUtil.createAndFailSilent(getZooKeeper(), node);
      } catch (KeeperException ke) {
        throw new HBaseIOException(
          this.zooKeeper.prefix("Unable to decommission '" + server.getServerName() + "'."), ke);
      }
      if (this.serverManager.addServerToDrainList(server)) {
        serversAdded.add(server);
      }
    }
    // Move the regions off the decommissioned servers.
    if (offload) {
      final List<ServerName> destServers = this.serverManager.createDestinationServersList();
      for (ServerName server : serversAdded) {
        final List<RegionInfo> regionsOnServer = this.assignmentManager.getRegionsOnServer(server);
        for (RegionInfo hri : regionsOnServer) {
          ServerName dest = balancer.randomAssignment(hri, destServers);
          if (dest == null) {
            throw new HBaseIOException("Unable to determine a plan to move " + hri);
          }
          RegionPlan rp = new RegionPlan(hri, server, dest);
          this.assignmentManager.moveAsync(rp);
        }
      }
    }
  }

  /**
   * List region servers marked as decommissioned (previously called 'draining') to not get regions
   * assigned to them.
   * @return List of decommissioned servers.
   */
  public List<ServerName> listDecommissionedRegionServers() {
    return this.serverManager.getDrainingServersList();
  }

  /**
   * Remove decommission marker (previously called 'draining') from a region server to allow regions
   * assignments. Load regions onto the server asynchronously if a list of regions is given
   * @param server Region server to remove decommission marker from.
   */
  public void recommissionRegionServer(final ServerName server,
    final List<byte[]> encodedRegionNames) throws IOException {
    // Remove the server from decommissioned (draining) server list.
    String parentZnode = getZooKeeper().getZNodePaths().drainingZNode;
    String node = ZNodePaths.joinZNode(parentZnode, server.getServerName());
    try {
      ZKUtil.deleteNodeFailSilent(getZooKeeper(), node);
    } catch (KeeperException ke) {
      throw new HBaseIOException(
        this.zooKeeper.prefix("Unable to recommission '" + server.getServerName() + "'."), ke);
    }
    this.serverManager.removeServerFromDrainList(server);

    // Load the regions onto the server if we are given a list of regions.
    if (encodedRegionNames == null || encodedRegionNames.isEmpty()) {
      return;
    }
    if (!this.serverManager.isServerOnline(server)) {
      return;
    }
    for (byte[] encodedRegionName : encodedRegionNames) {
      RegionState regionState =
        assignmentManager.getRegionStates().getRegionState(Bytes.toString(encodedRegionName));
      if (regionState == null) {
        LOG.warn("Unknown region " + Bytes.toStringBinary(encodedRegionName));
        continue;
      }
      RegionInfo hri = regionState.getRegion();
      if (server.equals(regionState.getServerName())) {
        LOG.info("Skipping move of region " + hri.getRegionNameAsString()
          + " because region already assigned to the same server " + server + ".");
        continue;
      }
      RegionPlan rp = new RegionPlan(hri, regionState.getServerName(), server);
      this.assignmentManager.moveAsync(rp);
    }
  }

  @Override
  public LockManager getLockManager() {
    return lockManager;
  }

  public QuotaObserverChore getQuotaObserverChore() {
    return this.quotaObserverChore;
  }

  public SpaceQuotaSnapshotNotifier getSpaceQuotaSnapshotNotifier() {
    return this.spaceQuotaSnapshotNotifier;
  }

  @SuppressWarnings("unchecked")
  private RemoteProcedure<MasterProcedureEnv, ?> getRemoteProcedure(long procId) {
    Procedure<?> procedure = procedureExecutor.getProcedure(procId);
    if (procedure == null) {
      return null;
    }
    assert procedure instanceof RemoteProcedure;
    return (RemoteProcedure<MasterProcedureEnv, ?>) procedure;
  }

  public void remoteProcedureCompleted(long procId) {
    LOG.debug("Remote procedure done, pid={}", procId);
    RemoteProcedure<MasterProcedureEnv, ?> procedure = getRemoteProcedure(procId);
    if (procedure != null) {
      procedure.remoteOperationCompleted(procedureExecutor.getEnvironment());
    }
  }

  public void remoteProcedureFailed(long procId, RemoteProcedureException error) {
    LOG.debug("Remote procedure failed, pid={}", procId, error);
    RemoteProcedure<MasterProcedureEnv, ?> procedure = getRemoteProcedure(procId);
    if (procedure != null) {
      procedure.remoteOperationFailed(procedureExecutor.getEnvironment(), error);
    }
  }

  /**
   * Reopen regions provided in the argument
   * @param tableName   The current table name
   * @param regionNames The region names of the regions to reopen
   * @param nonceGroup  Identifier for the source of the request, a client or process
   * @param nonce       A unique identifier for this operation from the client or process identified
   *                    by <code>nonceGroup</code> (the source must ensure each operation gets a
   *                    unique id).
   * @return procedure Id
   * @throws IOException if reopening region fails while running procedure
   */
  long reopenRegions(final TableName tableName, final List<byte[]> regionNames,
    final long nonceGroup, final long nonce) throws IOException {

    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {

        @Override
        protected void run() throws IOException {
          submitProcedure(new ReopenTableRegionsProcedure(tableName, regionNames));
        }

        @Override
        protected String getDescription() {
          return "ReopenTableRegionsProcedure";
        }

      });

  }

  @Override
  public ReplicationPeerManager getReplicationPeerManager() {
    return replicationPeerManager;
  }

  @Override
  public ReplicationLogCleanerBarrier getReplicationLogCleanerBarrier() {
    return replicationLogCleanerBarrier;
  }

  @Override
  public Semaphore getSyncReplicationPeerLock() {
    return syncReplicationPeerLock;
  }

  public HashMap<String, List<Pair<ServerName, ReplicationLoadSource>>>
    getReplicationLoad(ServerName[] serverNames) {
    List<ReplicationPeerDescription> peerList = this.getReplicationPeerManager().listPeers(null);
    if (peerList == null) {
      return null;
    }
    HashMap<String, List<Pair<ServerName, ReplicationLoadSource>>> replicationLoadSourceMap =
      new HashMap<>(peerList.size());
    peerList.stream()
      .forEach(peer -> replicationLoadSourceMap.put(peer.getPeerId(), new ArrayList<>()));
    for (ServerName serverName : serverNames) {
      List<ReplicationLoadSource> replicationLoadSources =
        getServerManager().getLoad(serverName).getReplicationLoadSourceList();
      for (ReplicationLoadSource replicationLoadSource : replicationLoadSources) {
        List<Pair<ServerName, ReplicationLoadSource>> replicationLoadSourceList =
          replicationLoadSourceMap.get(replicationLoadSource.getPeerID());
        if (replicationLoadSourceList == null) {
          LOG.debug("{} does not exist, but it exists "
            + "in znode(/hbase/replication/rs). when the rs restarts, peerId is deleted, so "
            + "we just need to ignore it", replicationLoadSource.getPeerID());
          continue;
        }
        replicationLoadSourceList.add(new Pair<>(serverName, replicationLoadSource));
      }
    }
    for (List<Pair<ServerName, ReplicationLoadSource>> loads : replicationLoadSourceMap.values()) {
      if (loads.size() > 0) {
        loads.sort(Comparator.comparingLong(load -> (-1) * load.getSecond().getReplicationLag()));
      }
    }
    return replicationLoadSourceMap;
  }

  /**
   * This method modifies the master's configuration in order to inject replication-related features
   */
  @InterfaceAudience.Private
  public static void decorateMasterConfiguration(Configuration conf) {
    String plugins = conf.get(HBASE_MASTER_LOGCLEANER_PLUGINS);
    String cleanerClass = ReplicationLogCleaner.class.getCanonicalName();
    if (plugins == null || !plugins.contains(cleanerClass)) {
      conf.set(HBASE_MASTER_LOGCLEANER_PLUGINS, plugins + "," + cleanerClass);
    }
    if (ReplicationUtils.isReplicationForBulkLoadDataEnabled(conf)) {
      plugins = conf.get(HFileCleaner.MASTER_HFILE_CLEANER_PLUGINS);
      cleanerClass = ReplicationHFileCleaner.class.getCanonicalName();
      if (!plugins.contains(cleanerClass)) {
        conf.set(HFileCleaner.MASTER_HFILE_CLEANER_PLUGINS, plugins + "," + cleanerClass);
      }
    }
  }

  public SnapshotQuotaObserverChore getSnapshotQuotaObserverChore() {
    return this.snapshotQuotaChore;
  }

  public ActiveMasterManager getActiveMasterManager() {
    return activeMasterManager;
  }

  @Override
  public SyncReplicationReplayWALManager getSyncReplicationReplayWALManager() {
    return this.syncReplicationReplayWALManager;
  }

  @Override
  public HbckChore getHbckChore() {
    return this.hbckChore;
  }

  @Override
  public void runReplicationBarrierCleaner() {
    ReplicationBarrierCleaner rbc = this.replicationBarrierCleaner;
    if (rbc != null) {
      rbc.chore();
    }
  }

  @Override
  public RSGroupInfoManager getRSGroupInfoManager() {
    return rsGroupInfoManager;
  }

  /**
   * Get the compaction state of the table
   * @param tableName The table name
   * @return CompactionState Compaction state of the table
   */
  public CompactionState getCompactionState(final TableName tableName) {
    CompactionState compactionState = CompactionState.NONE;
    try {
      List<RegionInfo> regions = assignmentManager.getRegionStates().getRegionsOfTable(tableName);
      for (RegionInfo regionInfo : regions) {
        ServerName serverName =
          assignmentManager.getRegionStates().getRegionServerOfRegion(regionInfo);
        if (serverName == null) {
          continue;
        }
        ServerMetrics sl = serverManager.getLoad(serverName);
        if (sl == null) {
          continue;
        }
        RegionMetrics regionMetrics = sl.getRegionMetrics().get(regionInfo.getRegionName());
        if (regionMetrics == null) {
          LOG.warn("Can not get compaction details for the region: {} , it may be not online.",
            regionInfo.getRegionNameAsString());
          continue;
        }
        if (regionMetrics.getCompactionState() == CompactionState.MAJOR) {
          if (compactionState == CompactionState.MINOR) {
            compactionState = CompactionState.MAJOR_AND_MINOR;
          } else {
            compactionState = CompactionState.MAJOR;
          }
        } else if (regionMetrics.getCompactionState() == CompactionState.MINOR) {
          if (compactionState == CompactionState.MAJOR) {
            compactionState = CompactionState.MAJOR_AND_MINOR;
          } else {
            compactionState = CompactionState.MINOR;
          }
        }
      }
    } catch (Exception e) {
      compactionState = null;
      LOG.error("Exception when get compaction state for " + tableName.getNameAsString(), e);
    }
    return compactionState;
  }

  @Override
  public MetaLocationSyncer getMetaLocationSyncer() {
    return metaLocationSyncer;
  }

  @RestrictedApi(explanation = "Should only be called in tests", link = "",
      allowedOnPath = ".*/src/test/.*")
  public MasterRegion getMasterRegion() {
    return masterRegion;
  }

  @Override
  public void onConfigurationChange(Configuration newConf) {
    try {
      Superusers.initialize(newConf);
    } catch (IOException e) {
      LOG.warn("Failed to initialize SuperUsers on reloading of the configuration");
    }
    // append the quotas observer back to the master coprocessor key
    setQuotasObserver(newConf);
    // update region server coprocessor if the configuration has changed.
    if (
      CoprocessorConfigurationUtil.checkConfigurationChange(getConfiguration(), newConf,
        CoprocessorHost.MASTER_COPROCESSOR_CONF_KEY) && !maintenanceMode
    ) {
      LOG.info("Update the master coprocessor(s) because the configuration has changed");
      initializeCoprocessorHost(newConf);
    }
  }

  @Override
  protected NamedQueueRecorder createNamedQueueRecord() {
    final boolean isBalancerDecisionRecording =
      conf.getBoolean(BaseLoadBalancer.BALANCER_DECISION_BUFFER_ENABLED,
        BaseLoadBalancer.DEFAULT_BALANCER_DECISION_BUFFER_ENABLED);
    final boolean isBalancerRejectionRecording =
      conf.getBoolean(BaseLoadBalancer.BALANCER_REJECTION_BUFFER_ENABLED,
        BaseLoadBalancer.DEFAULT_BALANCER_REJECTION_BUFFER_ENABLED);
    if (isBalancerDecisionRecording || isBalancerRejectionRecording) {
      return NamedQueueRecorder.getInstance(conf);
    } else {
      return null;
    }
  }

  @Override
  protected boolean clusterMode() {
    return true;
  }

  public String getClusterId() {
    if (activeMaster) {
      return clusterId;
    }
    return cachedClusterId.getFromCacheOrFetch();
  }

  public Optional<ServerName> getActiveMaster() {
    return activeMasterManager.getActiveMasterServerName();
  }

  public List<ServerName> getBackupMasters() {
    return activeMasterManager.getBackupMasters();
  }

  @Override
  public Iterator<ServerName> getBootstrapNodes() {
    return regionServerTracker.getRegionServers().iterator();
  }

  @Override
  public List<HRegionLocation> getMetaLocations() {
    return metaRegionLocationCache.getMetaRegionLocations();
  }

  @Override
  public void flushMasterStore() throws IOException {
    LOG.info("Force flush master local region.");
    if (this.cpHost != null) {
      try {
        cpHost.preMasterStoreFlush();
      } catch (IOException ioe) {
        LOG.error("Error invoking master coprocessor preMasterStoreFlush()", ioe);
      }
    }
    masterRegion.flush(true);
    if (this.cpHost != null) {
      try {
        cpHost.postMasterStoreFlush();
      } catch (IOException ioe) {
        LOG.error("Error invoking master coprocessor postMasterStoreFlush()", ioe);
      }
    }
  }

  public Collection<ServerName> getLiveRegionServers() {
    return regionServerTracker.getRegionServers();
  }

  @RestrictedApi(explanation = "Should only be called in tests", link = "",
      allowedOnPath = ".*/src/test/.*")
  void setLoadBalancer(RSGroupBasedLoadBalancer loadBalancer) {
    this.balancer = loadBalancer;
  }

  @RestrictedApi(explanation = "Should only be called in tests", link = "",
      allowedOnPath = ".*/src/test/.*")
  void setAssignmentManager(AssignmentManager assignmentManager) {
    this.assignmentManager = assignmentManager;
  }

  @RestrictedApi(explanation = "Should only be called in tests", link = "",
      allowedOnPath = ".*/src/test/.*")
  static void setDisableBalancerChoreForTest(boolean disable) {
    disableBalancerChoreForTest = disable;
  }

  private void setQuotasObserver(Configuration conf) {
    // Add the Observer to delete quotas on table deletion before starting all CPs by
    // default with quota support, avoiding if user specifically asks to not load this Observer.
    if (QuotaUtil.isQuotaEnabled(conf)) {
      updateConfigurationForQuotasObserver(conf);
    }
  }

  private void initializeCoprocessorHost(Configuration conf) {
    // initialize master side coprocessors before we start handling requests
    this.cpHost = new MasterCoprocessorHost(this, conf);
  }

  @Override
  public long flushTable(TableName tableName, List<byte[]> columnFamilies, long nonceGroup,
    long nonce) throws IOException {
    checkInitialized();

    if (
      !getConfiguration().getBoolean(MasterFlushTableProcedureManager.FLUSH_PROCEDURE_ENABLED,
        MasterFlushTableProcedureManager.FLUSH_PROCEDURE_ENABLED_DEFAULT)
    ) {
      throw new DoNotRetryIOException("FlushTableProcedureV2 is DISABLED");
    }

    return MasterProcedureUtil
      .submitProcedure(new MasterProcedureUtil.NonceProcedureRunnable(this, nonceGroup, nonce) {
        @Override
        protected void run() throws IOException {
          getMaster().getMasterCoprocessorHost().preTableFlush(tableName);
          LOG.info(getClientIdAuditPrefix() + " flush " + tableName);
          submitProcedure(
            new FlushTableProcedure(procedureExecutor.getEnvironment(), tableName, columnFamilies));
          getMaster().getMasterCoprocessorHost().postTableFlush(tableName);
        }

        @Override
        protected String getDescription() {
          return "FlushTableProcedure";
        }
      });
  }
}