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