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

import java.io.IOException;
import java.net.InetAddress;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentNavigableMap;
import java.util.concurrent.ConcurrentSkipListMap;
import java.util.concurrent.CopyOnWriteArrayList;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.ClockOutOfSyncException;
import org.apache.hadoop.hbase.HConstants;
import org.apache.hadoop.hbase.HRegionInfo;
import org.apache.hadoop.hbase.NotServingRegionException;
import org.apache.hadoop.hbase.RegionLoad;
import org.apache.hadoop.hbase.ServerLoad;
import org.apache.hadoop.hbase.ServerName;
import org.apache.hadoop.hbase.YouAreDeadException;
import org.apache.hadoop.hbase.ZooKeeperConnectionException;
import org.apache.hadoop.hbase.classification.InterfaceAudience;
import org.apache.hadoop.hbase.client.ClusterConnection;
import org.apache.hadoop.hbase.client.RetriesExhaustedException;
import org.apache.hadoop.hbase.ipc.PayloadCarryingRpcController;
import org.apache.hadoop.hbase.ipc.RpcControllerFactory;
import org.apache.hadoop.hbase.master.balancer.BaseLoadBalancer;
import org.apache.hadoop.hbase.master.procedure.ServerCrashProcedure;
import org.apache.hadoop.hbase.monitoring.MonitoredTask;
import org.apache.hadoop.hbase.protobuf.ProtobufUtil;
import org.apache.hadoop.hbase.protobuf.RequestConverter;
import org.apache.hadoop.hbase.protobuf.ResponseConverter;
import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.AdminService;
import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionRequest;
import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.OpenRegionResponse;
import org.apache.hadoop.hbase.protobuf.generated.AdminProtos.ServerInfo;
import org.apache.hadoop.hbase.protobuf.generated.RegionServerStatusProtos.RegionServerStartupRequest;
import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.RegionStoreSequenceIds;
import org.apache.hadoop.hbase.protobuf.generated.ClusterStatusProtos.StoreSequenceId;
import org.apache.hadoop.hbase.protobuf.generated.ZooKeeperProtos.SplitLogTask.RecoveryMode;
import org.apache.hadoop.hbase.regionserver.HRegionServer;
import org.apache.hadoop.hbase.regionserver.RegionOpeningState;
import org.apache.hadoop.hbase.security.User;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.hbase.util.Pair;
import org.apache.hadoop.hbase.util.RetryCounter;
import org.apache.hadoop.hbase.util.RetryCounterFactory;
import org.apache.hadoop.hbase.zookeeper.ZKUtil;
import org.apache.hadoop.hbase.zookeeper.ZooKeeperWatcher;
import org.apache.zookeeper.KeeperException;

import com.google.common.annotations.VisibleForTesting;
import com.google.protobuf.ByteString;
import com.google.protobuf.ServiceException;

/**
 * The ServerManager class manages info about region servers.
 * <p>
 * Maintains lists of online and dead servers.  Processes the startups,
 * shutdowns, and deaths of region servers.
 * <p>
 * Servers are distinguished in two different ways.  A given server has a
 * location, specified by hostname and port, and of which there can only be one
 * online at any given time.  A server instance is specified by the location
 * (hostname and port) as well as the startcode (timestamp from when the server
 * was started).  This is used to differentiate a restarted instance of a given
 * server from the original instance.
 * <p>
 * If a sever is known not to be running any more, it is called dead. The dead
 * server needs to be handled by a ServerShutdownHandler.  If the handler is not
 * enabled yet, the server can't be handled right away so it is queued up.
 * After the handler is enabled, the server will be submitted to a handler to handle.
 * However, the handler may be just partially enabled.  If so,
 * the server cannot be fully processed, and be queued up for further processing.
 * A server is fully processed only after the handler is fully enabled
 * and has completed the handling.
 */
@InterfaceAudience.Private
public class ServerManager {
  public static final String WAIT_ON_REGIONSERVERS_MAXTOSTART =
      "hbase.master.wait.on.regionservers.maxtostart";

  public static final String WAIT_ON_REGIONSERVERS_MINTOSTART =
      "hbase.master.wait.on.regionservers.mintostart";

  public static final String WAIT_ON_REGIONSERVERS_TIMEOUT =
      "hbase.master.wait.on.regionservers.timeout";

  public static final String WAIT_ON_REGIONSERVERS_INTERVAL =
      "hbase.master.wait.on.regionservers.interval";

  private static final Log LOG = LogFactory.getLog(ServerManager.class);

  // Set if we are to shutdown the cluster.
  private volatile boolean clusterShutdown = false;

  /**
   * The last flushed sequence id for a region.
   */
  private final ConcurrentNavigableMap<byte[], Long> flushedSequenceIdByRegion =
    new ConcurrentSkipListMap<byte[], Long>(Bytes.BYTES_COMPARATOR);

  /**
   * The last flushed sequence id for a store in a region.
   */
  private final ConcurrentNavigableMap<byte[], ConcurrentNavigableMap<byte[], Long>>
    storeFlushedSequenceIdsByRegion =
    new ConcurrentSkipListMap<byte[], ConcurrentNavigableMap<byte[], Long>>(Bytes.BYTES_COMPARATOR);

  /** Map of registered servers to their current load */
  private final ConcurrentNavigableMap<ServerName, ServerLoad> onlineServers =
    new ConcurrentSkipListMap<ServerName, ServerLoad>();

  /**
   * Map of admin interfaces per registered regionserver; these interfaces we use to control
   * regionservers out on the cluster
   */
  private final Map<ServerName, AdminService.BlockingInterface> rsAdmins =
    new HashMap<ServerName, AdminService.BlockingInterface>();

