hbase-server/src/main/java/org/apache/hadoop/hbase/master/HMaster.java

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