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