  /** List of region servers that should not get any more new regions. */
  private final ArrayList<ServerName> drainingServers =
    new ArrayList<ServerName>();

  private final MasterServices master;
  private final ClusterConnection connection;

  private final DeadServer deadservers = new DeadServer();

  private final long maxSkew;
  private final long warningSkew;

  private final RetryCounterFactory pingRetryCounterFactory;
  private final RpcControllerFactory rpcControllerFactory;

  /**
   * Set of region servers which are dead but not processed immediately. If one
   * server died before master enables ServerShutdownHandler, the server will be
   * added to this set and will be processed through calling
   * {@link ServerManager#processQueuedDeadServers()} by master.
   * <p>
   * A dead server is a server instance known to be dead, not listed in the /hbase/rs
   * znode any more. It may have not been submitted to ServerShutdownHandler yet
   * because the handler is not enabled.
   * <p>
   * A dead server, which has been submitted to ServerShutdownHandler while the
   * handler is not enabled, is queued up.
   * <p>
   * So this is a set of region servers known to be dead but not submitted to
   * ServerShutdownHandler for processing yet.
   */
  private Set<ServerName> queuedDeadServers = new HashSet<ServerName>();

  /**
   * Set of region servers which are dead and submitted to ServerShutdownHandler to process but not
   * fully processed immediately.
   * <p>
   * If one server died before assignment manager finished the failover cleanup, the server will be
   * added to this set and will be processed through calling
   * {@link ServerManager#processQueuedDeadServers()} by assignment manager.
   * <p>
   * The Boolean value indicates whether log split is needed inside ServerShutdownHandler
   * <p>
   * ServerShutdownHandler processes a dead server submitted to the handler after the handler is
   * enabled. It may not be able to complete the processing because meta is not yet online or master
   * is currently in startup mode. In this case, the dead server will be parked in this set
   * temporarily.
   */
  private Map<ServerName, Boolean> requeuedDeadServers
    = new ConcurrentHashMap<ServerName, Boolean>();

  /** Listeners that are called on server events. */
  private List<ServerListener> listeners = new CopyOnWriteArrayList<ServerListener>();

  /**
   * Constructor.
   * @param master
   * @throws ZooKeeperConnectionException
   */
  public ServerManager(final MasterServices master) throws IOException {
    this(master, true);
  }

  ServerManager(final MasterServices master, final boolean connect) throws IOException {
    this.master = master;
    Configuration c = master.getConfiguration();
    maxSkew = c.getLong("hbase.master.maxclockskew", 30000);
    warningSkew = c.getLong("hbase.master.warningclockskew", 10000);
    this.connection = connect ? master.getClusterConnection() : null;
    int pingMaxAttempts = Math.max(1, master.getConfiguration().getInt(
      "hbase.master.maximum.ping.server.attempts", 10));
    int pingSleepInterval = Math.max(1, master.getConfiguration().getInt(
      "hbase.master.ping.server.retry.sleep.interval", 100));
    this.pingRetryCounterFactory = new RetryCounterFactory(pingMaxAttempts, pingSleepInterval);
    this.rpcControllerFactory = this.connection == null
        ? null
        : connection.getRpcControllerFactory();
  }

  /**
   * Add the listener to the notification list.
   * @param listener The ServerListener to register
   */
  public void registerListener(final ServerListener listener) {
    this.listeners.add(listener);
  }

  /**
   * Remove the listener from the notification list.
   * @param listener The ServerListener to unregister
   */
  public boolean unregisterListener(final ServerListener listener) {
    return this.listeners.remove(listener);
  }

  /**
   * Let the server manager know a new regionserver has come online
   * @param request the startup request
   * @param ia the InetAddress from which request is received
   * @return The ServerName we know this server as.
   * @throws IOException
   */
  ServerName regionServerStartup(RegionServerStartupRequest request, InetAddress ia)
      throws IOException {
    // Test for case where we get a region startup message from a regionserver
    // that has been quickly restarted but whose znode expiration handler has
    // not yet run, or from a server whose fail we are currently processing.
    // Test its host+port combo is present in serverAddresstoServerInfo.  If it
    // is, reject the server and trigger its expiration. The next time it comes
    // in, it should have been removed from serverAddressToServerInfo and queued
    // for processing by ProcessServerShutdown.

    final String hostname = request.hasUseThisHostnameInstead() ?
        request.getUseThisHostnameInstead() :ia.getHostName();
    ServerName sn = ServerName.valueOf(hostname, request.getPort(),
      request.getServerStartCode());
    checkClockSkew(sn, request.getServerCurrentTime());
    checkIsDead(sn, "STARTUP");
    if (!checkAndRecordNewServer(sn, ServerLoad.EMPTY_SERVERLOAD)) {
      LOG.warn("THIS SHOULD NOT HAPPEN, RegionServerStartup"
        + " could not record the server: " + sn);
    }
    return sn;
  }

