content: /**
   * 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 com.alibaba.wasp.master;
  
  import com.alibaba.wasp.ClockOutOfSyncException;
  import com.alibaba.wasp.EntityGroupInfo;
  import com.alibaba.wasp.PleaseHoldException;
  import com.alibaba.wasp.Server;
  import com.alibaba.wasp.ServerLoad;
  import com.alibaba.wasp.ServerName;
  import com.alibaba.wasp.YouAreDeadException;
  import com.alibaba.wasp.ZooKeeperConnectionException;
  import com.alibaba.wasp.client.FConnection;
  import com.alibaba.wasp.client.FConnectionManager;
  import com.alibaba.wasp.fserver.AdminProtocol;
  import com.alibaba.wasp.fserver.EntityGroupOpeningState;
  import com.alibaba.wasp.master.handler.ServerShutdownHandler;
  import com.alibaba.wasp.monitoring.MonitoredTask;
  import com.alibaba.wasp.protobuf.ProtobufUtil;
  import com.alibaba.wasp.protobuf.RequestConverter;
  import com.alibaba.wasp.protobuf.ResponseConverter;
  import com.alibaba.wasp.protobuf.generated.FServerAdminProtos.OpenEntityGroupRequest;
  import com.alibaba.wasp.protobuf.generated.FServerAdminProtos.OpenEntityGroupResponse;
  import com.google.protobuf.ServiceException;
  import org.apache.commons.logging.Log;
  import org.apache.commons.logging.LogFactory;
  import org.apache.hadoop.conf.Configuration;
  
  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.Set;
  import java.util.concurrent.ConcurrentHashMap;
  
  /**
   * The FServerManager class manages info about fservers.
   * <p>
   * Maintains lists of online and dead servers. Processes the startups,
   * shutdowns, and deaths of fservers.
   * <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.
   */
  public class FServerManager {
    private static final Log LOG = LogFactory.getLog(FServerManager.class);
  
    // Set if we are to shutdown the cluster.
    private volatile boolean clusterShutdown = false;
  
    /** Map of registered servers to their current load */
    private final Map<ServerName, ServerLoad> onlineServers = new ConcurrentHashMap<ServerName, ServerLoad>();
  
    // TODO: This is strange to have two maps but HSI above is used on both sides
    /**
     * Map from full server-instance name to the RPC connection for this server.
     */
    private final Map<ServerName, AdminProtocol> serverConnections = new HashMap<ServerName, AdminProtocol>();
  
    /**
     * List of fservers <ServerName> that should not get any more new
     * entityGroups.
     */
    private final ArrayList<ServerName> drainingServers = new ArrayList<ServerName>();
  
    private final Server master;
    private final FMasterServices services;
    private final FConnection connection;
  
    private final DeadServer deadservers;
  
    private final long maxSkew;
    private final long warningSkew;
  
    /**
     * Set of fservers 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 org.apache.hadoop.hbase.master.ServerManager#processQueuedDeadServers()} by master.
     * <p>
     * A dead server is a server instance known to be dead, not listed in the
     * /wasp/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 fservers known to be dead but not submitted to
     * ServerShutdownHander for processing yet.
     */
    private Set<ServerName> queuedDeadServers = new HashSet<ServerName>();
  
    /**
     * Set of fservers 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 org.apache.hadoop.hbase.master.ServerManager#processQueuedDeadServers()} by assignment manager.
     * <p>
     * For all the fservers in this set, HLog split is already completed.
     * <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 root/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 Set<ServerName> requeuedDeadServers = new HashSet<ServerName>();
  
    /**
     * Constructor.
     *
     * @param master
     * @param services
     * @throws com.alibaba.wasp.ZooKeeperConnectionException
     */
    public FServerManager(final Server master, final FMasterServices services)
        throws ZooKeeperConnectionException {
      this(master, services, true);
    }
  
    FServerManager(final Server master, final FMasterServices services,
        final boolean connect) throws ZooKeeperConnectionException {
      this.master = master;
      this.services = services;
      Configuration c = master.getConfiguration();
      maxSkew = c.getLong("wasp.master.maxclockskew", 30000);
      warningSkew = c.getLong("wasp.master.warningclockskew", 10000);
      this.deadservers = new DeadServer();
      this.connection = connect ? FConnectionManager.getConnection(c) : null;
    }
  
    /**
     * Let the server manager know a new fserver has come online
     *
     * @param ia
     *          The remote address
     * @param port
     *          The remote port
     * @param serverStartcode
     * @param serverCurrentTime
     *          The current time of the fserver in ms
     * @return The ServerName we know this server as.
     * @throws java.io.IOException
     */
    ServerName fserverStartup(final InetAddress ia, final int port,
        final long serverStartcode, long serverCurrentTime) throws IOException {
      // Test for case where we get a entityGroup startup message from a fserver
      // 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.
      ServerName sn = new ServerName(ia.getHostName(), port, serverStartcode);
      checkClockSkew(sn, serverCurrentTime);
      checkIsDead(sn, "STARTUP");
      checkAlreadySameHostPort(sn);
      recordNewServer(sn, ServerLoad.EMPTY_SERVERLOAD);
      return sn;
    }
  
    void fserverReport(ServerName sn, ServerLoad sl) throws YouAreDeadException,
        PleaseHoldException {
      checkIsDead(sn, "REPORT");
      if (!this.onlineServers.containsKey(sn)) {
        // Already have this host+port combo and its just different start code?
        checkAlreadySameHostPort(sn);
        // 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 fserver
        // 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.
        recordNewServer(sn, sl);
      } else {
        this.onlineServers.put(sn, sl);
      }
    }
  
    /**
     * Test to see if we have a server of same host and port already.
     *
     * @param serverName
     * @throws com.alibaba.wasp.PleaseHoldException
     */
    void checkAlreadySameHostPort(final ServerName serverName)
        throws PleaseHoldException {
      ServerName existingServer = ServerName.findServerWithSameHostnamePort(
          getOnlineServersList(), serverName);
      if (existingServer != null) {
        String message = "Server serverName=" + serverName
            + " rejected; we already have " + existingServer.toString()
            + " registered with same hostname and port";
        LOG.info(message);
        if (existingServer.getStartcode() < serverName.getStartcode()) {
          LOG.info("Triggering server recovery; existingServer " + existingServer
              + " looks stale, new server:" + serverName);
          expireServer(existingServer);
        }
        // master has completed the initialization
        throw new PleaseHoldException(message);
      }
    }
  
    /**
     * 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 com.alibaba.wasp.ClockOutOfSyncException
     *           if the skew exceeds the configured max value
     */
    private void checkClockSkew(final ServerName serverName,
        final long serverCurrentTime) throws ClockOutOfSyncException {
      long skew = 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 com.alibaba.wasp.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.
      if ((this.services == null || ((FMaster) this.services).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");
      }
    }
  
    /**
     * Adds the onlineServers list.
     *
     * @param hsl
     * @param serverName
     *          The remote servers name.
     */
    void recordNewServer(final ServerName serverName, final ServerLoad sl) {
      LOG.info("Registering server=" + serverName);
      this.onlineServers.put(serverName, sl);
      this.serverConnections.remove(serverName);
    }
  
    /**
     * @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 fservers. Currently, this uses a very
     * naive computation - just uses the number of entityGroups being served,
     * ignoring stats about number of requests.
     *
     * @return the average load
     */
    public double getAverageLoad() {
      int totalLoad = 0;
      int numServers = 0;
      double averageLoad = 0.0;
      for (ServerLoad sl : this.onlineServers.values()) {
        numServers++;
        totalLoad += sl.getNumberOfEntityGroups();
      }
      averageLoad = (double) totalLoad / (double) numServers;
      return averageLoad;
    }
  
    /** @return the count of active fservers */
    int countOfFServers() {
      // 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 Set<ServerName> getDeadServers() {
      return this.deadservers.clone();
    }
  
    /**
     * Checks if any dead servers are currently in progress.
     *
     * @return true if any FS are being processed as dead, false if not
     */
    public boolean areDeadServersInProgress() {
      return this.deadservers.areDeadServersInProgress();
    }
  
    void letFServersShutdown() {
      long previousLogTime = 0;
      while (!onlineServers.isEmpty()) {
  
        if (System.currentTimeMillis() > (previousLogTime + 1000)) {
          StringBuilder sb = new StringBuilder();
          for (ServerName key : this.onlineServers.keySet()) {
            if (sb.length() > 0) {
              sb.append(", ");
            }
            sb.append(key);
          }
          LOG.info("Waiting on fserver(s) to go down " + sb.toString());
          previousLogTime = System.currentTimeMillis();
        }
  
        synchronized (onlineServers) {
          try {
            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 (!this.onlineServers.containsKey(serverName)) {
        LOG.warn("Received expiration of " + serverName
            + " but server is not currently online");
      }
      if (this.deadservers.contains(serverName)) {
        // TODO: Can this happen? It shouldn't be online in this case?
        LOG.warn("Received expiration of " + serverName
            + " but server shutdown is already in progress");
        return;
      }
      // 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(serverName);
      this.onlineServers.remove(serverName);
      synchronized (onlineServers) {
        onlineServers.notifyAll();
      }
      this.serverConnections.remove(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;
      }
      this.services.getExecutorService().submit(
          new ServerShutdownHandler(this.master, this.services, this.deadservers,
              serverName));
      LOG.debug("Added=" + serverName
          + " to dead servers, submitted shutdown handler to be executed");
    }
  
    public synchronized void processDeadServer(final ServerName serverName) {
      // When assignment manager is cleaning up the zookeeper nodes and rebuilding
      // the
      // in-memory entityGroup states, fservers could be down. Root/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 entityGroups.
      //
      // We should not wait in the server shutdown handler thread since it can
      // clog
      // the handler threads and root/meta table could not be re-assigned in case
      // the corresponding server is down. So we queue them up here instead.
      if (!services.getAssignmentManager().isFailoverCleanupDone()) {
        requeuedDeadServers.add(serverName);
        return;
      }
  
      this.deadservers.add(serverName);
      this.services.getExecutorService().submit(
          new ServerShutdownHandler(this.master, this.services, this.deadservers,
              serverName));
    }
  
    /**
     * Process the servers which died during master's initialization. It will be
     * called after HMaster#assignRootAndMeta and AssignmentManager#joinCluster.
     * */
    synchronized void processQueuedDeadServers() {
      Iterator<ServerName> serverIterator = queuedDeadServers.iterator();
      while (serverIterator.hasNext()) {
        expireServer(serverIterator.next());
        serverIterator.remove();
      }
  
      if (!services.getAssignmentManager().isFailoverCleanupDone()) {
        LOG.info("AssignmentManager hasn't finished failover cleanup");
      }
      serverIterator = requeuedDeadServers.iterator();
      while (serverIterator.hasNext()) {
        processDeadServer(serverIterator.next());
        serverIterator.remove();
      }
    }
  
    /*
     * 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 fservers
  
    /**
     * Sends an OPEN RPC to the specified server to open the specified
     * entityGroup.
     * <p>
     * Open should not fail but can if server just crashed.
     * <p>
     *
     * @param server
     *          server to open a entityGroup
     * @param entityGroup
     *          entityGroup to open
     * @param versionOfOfflineNode
     *          that needs to be present in the offline node when FS tries to
     *          change the state from OFFLINE to other states.
     */
    public EntityGroupOpeningState sendEntityGroupOpen(final ServerName server,
        EntityGroupInfo entityGroup, int versionOfOfflineNode) throws IOException {
      AdminProtocol admin = getServerConnection(server);
      if (admin == null) {
        LOG.warn("Attempting to send OPEN RPC to server " + server.toString()
            + " failed because no RPC connection found to this server");
        return EntityGroupOpeningState.FAILED_OPENING;
      }
      OpenEntityGroupRequest request = RequestConverter
          .buildOpenEntityGroupRequest(entityGroup, versionOfOfflineNode);
      try {
        OpenEntityGroupResponse response = admin.openEntityGroup(null, request);
        return ResponseConverter.getEntityGroupOpeningState(response);
      } catch (ServiceException se) {
        throw ProtobufUtil.getRemoteException(se);
      }
    }
  
    /**
     * Sends an OPEN RPC to the specified server to open the specified
     * entityGroup.
     * <p>
     * Open should not fail but can if server just crashed.
     * <p>
     *
     * @param server
     *          server to open a entityGroup
     * @param entityGroupOpenInfos
     *          info of a list of entityGroups to open
     * @return a list of entityGroup opening states
     */
    public List<EntityGroupOpeningState> sendEntityGroupsOpen(ServerName server,
        List<EntityGroupInfo> entityGroupOpenInfos) throws IOException {
      AdminProtocol admin = getServerConnection(server);
      if (admin == null) {
        LOG.warn("Attempting to send OPEN RPC to server " + server.toString()
            + " failed because no RPC connection found to this server");
        return null;
      }
  
      OpenEntityGroupRequest request = RequestConverter
          .buildOpenEntityGroupRequest(entityGroupOpenInfos);
      try {
        OpenEntityGroupResponse response = admin.openEntityGroup(null, request);
        return ResponseConverter.getEntityGroupOpeningStateList(response);
      } catch (ServiceException se) {
        throw ProtobufUtil.getRemoteException(se);
      }
    }
  
    /**
     * Sends an CLOSE RPC to the specified server to close the specified
     * entityGroup.
     * <p>
     * A entityGroup server could reject the close request because it either does
     * not have the specified entityGroup or the entityGroup is being split.
     *
     * @param server
     *          server to open a entityGroup
     * @param entityGroup
     *          entityGroup to open
     * @param versionOfClosingNode
     *          the version of znode to compare when FS transitions the znode from
     *          CLOSING state.
     * @param dest
     *          - if the entityGroup is moved to another server, the destination
     *          server. null otherwise.
     * @return true if server acknowledged close, false if not
     * @throws java.io.IOException
     */
    public boolean sendEntityGroupClose(ServerName server,
        EntityGroupInfo entityGroup, int versionOfClosingNode, ServerName dest,
        boolean transitionInZK) throws IOException {
      if (server == null) {
        throw new NullPointerException("Passed server is null");
      }
      AdminProtocol admin = getServerConnection(server);
      if (admin == null) {
        throw new IOException("Attempting to send CLOSE RPC to server "
            + server.toString() + " for entityGroup "
            + entityGroup.getEntityGroupNameAsString()
            + " failed because no RPC connection found to this server");
      }
      return ProtobufUtil.closeEntityGroup(admin, entityGroup,
          versionOfClosingNode, transitionInZK);
    }
  
    public boolean sendEntityGroupClose(ServerName server,
        EntityGroupInfo entityGroup, int versionOfClosingNode) throws IOException {
      return sendEntityGroupClose(server, entityGroup, versionOfClosingNode,
          null, true);
    }
  
    /**
     * Sends an ENABLE TABLE RPC to the specified server to close the specified
     * entityGroup.
     *
     * @param server
     *          server to disable a table.
     * @return true if server acknowledged close, false if not
     * @throws java.io.IOException
     */
    public boolean sendEnableTable(ServerName server, String tableName)
        throws IOException {
      if (server == null) {
        throw new NullPointerException("Passed server is null");
      }
      AdminProtocol admin = getServerConnection(server);
      if (admin == null) {
        throw new IOException("Attempting to send enableTable RPC to server "
            + server.toString()
            + " failed because no RPC connection found to this server");
      }
  
      com.alibaba.wasp.protobuf.generated.FServerAdminProtos.EnableTableRequest request = RequestConverter
          .buildServerEnableTableRequest(tableName);
      try {
        return admin.enableServerTable(null, request).getSuccess();
      } catch (ServiceException e) {
        throw ProtobufUtil.getRemoteException(e);
      }
    }
  
    /**
     * Sends an DISABLE TABLE RPC to the specified server to close the specified
     * entityGroup.
     *
     * @param server
     *          server to disable a table.
     * @return true if server acknowledged close, false if not
     * @throws java.io.IOException
     */
    public boolean sendDisableTable(ServerName server, String tableName)
        throws IOException {
      if (server == null) {
        throw new NullPointerException("Passed server is null");
      }
      AdminProtocol admin = getServerConnection(server);
      if (admin == null) {
        throw new IOException("Attempting to send disableTable RPC to server "
            + server.toString()
            + " failed because no RPC connection found to this server");
      }
  
      com.alibaba.wasp.protobuf.generated.FServerAdminProtos.DisableTableRequest request = RequestConverter
          .buildServerDisableTableRequest(tableName);
      try {
        return admin.disableServerTable(null, request).getSuccess();
      } catch (ServiceException e) {
        throw ProtobufUtil.getRemoteException(e);
      }
    }
  
    /**
     * @param sn
     * @return
     * @throws java.io.IOException
     * @throws com.alibaba.wasp.client.RetriesExhaustedException
     *           wrapping a ConnectException if failed putting up proxy.
     */
    private AdminProtocol getServerConnection(final ServerName sn)
        throws IOException {
      AdminProtocol admin = this.serverConnections.get(sn);
      if (admin == null) {
        LOG.debug("New connection to " + sn.toString());
        admin = this.connection.getAdmin(sn.getHostname(), sn.getPort());
        this.serverConnections.put(sn, admin);
      }
      return admin;
    }
  
    /**
     * Wait for the fservers to report in. We will wait until one of this
     * condition is met: - the master is stopped - the
     * 'wasp.master.wait.on.fservers.timeout' is reached - the
     * 'wasp.master.wait.on.fservers.maxtostart' number of fservers is reached -
     * the 'wasp.master.wait.on.fservers.mintostart' is reached AND there have
     * been no new fserver in for 'wasp.master.wait.on.fservers.interval' time
     * 
     * @throws InterruptedException
     */
    public void waitForFServers(MonitoredTask status) throws InterruptedException {
      final long interval = this.master.getConfiguration().getLong(
          "wasp.master.wait.on.fservers.interval", 1500);
      final long timeout = this.master.getConfiguration().getLong(
          "wasp.master.wait.on.fservers.timeout", 4500);
      final int minToStart = this.master.getConfiguration().getInt(
          "wasp.master.wait.on.fservers.mintostart", 1);
      final int maxToStart = this.master.getConfiguration().getInt(
          "wasp.master.wait.on.fservers.maxtostart", Integer.MAX_VALUE);
  
      long now = System.currentTimeMillis();
      final long startTime = now;
      long slept = 0;
      long lastLogTime = 0;
      long lastCountChange = startTime;
      int count = countOfFServers();
      int oldCount = 0;
      while (!this.master.isStopped() && slept < timeout && count < maxToStart
          && (lastCountChange + interval > now || 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 fservers 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 = countOfFServers();
        if (count != oldCount) {
          lastCountChange = now;
        }
      }
  
      LOG.info("Finished waiting for fservers 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);
    }
  
    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.contains(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 entityGroup. 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() {
      ServerName sn = null;
      for (ServerName serverName : getOnlineServersList()) {
        while ((sn = ServerName.findServerWithSameHostnamePort(this.deadservers,
            serverName)) != null) {
          this.deadservers.remove(sn);
        }
      }
    }
  
  }
  