  private ConcurrentNavigableMap<byte[], Long> getOrCreateStoreFlushedSequenceId(
    byte[] regionName) {
    ConcurrentNavigableMap<byte[], Long> storeFlushedSequenceId =
        storeFlushedSequenceIdsByRegion.get(regionName);
    if (storeFlushedSequenceId != null) {
      return storeFlushedSequenceId;
    }
    storeFlushedSequenceId = new ConcurrentSkipListMap<byte[], Long>(Bytes.BYTES_COMPARATOR);
    ConcurrentNavigableMap<byte[], Long> alreadyPut =
        storeFlushedSequenceIdsByRegion.putIfAbsent(regionName, storeFlushedSequenceId);
    return alreadyPut == null ? storeFlushedSequenceId : alreadyPut;
  }
  /**
   * Updates last flushed sequence Ids for the regions on server sn
   * @param sn
   * @param hsl
   */
  private void updateLastFlushedSequenceIds(ServerName sn, ServerLoad hsl) {
    Map<byte[], RegionLoad> regionsLoad = hsl.getRegionsLoad();
    for (Entry<byte[], RegionLoad> entry : regionsLoad.entrySet()) {
      byte[] encodedRegionName = Bytes.toBytes(HRegionInfo.encodeRegionName(entry.getKey()));
      Long existingValue = flushedSequenceIdByRegion.get(encodedRegionName);
      long l = entry.getValue().getCompleteSequenceId();
      // Don't let smaller sequence ids override greater sequence ids.
      if (LOG.isTraceEnabled()) {
        LOG.trace(Bytes.toString(encodedRegionName) + ", existingValue=" + existingValue +
          ", completeSequenceId=" + l);
      }
      if (existingValue == null || (l != HConstants.NO_SEQNUM && l > existingValue)) {
        flushedSequenceIdByRegion.put(encodedRegionName, l);
      } else if (l != HConstants.NO_SEQNUM && l < existingValue) {
        LOG.warn("RegionServer " + sn + " indicates a last flushed sequence id ("
            + l + ") that is less than the previous last flushed sequence id ("
            + existingValue + ") for region " + Bytes.toString(entry.getKey()) + " Ignoring.");
      }
      ConcurrentNavigableMap<byte[], Long> storeFlushedSequenceId =
          getOrCreateStoreFlushedSequenceId(encodedRegionName);
      for (StoreSequenceId storeSeqId : entry.getValue().getStoreCompleteSequenceId()) {
        byte[] family = storeSeqId.getFamilyName().toByteArray();
        existingValue = storeFlushedSequenceId.get(family);
        l = storeSeqId.getSequenceId();
        if (LOG.isTraceEnabled()) {
          LOG.trace(Bytes.toString(encodedRegionName) + ", family=" + Bytes.toString(family) +
            ", existingValue=" + existingValue + ", completeSequenceId=" + l);
        }
        // Don't let smaller sequence ids override greater sequence ids.
        if (existingValue == null || (l != HConstants.NO_SEQNUM && l > existingValue.longValue())) {
          storeFlushedSequenceId.put(family, l);
        }
      }
    }
  }

  void regionServerReport(ServerName sn,
      ServerLoad sl) throws YouAreDeadException {
    checkIsDead(sn, "REPORT");
    if (null == this.onlineServers.replace(sn, sl)) {
      // Already have this host+port combo and its just different start code?
      // Just let the server in. Presume master joining a running cluster.
      // recordNewServer is what happens at the end of reportServerStartup.
      // The only thing we are skipping is passing back to the regionserver
      // the ServerName to use. Here we presume a master has already done
      // that so we'll press on with whatever it gave us for ServerName.
      if (!checkAndRecordNewServer(sn, sl)) {
        LOG.info("RegionServerReport ignored, could not record the server: " + sn);
        return; // Not recorded, so no need to move on
      }
    }
    updateLastFlushedSequenceIds(sn, sl);
  }

  /**
   * Check is a server of same host and port already exists,
   * if not, or the existed one got a smaller start code, record it.
   *
   * @param serverName the server to check and record
   * @param sl the server load on the server
   * @return true if the server is recorded, otherwise, false
   */
  boolean checkAndRecordNewServer(
      final ServerName serverName, final ServerLoad sl) {
    ServerName existingServer = null;
    synchronized (this.onlineServers) {
      existingServer = findServerWithSameHostnamePortWithLock(serverName);
      if (existingServer != null && (existingServer.getStartcode() > serverName.getStartcode())) {
        LOG.info("Server serverName=" + serverName + " rejected; we already have "
            + existingServer.toString() + " registered with same hostname and port");
        return false;
      }
      recordNewServerWithLock(serverName, sl);
    }

    // Tell our listeners that a server was added
    if (!this.listeners.isEmpty()) {
      for (ServerListener listener : this.listeners) {
        listener.serverAdded(serverName);
      }
    }

    // Note that we assume that same ts means same server, and don't expire in that case.
    //  TODO: ts can theoretically collide due to clock shifts, so this is a bit hacky.
    if (existingServer != null && (existingServer.getStartcode() < serverName.getStartcode())) {
      LOG.info("Triggering server recovery; existingServer " +
          existingServer + " looks stale, new server:" + serverName);
      expireServer(existingServer);
    }
    return true;
  }

  /**
   * Checks if the clock skew between the server and the master. If the clock skew exceeds the
   * configured max, it will throw an exception; if it exceeds the configured warning threshold,
   * it will log a warning but start normally.
   * @param serverName Incoming servers's name
   * @param serverCurrentTime
   * @throws ClockOutOfSyncException if the skew exceeds the configured max value
   */
  private void checkClockSkew(final ServerName serverName, final long serverCurrentTime)
  throws ClockOutOfSyncException {
    long skew = Math.abs(System.currentTimeMillis() - serverCurrentTime);
    if (skew > maxSkew) {
      String message = "Server " + serverName + " has been " +
        "rejected; Reported time is too far out of sync with master.  " +
        "Time difference of " + skew + "ms > max allowed of " + maxSkew + "ms";
      LOG.warn(message);
      throw new ClockOutOfSyncException(message);
    } else if (skew > warningSkew){
      String message = "Reported time for server " + serverName + " is out of sync with master " +
        "by " + skew + "ms. (Warning threshold is " + warningSkew + "ms; " +
        "error threshold is " + maxSkew + "ms)";
      LOG.warn(message);
    }
  }

  /**
   * If this server is on the dead list, reject it with a YouAreDeadException.
   * If it was dead but came back with a new start code, remove the old entry
   * from the dead list.
   * @param serverName
   * @param what START or REPORT
   * @throws org.apache.hadoop.hbase.YouAreDeadException
   */
  private void checkIsDead(final ServerName serverName, final String what)
      throws YouAreDeadException {
    if (this.deadservers.isDeadServer(serverName)) {
      // host name, port and start code all match with existing one of the
      // dead servers. So, this server must be dead.
      String message = "Server " + what + " rejected; currently processing " +
          serverName + " as dead server";
      LOG.debug(message);
      throw new YouAreDeadException(message);
    }
    // remove dead server with same hostname and port of newly checking in rs after master
    // initialization.See HBASE-5916 for more information.
    if ((this.master == null || this.master.isInitialized())
        && this.deadservers.cleanPreviousInstance(serverName)) {
      // This server has now become alive after we marked it as dead.
      // We removed it's previous entry from the dead list to reflect it.
      LOG.debug(what + ":" + " Server " + serverName + " came back up," +
          " removed it from the dead servers list");
    }
  }

  /**
   * Assumes onlineServers is locked.
   * @return ServerName with matching hostname and port.
   */
  private ServerName findServerWithSameHostnamePortWithLock(
      final ServerName serverName) {
    ServerName end = ServerName.valueOf(serverName.getHostname(), serverName.getPort(),
        Long.MAX_VALUE);

    ServerName r = onlineServers.lowerKey(end);
    if (r != null) {
      if (ServerName.isSameHostnameAndPort(r, serverName)) {
        return r;
      }
    }
    return null;
  }

  /**
   * Adds the onlineServers list. onlineServers should be locked.
   * @param serverName The remote servers name.
   * @param sl
   * @return Server load from the removed server, if any.
   */
  @VisibleForTesting
  void recordNewServerWithLock(final ServerName serverName, final ServerLoad sl) {
    LOG.info("Registering server=" + serverName);
    this.onlineServers.put(serverName, sl);
    this.rsAdmins.remove(serverName);
  }

  public RegionStoreSequenceIds getLastFlushedSequenceId(byte[] encodedRegionName) {
    RegionStoreSequenceIds.Builder builder = RegionStoreSequenceIds.newBuilder();
    Long seqId = flushedSequenceIdByRegion.get(encodedRegionName);
    builder.setLastFlushedSequenceId(seqId != null ? seqId.longValue() : HConstants.NO_SEQNUM);
    Map<byte[], Long> storeFlushedSequenceId =
        storeFlushedSequenceIdsByRegion.get(encodedRegionName);
    if (storeFlushedSequenceId != null) {
      for (Map.Entry<byte[], Long> entry : storeFlushedSequenceId.entrySet()) {
        builder.addStoreSequenceId(StoreSequenceId.newBuilder()
            .setFamilyName(ByteString.copyFrom(entry.getKey()))
            .setSequenceId(entry.getValue().longValue()).build());
      }
    }
    return builder.build();
  }

  /**
   * @param serverName
   * @return ServerLoad if serverName is known else null
   */
  public ServerLoad getLoad(final ServerName serverName) {
    return this.onlineServers.get(serverName);
  }

  /**
   * Compute the average load across all region servers.
   * Currently, this uses a very naive computation - just uses the number of
   * regions being served, ignoring stats about number of requests.
   * @return the average load
   */
  public double getAverageLoad() {
    int totalLoad = 0;
    int numServers = 0;
    for (ServerLoad sl: this.onlineServers.values()) {
        numServers++;
        totalLoad += sl.getNumberOfRegions();
    }
    return numServers == 0 ? 0 :
      (double)totalLoad / (double)numServers;
  }

  /** @return the count of active regionservers */
  public int countOfRegionServers() {
    // Presumes onlineServers is a concurrent map
    return this.onlineServers.size();
  }

  /**
   * @return Read-only map of servers to serverinfo
   */
  public Map<ServerName, ServerLoad> getOnlineServers() {
    // Presumption is that iterating the returned Map is OK.
    synchronized (this.onlineServers) {
      return Collections.unmodifiableMap(this.onlineServers);
    }
  }


  public DeadServer getDeadServers() {
    return this.deadservers;
  }

  /**
   * Checks if any dead servers are currently in progress.
   * @return true if any RS are being processed as dead, false if not
   */
  public boolean areDeadServersInProgress() {
    return this.deadservers.areDeadServersInProgress();
  }

  void letRegionServersShutdown() {
    long previousLogTime = 0;
    ServerName sn = master.getServerName();
    ZooKeeperWatcher zkw = master.getZooKeeper();
    int onlineServersCt;
    while ((onlineServersCt = onlineServers.size()) > 0){

      if (System.currentTimeMillis() > (previousLogTime + 1000)) {
        Set<ServerName> remainingServers = onlineServers.keySet();
        synchronized (onlineServers) {
          if (remainingServers.size() == 1 && remainingServers.contains(sn)) {
            // Master will delete itself later.
            return;
          }
        }
        StringBuilder sb = new StringBuilder();
        // It's ok here to not sync on onlineServers - merely logging
        for (ServerName key : remainingServers) {
          if (sb.length() > 0) {
            sb.append(", ");
          }
          sb.append(key);
        }
        LOG.info("Waiting on regionserver(s) to go down " + sb.toString());
        previousLogTime = System.currentTimeMillis();
      }

      try {
        List<String> servers = ZKUtil.listChildrenNoWatch(zkw, zkw.rsZNode);
        if (servers == null || servers.size() == 0 || (servers.size() == 1
            && servers.contains(sn.toString()))) {
          LOG.info("ZK shows there is only the master self online, exiting now");
          // Master could have lost some ZK events, no need to wait more.
          break;
        }
      } catch (KeeperException ke) {
        LOG.warn("Failed to list regionservers", ke);
        // ZK is malfunctioning, don't hang here
        break;
      }
      synchronized (onlineServers) {
        try {
          if (onlineServersCt == onlineServers.size()) onlineServers.wait(100);
        } catch (InterruptedException ignored) {
          // continue
        }
      }
    }
  }

  /*
   * Expire the passed server.  Add it to list of dead servers and queue a
   * shutdown processing.
   */
  public synchronized void expireServer(final ServerName serverName) {
    if (serverName.equals(master.getServerName())) {
      if (!(master.isAborted() || master.isStopped())) {
        master.stop("We lost our znode?");
      }
      return;
    }
    if (!master.isServerCrashProcessingEnabled()) {
      LOG.info("Master doesn't enable ServerShutdownHandler during initialization, "
          + "delay expiring server " + serverName);
      this.queuedDeadServers.add(serverName);
      return;
    }
    if (this.deadservers.isDeadServer(serverName)) {
      // TODO: Can this happen?  It shouldn't be online in this case?
      LOG.warn("Expiration of " + serverName +
          " but server shutdown already in progress");
      return;
    }
    moveFromOnelineToDeadServers(serverName);

    // If cluster is going down, yes, servers are going to be expiring; don't
    // process as a dead server
    if (this.clusterShutdown) {
      LOG.info("Cluster shutdown set; " + serverName +
        " expired; onlineServers=" + this.onlineServers.size());
      if (this.onlineServers.isEmpty()) {
        master.stop("Cluster shutdown set; onlineServer=0");
      }
      return;
    }

    boolean carryingMeta = master.getAssignmentManager().isCarryingMeta(serverName);
    this.master.getMasterProcedureExecutor().
      submitProcedure(new ServerCrashProcedure(serverName, true, carryingMeta));
    LOG.debug("Added=" + serverName +
      " to dead servers, submitted shutdown handler to be executed meta=" + carryingMeta);

    // Tell our listeners that a server was removed
    if (!this.listeners.isEmpty()) {
      for (ServerListener listener : this.listeners) {
        listener.serverRemoved(serverName);
      }
    }
  }

  @VisibleForTesting
  public void moveFromOnelineToDeadServers(final ServerName sn) {
    synchronized (onlineServers) {
      if (!this.onlineServers.containsKey(sn)) {
        LOG.warn("Expiration of " + sn + " but server not online");
      }
      // Remove the server from the known servers lists and update load info BUT
      // add to deadservers first; do this so it'll show in dead servers list if
      // not in online servers list.
      this.deadservers.add(sn);
      this.onlineServers.remove(sn);
      onlineServers.notifyAll();
    }
    this.rsAdmins.remove(sn);
  }

  public synchronized void processDeadServer(final ServerName serverName, boolean shouldSplitWal) {
    // When assignment manager is cleaning up the zookeeper nodes and rebuilding the
    // in-memory region states, region servers could be down. Meta table can and
    // should be re-assigned, log splitting can be done too. However, it is better to
    // wait till the cleanup is done before re-assigning user regions.
    //
    // We should not wait in the server shutdown handler thread since it can clog
    // the handler threads and meta table could not be re-assigned in case
    // the corresponding server is down. So we queue them up here instead.
    if (!master.getAssignmentManager().isFailoverCleanupDone()) {
      requeuedDeadServers.put(serverName, shouldSplitWal);
      return;
    }

    this.deadservers.add(serverName);
    this.master.getMasterProcedureExecutor().
    submitProcedure(new ServerCrashProcedure(serverName, shouldSplitWal, false));
  }

  /**
   * Process the servers which died during master's initialization. It will be
   * called after HMaster#assignMeta and AssignmentManager#joinCluster.
   * */
  synchronized void processQueuedDeadServers() {
    if (!master.isServerCrashProcessingEnabled()) {
      LOG.info("Master hasn't enabled ServerShutdownHandler");
    }
    Iterator<ServerName> serverIterator = queuedDeadServers.iterator();
    while (serverIterator.hasNext()) {
      ServerName tmpServerName = serverIterator.next();
      expireServer(tmpServerName);
      serverIterator.remove();
      requeuedDeadServers.remove(tmpServerName);
    }

    if (!master.getAssignmentManager().isFailoverCleanupDone()) {
      LOG.info("AssignmentManager hasn't finished failover cleanup; waiting");
    }

    for (Map.Entry<ServerName, Boolean> entry : requeuedDeadServers.entrySet()) {
      processDeadServer(entry.getKey(), entry.getValue());
    }
    requeuedDeadServers.clear();
  }

  /*
   * Remove the server from the drain list.
   */
  public boolean removeServerFromDrainList(final ServerName sn) {
    // Warn if the server (sn) is not online.  ServerName is of the form:
    // <hostname> , <port> , <startcode>

    if (!this.isServerOnline(sn)) {
      LOG.warn("Server " + sn + " is not currently online. " +
               "Removing from draining list anyway, as requested.");
    }
    // Remove the server from the draining servers lists.
    return this.drainingServers.remove(sn);
  }

  /*
   * Add the server to the drain list.
   */
  public boolean addServerToDrainList(final ServerName sn) {
    // Warn if the server (sn) is not online.  ServerName is of the form:
    // <hostname> , <port> , <startcode>

    if (!this.isServerOnline(sn)) {
      LOG.warn("Server " + sn + " is not currently online. " +
               "Ignoring request to add it to draining list.");
      return false;
    }
    // Add the server to the draining servers lists, if it's not already in
    // it.
    if (this.drainingServers.contains(sn)) {
      LOG.warn("Server " + sn + " is already in the draining server list." +
               "Ignoring request to add it again.");
      return false;
    }
    return this.drainingServers.add(sn);
  }

  // RPC methods to region servers

  /**
   * Sends an OPEN RPC to the specified server to open the specified region.
   * <p>
   * Open should not fail but can if server just crashed.
   * <p>
   * @param server server to open a region
   * @param region region to open
   * @param favoredNodes
   */
  public RegionOpeningState sendRegionOpen(final ServerName server,
      HRegionInfo region, List<ServerName> favoredNodes)
  throws IOException {
    AdminService.BlockingInterface admin = getRsAdmin(server);
    if (admin == null) {
      throw new IOException("Attempting to send OPEN RPC to server " + server.toString() +
        " failed because no RPC connection found to this server");
    }
    OpenRegionRequest request = RequestConverter.buildOpenRegionRequest(server,
      region, favoredNodes,
      (RecoveryMode.LOG_REPLAY == this.master.getMasterWalManager().getLogRecoveryMode()));
    try {
      OpenRegionResponse response = admin.openRegion(null, request);
      return ResponseConverter.getRegionOpeningState(response);
    } catch (ServiceException se) {
      throw ProtobufUtil.getRemoteException(se);
    }
  }

  /**
   * Sends an OPEN RPC to the specified server to open the specified region.
   * <p>
   * Open should not fail but can if server just crashed.
   * <p>
   * @param server server to open a region
   * @param regionOpenInfos info of a list of regions to open
   * @return a list of region opening states
   */
  public List<RegionOpeningState> sendRegionOpen(ServerName server,
      List<Pair<HRegionInfo, List<ServerName>>> regionOpenInfos)
  throws IOException {
    AdminService.BlockingInterface admin = getRsAdmin(server);
    if (admin == null) {
      throw new IOException("Attempting to send OPEN RPC to server " + server.toString() +
        " failed because no RPC connection found to this server");
    }

    OpenRegionRequest request = RequestConverter.buildOpenRegionRequest(server, regionOpenInfos,
      (RecoveryMode.LOG_REPLAY == this.master.getMasterWalManager().getLogRecoveryMode()));
    try {
      OpenRegionResponse response = admin.openRegion(null, request);
      return ResponseConverter.getRegionOpeningStateList(response);
    } catch (ServiceException se) {
      throw ProtobufUtil.getRemoteException(se);
    }
  }

  private PayloadCarryingRpcController newRpcController() {
    return rpcControllerFactory == null ? null : rpcControllerFactory.newController();
  }

  /**
   * Sends an CLOSE RPC to the specified server to close the specified region.
   * <p>
   * A region server could reject the close request because it either does not
   * have the specified region or the region is being split.
   * @param server server to open a region
   * @param region region to open
   * @param dest - if the region is moved to another server, the destination server. null otherwise.
   * @throws IOException
   */
  public boolean sendRegionClose(ServerName server, HRegionInfo region,
      ServerName dest) throws IOException {
    if (server == null) throw new NullPointerException("Passed server is null");
    AdminService.BlockingInterface admin = getRsAdmin(server);
    if (admin == null) {
      throw new IOException("Attempting to send CLOSE RPC to server " +
        server.toString() + " for region " +
        region.getRegionNameAsString() +
        " failed because no RPC connection found to this server");
    }
    PayloadCarryingRpcController controller = newRpcController();
    return ProtobufUtil.closeRegion(controller, admin, server, region.getRegionName(), dest);
  }

  public boolean sendRegionClose(ServerName server,
      HRegionInfo region) throws IOException {
    return sendRegionClose(server, region, null);
  }

  /**
   * Sends a WARMUP RPC to the specified server to warmup the specified region.
   * <p>
   * A region server could reject the close request because it either does not
   * have the specified region or the region is being split.
   * @param server server to warmup a region
   * @param region region to  warmup
   */
  public void sendRegionWarmup(ServerName server,
      HRegionInfo region) {
    if (server == null) return;
    try {
      AdminService.BlockingInterface admin = getRsAdmin(server);
      PayloadCarryingRpcController controller = newRpcController();
      ProtobufUtil.warmupRegion(controller, admin, region);
    } catch (IOException e) {
      LOG.error("Received exception in RPC for warmup server:" +
        server + "region: " + region +
        "exception: " + e);
    }
  }

  /**
   * Contacts a region server and waits up to timeout ms
   * to close the region.  This bypasses the active hmaster.
   */
  public static void closeRegionSilentlyAndWait(ClusterConnection connection,
    ServerName server, HRegionInfo region, long timeout) throws IOException, InterruptedException {
    AdminService.BlockingInterface rs = connection.getAdmin(server);
    PayloadCarryingRpcController controller = connection.getRpcControllerFactory().newController();
    try {
      ProtobufUtil.closeRegion(controller, rs, server, region.getRegionName());
    } catch (IOException e) {
      LOG.warn("Exception when closing region: " + region.getRegionNameAsString(), e);
    }
    long expiration = timeout + System.currentTimeMillis();
    while (System.currentTimeMillis() < expiration) {
      try {
        HRegionInfo rsRegion =
          ProtobufUtil.getRegionInfo(controller, rs, region.getRegionName());
        if (rsRegion == null) return;
      } catch (IOException ioe) {
        if (ioe instanceof NotServingRegionException) // no need to retry again
          return;
        LOG.warn("Exception when retrieving regioninfo from: "
          + region.getRegionNameAsString(), ioe);
      }
      Thread.sleep(1000);
    }
    throw new IOException("Region " + region + " failed to close within"
        + " timeout " + timeout);
  }

  /**
   * Sends an MERGE REGIONS RPC to the specified server to merge the specified
   * regions.
   * <p>
   * A region server could reject the close request because it either does not
   * have the specified region.
   * @param server server to merge regions
   * @param region_a region to merge
   * @param region_b region to merge
   * @param forcible true if do a compulsory merge, otherwise we will only merge
   *          two adjacent regions
   * @throws IOException
   */
  public void sendRegionsMerge(ServerName server, HRegionInfo region_a,
      HRegionInfo region_b, boolean forcible, final User user) throws IOException {
    if (server == null)
      throw new NullPointerException("Passed server is null");
    if (region_a == null || region_b == null)
      throw new NullPointerException("Passed region is null");
    AdminService.BlockingInterface admin = getRsAdmin(server);
    if (admin == null) {
      throw new IOException("Attempting to send MERGE REGIONS RPC to server "
          + server.toString() + " for region "
          + region_a.getRegionNameAsString() + ","
          + region_b.getRegionNameAsString()
          + " failed because no RPC connection found to this server");
    }
    PayloadCarryingRpcController controller = newRpcController();
    ProtobufUtil.mergeRegions(controller, admin, region_a, region_b, forcible, user);
  }

  /**
   * Check if a region server is reachable and has the expected start code
   */
  public boolean isServerReachable(ServerName server) {
    if (server == null) throw new NullPointerException("Passed server is null");


    RetryCounter retryCounter = pingRetryCounterFactory.create();
    while (retryCounter.shouldRetry()) {
      try {
        PayloadCarryingRpcController controller = newRpcController();
        AdminService.BlockingInterface admin = getRsAdmin(server);
        if (admin != null) {
          ServerInfo info = ProtobufUtil.getServerInfo(controller, admin);
          return info != null && info.hasServerName()
            && server.getStartcode() == info.getServerName().getStartCode();
        }
      } catch (IOException ioe) {
        LOG.debug("Couldn't reach " + server + ", try=" + retryCounter.getAttemptTimes()
          + " of " + retryCounter.getMaxAttempts(), ioe);
        try {
          retryCounter.sleepUntilNextRetry();
        } catch(InterruptedException ie) {
          Thread.currentThread().interrupt();
        }
      }
    }
    return false;
  }

    /**
    * @param sn
    * @return Admin interface for the remote regionserver named <code>sn</code>
    * @throws IOException
    * @throws RetriesExhaustedException wrapping a ConnectException if failed
    */
  private AdminService.BlockingInterface getRsAdmin(final ServerName sn)
  throws IOException {
    AdminService.BlockingInterface admin = this.rsAdmins.get(sn);
    if (admin == null) {
      LOG.debug("New admin connection to " + sn.toString());
      if (sn.equals(master.getServerName()) && master instanceof HRegionServer) {
        // A master is also a region server now, see HBASE-10569 for details
        admin = ((HRegionServer)master).getRSRpcServices();
      } else {
        admin = this.connection.getAdmin(sn);
      }
      this.rsAdmins.put(sn, admin);
    }
    return admin;
  }

  /**
   * Wait for the region servers to report in.
   * We will wait until one of this condition is met:
   *  - the master is stopped
   *  - the 'hbase.master.wait.on.regionservers.maxtostart' number of
   *    region servers is reached
   *  - the 'hbase.master.wait.on.regionservers.mintostart' is reached AND
   *   there have been no new region server in for
   *      'hbase.master.wait.on.regionservers.interval' time AND
   *   the 'hbase.master.wait.on.regionservers.timeout' is reached
   *
   * @throws InterruptedException
   */
  public void waitForRegionServers(MonitoredTask status)
  throws InterruptedException {
    final long interval = this.master.getConfiguration().
      getLong(WAIT_ON_REGIONSERVERS_INTERVAL, 1500);
    final long timeout = this.master.getConfiguration().
      getLong(WAIT_ON_REGIONSERVERS_TIMEOUT, 4500);
    int defaultMinToStart = 1;
    if (BaseLoadBalancer.tablesOnMaster(master.getConfiguration())) {
      // If we assign regions to master, we'd like to start
      // at least another region server so that we don't
      // assign all regions to master if other region servers
      // don't come up in time.
      defaultMinToStart = 2;
    }
    int minToStart = this.master.getConfiguration().
      getInt(WAIT_ON_REGIONSERVERS_MINTOSTART, defaultMinToStart);
    if (minToStart < 1) {
      LOG.warn(String.format(
        "The value of '%s' (%d) can not be less than 1, ignoring.",
        WAIT_ON_REGIONSERVERS_MINTOSTART, minToStart));
      minToStart = 1;
    }
    int maxToStart = this.master.getConfiguration().
      getInt(WAIT_ON_REGIONSERVERS_MAXTOSTART, Integer.MAX_VALUE);
    if (maxToStart < minToStart) {
        LOG.warn(String.format(
            "The value of '%s' (%d) is set less than '%s' (%d), ignoring.",
            WAIT_ON_REGIONSERVERS_MAXTOSTART, maxToStart,
            WAIT_ON_REGIONSERVERS_MINTOSTART, minToStart));
        maxToStart = Integer.MAX_VALUE;
    }

    long now =  System.currentTimeMillis();
    final long startTime = now;
    long slept = 0;
    long lastLogTime = 0;
    long lastCountChange = startTime;
    int count = countOfRegionServers();
    int oldCount = 0;
    while (!this.master.isStopped() && count < maxToStart
        && (lastCountChange+interval > now || timeout > slept || count < minToStart)) {
      // Log some info at every interval time or if there is a change
      if (oldCount != count || lastLogTime+interval < now){
        lastLogTime = now;
        String msg =
          "Waiting for region servers count to settle; currently"+
            " checked in " + count + ", slept for " + slept + " ms," +
            " expecting minimum of " + minToStart + ", maximum of "+ maxToStart+
            ", timeout of "+timeout+" ms, interval of "+interval+" ms.";
        LOG.info(msg);
        status.setStatus(msg);
      }

      // We sleep for some time
      final long sleepTime = 50;
      Thread.sleep(sleepTime);
      now =  System.currentTimeMillis();
      slept = now - startTime;

      oldCount = count;
      count = countOfRegionServers();
      if (count != oldCount) {
        lastCountChange = now;
      }
    }

    LOG.info("Finished waiting for region servers count to settle;" +
      " checked in " + count + ", slept for " + slept + " ms," +
      " expecting minimum of " + minToStart + ", maximum of "+ maxToStart+","+
      " master is "+ (this.master.isStopped() ? "stopped.": "running")
    );
  }

  /**
   * @return A copy of the internal list of online servers.
   */
  public List<ServerName> getOnlineServersList() {
    // TODO: optimize the load balancer call so we don't need to make a new list
    // TODO: FIX. THIS IS POPULAR CALL.
    return new ArrayList<ServerName>(this.onlineServers.keySet());
  }

  /**
   * @return A copy of the internal list of draining servers.
   */
  public List<ServerName> getDrainingServersList() {
    return new ArrayList<ServerName>(this.drainingServers);
  }

  /**
   * @return A copy of the internal set of deadNotExpired servers.
   */
  Set<ServerName> getDeadNotExpiredServers() {
    return new HashSet<ServerName>(this.queuedDeadServers);
  }

  /**
   * During startup, if we figure it is not a failover, i.e. there is
   * no more WAL files to split, we won't try to recover these dead servers.
   * So we just remove them from the queue. Use caution in calling this.
   */
  void removeRequeuedDeadServers() {
    requeuedDeadServers.clear();
  }

  /**
   * @return A copy of the internal map of requeuedDeadServers servers and their corresponding
   *         splitlog need flag.
   */
  Map<ServerName, Boolean> getRequeuedDeadServers() {
    return Collections.unmodifiableMap(this.requeuedDeadServers);
  }

  public boolean isServerOnline(ServerName serverName) {
    return serverName != null && onlineServers.containsKey(serverName);
  }

  /**
   * Check if a server is known to be dead.  A server can be online,
   * or known to be dead, or unknown to this manager (i.e, not online,
   * not known to be dead either. it is simply not tracked by the
   * master any more, for example, a very old previous instance).
   */
  public synchronized boolean isServerDead(ServerName serverName) {
    return serverName == null || deadservers.isDeadServer(serverName)
      || queuedDeadServers.contains(serverName)
      || requeuedDeadServers.containsKey(serverName);
  }

  public void shutdownCluster() {
    this.clusterShutdown = true;
    this.master.stop("Cluster shutdown requested");
  }

  public boolean isClusterShutdown() {
    return this.clusterShutdown;
  }

  /**
   * Stop the ServerManager.  Currently closes the connection to the master.
   */
  public void stop() {
    if (connection != null) {
      try {
        connection.close();
      } catch (IOException e) {
        LOG.error("Attempt to close connection to master failed", e);
      }
    }
  }

  /**
   * Creates a list of possible destinations for a region. It contains the online servers, but not
   *  the draining or dying servers.
   *  @param serverToExclude can be null if there is no server to exclude
   */
  public List<ServerName> createDestinationServersList(final ServerName serverToExclude){
    final List<ServerName> destServers = getOnlineServersList();

    if (serverToExclude != null){
      destServers.remove(serverToExclude);
    }

    // Loop through the draining server list and remove them from the server list
    final List<ServerName> drainingServersCopy = getDrainingServersList();
    if (!drainingServersCopy.isEmpty()) {
      for (final ServerName server: drainingServersCopy) {
        destServers.remove(server);
      }
    }

    // Remove the deadNotExpired servers from the server list.
    removeDeadNotExpiredServers(destServers);
    return destServers;
  }

  /**
   * Calls {@link #createDestinationServersList} without server to exclude.
   */
  public List<ServerName> createDestinationServersList(){
    return createDestinationServersList(null);
  }

    /**
    * Loop through the deadNotExpired server list and remove them from the
    * servers.
    * This function should be used carefully outside of this class. You should use a high level
    *  method such as {@link #createDestinationServersList()} instead of managing you own list.
    */
  void removeDeadNotExpiredServers(List<ServerName> servers) {
    Set<ServerName> deadNotExpiredServersCopy = this.getDeadNotExpiredServers();
    if (!deadNotExpiredServersCopy.isEmpty()) {
      for (ServerName server : deadNotExpiredServersCopy) {
        LOG.debug("Removing dead but not expired server: " + server
          + " from eligible server pool.");
        servers.remove(server);
      }
    }
  }

  /**
   * To clear any dead server with same host name and port of any online server
   */
  void clearDeadServersWithSameHostNameAndPortOfOnlineServer() {
    for (ServerName serverName : getOnlineServersList()) {
      deadservers.cleanAllPreviousInstances(serverName);
    }
  }
}