/*
 * 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.geode.distributed;

import static java.lang.System.lineSeparator;
import static org.apache.commons.lang3.StringUtils.EMPTY;
import static org.apache.commons.lang3.StringUtils.defaultIfBlank;
import static org.apache.commons.lang3.StringUtils.isBlank;
import static org.apache.commons.lang3.StringUtils.isNotBlank;
import static org.apache.commons.lang3.StringUtils.lowerCase;
import static org.apache.geode.distributed.ConfigurationProperties.LOG_FILE;
import static org.apache.geode.distributed.ConfigurationProperties.NAME;
import static org.apache.geode.distributed.ConfigurationProperties.SERVER_BIND_ADDRESS;
import static org.apache.geode.internal.lang.StringUtils.wrap;
import static org.apache.geode.internal.lang.SystemUtils.CURRENT_DIRECTORY;
import static org.apache.geode.internal.process.ProcessLauncherContext.OVERRIDDEN_DEFAULTS_PREFIX;
import static org.apache.geode.internal.util.IOUtils.tryGetCanonicalPathElseGetAbsolutePath;

import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.lang.management.ManagementFactory;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.ServiceLoader;
import java.util.TreeMap;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.function.Supplier;

import javax.management.MalformedObjectNameException;
import javax.management.ObjectName;

import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import joptsimple.OptionException;
import joptsimple.OptionParser;
import joptsimple.OptionSet;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.apache.logging.log4j.Logger;

import org.apache.geode.SystemFailure;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.cache.Cache;
import org.apache.geode.cache.Region;
import org.apache.geode.cache.control.ResourceManager;
import org.apache.geode.cache.partition.PartitionRegionHelper;
import org.apache.geode.cache.server.CacheServer;
import org.apache.geode.distributed.internal.DefaultServerLauncherCacheProvider;
import org.apache.geode.distributed.internal.InternalDistributedSystem;
import org.apache.geode.internal.DistributedSerializableObjectConfig;
import org.apache.geode.internal.GemFireVersion;
import org.apache.geode.internal.cache.AbstractCacheServer;
import org.apache.geode.internal.cache.CacheConfig;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.PartitionedRegion;
import org.apache.geode.internal.cache.control.InternalResourceManager;
import org.apache.geode.internal.cache.tier.sockets.CacheServerHelper;
import org.apache.geode.internal.inet.LocalHostUtil;
import org.apache.geode.internal.lang.ObjectUtils;
import org.apache.geode.internal.process.ConnectionFailedException;
import org.apache.geode.internal.process.ControlNotificationHandler;
import org.apache.geode.internal.process.ControllableProcess;
import org.apache.geode.internal.process.FileAlreadyExistsException;
import org.apache.geode.internal.process.FileControllableProcess;
import org.apache.geode.internal.process.MBeanInvocationFailedException;
import org.apache.geode.internal.process.PidFile;
import org.apache.geode.internal.process.PidUnavailableException;
import org.apache.geode.internal.process.ProcessController;
import org.apache.geode.internal.process.ProcessControllerFactory;
import org.apache.geode.internal.process.ProcessControllerParameters;
import org.apache.geode.internal.process.ProcessLauncherContext;
import org.apache.geode.internal.process.ProcessType;
import org.apache.geode.internal.process.UnableToControlProcessException;
import org.apache.geode.internal.serialization.filter.SystemPropertyGlobalSerialFilterConfigurationFactory;
import org.apache.geode.internal.serialization.filter.UnableToSetSerialFilterException;
import org.apache.geode.internal.util.ArgumentRedactor;
import org.apache.geode.lang.AttachAPINotFoundException;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.management.internal.i18n.CliStrings;
import org.apache.geode.management.internal.util.HostUtils;
import org.apache.geode.management.internal.util.JsonUtil;
import org.apache.geode.pdx.PdxSerializer;
import org.apache.geode.security.AuthenticationRequiredException;
import org.apache.geode.security.GemFireSecurityException;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * The ServerLauncher class is a launcher class with main method to start a Geode Server (implying
 * a Geode Cache Server process).
 *
 * @see AbstractLauncher
 * @see LocatorLauncher
 * @since GemFire 7.0
 */
@SuppressWarnings("unused")
public class ServerLauncher extends AbstractLauncher<String> {

  private static final Logger log = LogService.getLogger();

  @Immutable
  private static final Map<String, String> helpMap;

  static {
    Map<String, String> help = new HashMap<>();
    help.put("launcher",
        "A Geode launcher used to start, stop and determine a Server's status.");
    help.put(Command.START.getName(), String.format(
        "Starts a Server running in the current working directory listening on the default port (%s) bound to all IP addresses available to the localhost.  The Server must be given a member name in the Geode cluster.  The default server-bind-address and server-port may be overridden using the corresponding command-line options.",
        getDefaultServerPort()));
    help.put(Command.STATUS.getName(),
        "Displays the status of a Server given any combination of the member name/ID, PID, or the directory in which the Server is running.");
    help.put(Command.STOP.getName(),
        "Stops a running Server given given a member name/ID, PID, or the directory in which the Server is running.");
    help.put(Command.VERSION.getName(),
        "Displays Geode product version information.");
    help.put("assign-buckets",
        "Causes buckets to be assigned to the partitioned regions in the Geode cache on Server start.");
    help.put("debug", "Displays verbose information during the invocation of the launcher.");
    help.put("delete-pid-file-on-stop",
        "Specifies that this Server's PID file should be deleted on stop.  The default is to not delete this Server's PID file until JVM exit if --delete-pid-file-on-stop is not specified.");
    help.put("dir",
        "Specifies the working directory where the Server is running.  Defaults to the current working directory.");
    help.put("disable-default-server",
        "Disables the addition of a default Geode cache server.");
    help.put("force",
        "Enables any existing Server PID file to be overwritten on start.  The default is to throw an error if a PID file already exists and --force is not specified.");
    help.put("help",
        "Causes Geode to print out information instead of performing the command. This option is supported by all commands.");
    help.put("member", "Identifies the Server by member name or ID in the Geode cluster.");
    help.put("pid", "Indicates the OS process ID of the running Server.");
    help.put("rebalance",
        "An option to cause the Geode cache's partitioned regions to be rebalanced on start.");
    help.put("redirect-output",
        "An option to cause the Server to redirect standard out and standard error to the Geode log file.");
    help.put(SERVER_BIND_ADDRESS,
        "Specifies the IP address on which to bind, or on which the Server is bound, listening for client requests.  Defaults to all IP addresses available to the localhost.");
    help.put("hostname-for-clients",
        "An option to specify the hostname or IP address to send to clients so they can connect to this Server. The default is to use the IP address to which the Server is bound.");
    help.put("server-port", String.format(
        "Specifies the port on which the Server is listening for client requests. Defaults to %s.",
        getDefaultServerPort()));
    helpMap = Collections.unmodifiableMap(help);
  }

  @Immutable
  private static final Map<Command, String> usageMap;

  static {
    Map<Command, String> usage = new TreeMap<>();
    usage.put(Command.START,
        "start <member-name> [--assign-buckets] [--disable-default-server] [--rebalance] [--server-bind-address=<IP-address>] [--server-port=<port>] [--force] [--debug] [--help]");
    usage.put(Command.STATUS,
        "status [--member=<member-ID/Name>] [--pid=<process-ID>] [--dir=<Server-working-directory>] [--debug] [--help]");
    usage.put(Command.STOP,
        "stop [--member=<member-ID/Name>] [--pid=<process-ID>] [--dir=<Server-working-directory>] [--debug] [--help]");
    usage.put(Command.VERSION, "version");
    usageMap = Collections.unmodifiableMap(usage);
  }

  private static final String DEFAULT_SERVER_LOG_EXT = ".log";
  private static final String DEFAULT_SERVER_LOG_NAME = "gemfire";
  private static final String SERVER_SERVICE_NAME = "Server";

  @MakeNotStatic
  private static final AtomicReference<ServerLauncher> INSTANCE = new AtomicReference<>();

  @Immutable
  private static final ServerLauncherCacheProvider DEFAULT_CACHE_PROVIDER =
      new DefaultServerLauncherCacheProvider();

  private static final Logger logger = LogService.getLogger();
  private volatile boolean debug;

  private final ControlNotificationHandler controlHandler;

  private final AtomicBoolean starting = new AtomicBoolean(false);

  private final boolean assignBuckets;
  private final boolean deletePidFileOnStop;
  private final boolean disableDefaultServer;
  private final boolean force;
  private final boolean help;
  private final boolean rebalance;
  private final boolean redirectOutput;

  private volatile Cache cache;

  private final CacheConfig cacheConfig;

  private final Command command;

  private final InetAddress serverBindAddress;

  private final Integer pid;
  private final Integer serverPort;

  private final Properties distributedSystemProperties;

  private final String memberName;
  private final String springXmlLocation;
  private final String workingDirectory;

  // NOTE in addition to debug, the other shared, mutable state
  private volatile String statusMessage;

  private final Float criticalHeapPercentage;
  private final Float evictionHeapPercentage;

  private final Float criticalOffHeapPercentage;
  private final Float evictionOffHeapPercentage;

  private final String hostNameForClients;
  private final Integer maxConnections;
  private final Integer maxMessageCount;
  private final Integer messageTimeToLive;
  private final Integer socketBufferSize;

  private final Integer maxThreads;

  private volatile ControllableProcess process;

  private final ServerControllerParameters controllerParameters;
  private final Runnable startupCompletionAction;
  private final Consumer<Throwable> startupExceptionAction;
  private final ServerLauncherCacheProvider serverLauncherCacheProvider;
  private final Supplier<ControllableProcess> controllableProcessFactory;

  /**
   * Launches a Geode Server from the command-line configured with the given arguments.
   *
   * @param args the command-line arguments used to configure the Geode Server at runtime.
   */
  public static void main(final String... args) {
    try {
      new Builder(args).build().run();
    } catch (AttachAPINotFoundException handled) {
      System.err.println(handled.getMessage());
    }
  }

  private static Integer getDefaultServerPort() {
    return Integer.getInteger(AbstractCacheServer.TEST_OVERRIDE_DEFAULT_PORT_PROPERTY,
        CacheServer.DEFAULT_PORT);
  }

  /**
   * Gets the instance of the ServerLauncher used to launch the Geode Cache Server, or null if
   * this VM does not have an instance of ServerLauncher indicating no Geode Cache Server is
   * running.
   *
   * @return the instance of ServerLauncher used to launcher a Geode Cache Server in this VM.
   */
  public static ServerLauncher getInstance() {
    return INSTANCE.get();
  }

  /**
   * Gets the ServerState for this process or null if this process was not launched using this VM's
   * ServerLauncher reference .
   *
   * @return the ServerState for this process or null.
   */
  public static ServerState getServerState() {
    return getInstance() != null ? getInstance().status() : null;
  }

  /**
   * Private constructor used to properly construct an immutable instance of the ServerLauncher
   * using a Builder. The Builder is used to configure a ServerLauncher instance. The Builder can
   * process user input from the command-line or be used programmatically to properly construct an
   * instance of the ServerLauncher using the API.
   *
   * @param builder an instance of ServerLauncher.Builder for configuring and constructing an
   *        instance of the ServerLauncher.
   * @see ServerLauncher.Builder
   */
  private ServerLauncher(final Builder builder) {
    cache = builder.getCache();
    cacheConfig = builder.getCacheConfig();
    command = builder.getCommand();
    assignBuckets = Boolean.TRUE.equals(builder.getAssignBuckets());
    setDebug(Boolean.TRUE.equals(builder.getDebug()));
    deletePidFileOnStop = Boolean.TRUE.equals(builder.getDeletePidFileOnStop());
    disableDefaultServer = Boolean.TRUE.equals(builder.getDisableDefaultServer());
    distributedSystemProperties = builder.getDistributedSystemProperties();
    force = Boolean.TRUE.equals(builder.getForce());
    help = Boolean.TRUE.equals(builder.getHelp());
    hostNameForClients = builder.getHostNameForClients();
    memberName = builder.getMemberName();
    pid = builder.getPid();
    rebalance = Boolean.TRUE.equals(builder.getRebalance());
    redirectOutput = Boolean.TRUE.equals(builder.getRedirectOutput());
    serverBindAddress = builder.getServerBindAddress();
    serverPort = builder.getServerPort();
    springXmlLocation = builder.getSpringXmlLocation();
    workingDirectory = builder.getWorkingDirectory();
    criticalHeapPercentage = builder.getCriticalHeapPercentage();
    evictionHeapPercentage = builder.getEvictionHeapPercentage();
    criticalOffHeapPercentage = builder.getCriticalOffHeapPercentage();
    evictionOffHeapPercentage = builder.getEvictionOffHeapPercentage();
    maxConnections = builder.getMaxConnections();
    maxMessageCount = builder.getMaxMessageCount();
    maxThreads = builder.getMaxThreads();
    messageTimeToLive = builder.getMessageTimeToLive();
    socketBufferSize = builder.getSocketBufferSize();
    controllerParameters = new ServerControllerParameters();
    startupCompletionAction = builder.getStartupCompletionAction();
    startupExceptionAction = builder.getStartupExceptionAction();
    controlHandler = new ControlNotificationHandler() {
      @Override
      public void handleStop() {
        if (isStoppable()) {
          stopInProcess();
        }
      }

      @Override
      public ServiceState<?> handleStatus() {
        return statusInProcess();
      }
    };
    serverLauncherCacheProvider = builder.getServerLauncherCacheProvider();
    controllableProcessFactory = builder.getControllableProcessFactory();

    Integer serverPort =
        builder.isServerPortSetByUser() && this.serverPort != null ? this.serverPort : null;
    String serverBindAddress =
        builder.isServerBindAddressSetByUser() && this.serverBindAddress != null
            ? this.serverBindAddress.getHostAddress() : null;

    ServerLauncherParameters.INSTANCE
        .withPort(serverPort)
        .withMaxThreads(maxThreads)
        .withBindAddress(serverBindAddress)
        .withMaxConnections(maxConnections)
        .withMaxMessageCount(maxMessageCount)
        .withSocketBufferSize(socketBufferSize)
        .withMessageTimeToLive(messageTimeToLive)
        .withHostnameForClients(hostNameForClients)
        .withDisableDefaultServer(disableDefaultServer);
  }

  /**
   * Gets a reference to the {@code Cache} that was created by this {@code ServerLauncher}.
   *
   * @return a reference to the Cache
   * @see Cache
   */
  public Cache getCache() {
    if (cache != null) {
      boolean isReconnecting = cache.isReconnecting();
      if (isReconnecting) {
        Cache newCache = cache.getReconnectedCache();
        if (newCache != null) {
          cache = newCache;
        }
      }
    }
    return cache;
  }

  /**
   * Gets the CacheConfig object used to configure additional Geode Cache components and features
   * (e.g. PDX).
   *
   * @return a CacheConfig object with additional Geode Cache configuration meta-data used on
   *         startup to configure the Cache.
   */
  public CacheConfig getCacheConfig() {
    final CacheConfig copy = new CacheConfig();
    copy.setDeclarativeConfig(cacheConfig);
    return copy;
  }

  /**
   * Gets an identifier that uniquely identifies and represents the Server associated with this
   * launcher.
   *
   * @return a String value identifier to uniquely identify the Server and it's launcher.
   * @see #getServerBindAddressAsString()
   * @see #getServerPortAsString()
   */
  public String getId() {
    final StringBuilder buffer = new StringBuilder(ServerState.getServerBindAddressAsString(this));
    final String serverPort = ServerState.getServerPortAsString(this);

    if (isNotBlank(serverPort)) {
      buffer.append("[").append(serverPort).append("]");
    }

    return buffer.toString();
  }

  /**
   * Get the Server launcher command used to invoke the Server.
   *
   * @return the Server launcher command used to invoke the Server.
   * @see ServerLauncher.Command
   */
  public Command getCommand() {
    return command;
  }

  /**
   * Determines whether buckets should be assigned to partitioned regions in the cache upon Server
   * start.
   *
   * @return a boolean indicating if buckets should be assigned upon Server start.
   */
  public boolean isAssignBuckets() {
    return assignBuckets;
  }

  /**
   * Determines whether a default cache server will be added when the Geode Server comes online.
   *
   * @return a boolean value indicating whether to add a default cache server.
   */
  public boolean isDisableDefaultServer() {
    return disableDefaultServer;
  }

  /**
   * Determines whether the PID file is allowed to be overwritten when the Server is started and a
   * PID file already exists in the Server's specified working directory.
   *
   * @return boolean indicating if force has been enabled.
   */
  public boolean isForcing() {
    return force;
  }

  /**
   * Determines whether this launcher will be used to display help information. If so, then none of
   * the standard Server launcher commands will be used to affect the state of the Server. A
   * launcher is said to be 'helping' if the user entered the "--help" option (switch) on the
   * command-line.
   *
   * @return a boolean value indicating if this launcher is used for displaying help information.
   * @see ServerLauncher.Command
   */
  public boolean isHelping() {
    return help;
  }

  /**
   * Determines whether a rebalance operation on the cache will occur upon starting the Geode
   * server using this launcher.
   *
   * @return a boolean indicating if the cache will be rebalance when the Geode server starts.
   */
  public boolean isRebalancing() {
    return rebalance;
  }

  /**
   * Determines whether this launcher will redirect output to system logs when starting a new Server
   * process.
   *
   * @return a boolean value indicating if this launcher will redirect output to system logs when
   *         starting a new Server process
   */
  public boolean isRedirectingOutput() {
    return redirectOutput;
  }

  /**
   * Gets the name of the log file used to log information about this Server.
   *
   * @return a String value indicating the name of this Server's log file.
   */
  @Override
  public String getLogFileName() {
    return defaultIfBlank(getMemberName(), DEFAULT_SERVER_LOG_NAME).concat(DEFAULT_SERVER_LOG_EXT);
  }

  /**
   * Gets the name of this member (this Server) in the Geode distributed system as determined by
   * the 'name' Geode property.
   *
   * @return a String indicating the name of the member (this Server) in the Geode distributed
   *         system.
   */
  @Override
  public String getMemberName() {
    return defaultIfBlank(memberName, super.getMemberName());
  }

  /**
   * Gets the user-specified process ID (PID) of the running Server that ServerLauncher uses to
   * issue status and stop commands to the Server.
   *
   * @return an Integer value indicating the process ID (PID) of the running Server.
   */
  @Override
  public Integer getPid() {
    return pid;
  }

  /**
   * Gets the Geode Distributed System (cluster) Properties.
   *
   * @return a Properties object containing the configuration settings for the Geode Distributed
   *         System (cluster).
   * @see Properties
   */
  public Properties getProperties() {
    return (Properties) distributedSystemProperties.clone();
  }

  /**
   * Gets the IP address to which the Server is bound listening for and accepting cache client
   * connections. This property should not be confused with 'bindAddress' ServerLauncher property,
   * which is the port for binding the Server's ServerSocket used in distribution and messaging
   * between the peers of the Geode distributed system.
   *
   * @return an InetAddress indicating the IP address that the Server is bound to listening for and
   *         accepting cache client connections in a client/server topology.
   */
  public InetAddress getServerBindAddress() {
    return serverBindAddress;
  }

  /**
   * Gets the host, as either hostname or IP address, on which the Server was bound and running. An
   * attempt is made to get the canonical hostname for IP address to which the Server was bound for
   * accepting client requests. If the server bind address is null or localhost is unknown, then a
   * default String value of "localhost/127.0.0.1" is returned.
   *
   * Note, this information is purely information and should not be used to re-construct state or
   * for other purposes.
   *
   * @return the hostname or IP address of the host running the Server, based on the bind-address,
   *         or 'localhost/127.0.0.1' if the bind address is null and localhost is unknown.
   * @see InetAddress
   * @see #getServerBindAddress()
   */
  public String getServerBindAddressAsString() {
    try {
      if (getServerBindAddress() != null) {
        return getServerBindAddress().getCanonicalHostName();
      }

      return LocalHostUtil.getCanonicalLocalHostName();
    } catch (UnknownHostException handled) {
      // Returning localhost/127.0.0.1 implies the serverBindAddress was null and no IP address
      // for localhost could be found
      return "localhost/127.0.0.1";
    }
  }

  /**
   * Gets the port on which the Server is listening for cache client connections. This property
   * should not be confused with the 'port' ServerLauncher property, which is used by the Server to
   * set the 'tcp-port' distribution config property and is used by the ServerSocket for peer
   * distribution and messaging.
   *
   * @return an Integer value indicating the port the Server is listening on for cache client
   *         connections in the client/server topology.
   */
  public Integer getServerPort() {
    return serverPort;
  }

  /**
   * Gets the server port on which the Server is listening for client requests represented as a
   * String value.
   *
   * @return a String representing the server port on which the Server is listening for client
   *         requests.
   * @see #getServerPort()
   */
  public String getServerPortAsString() {
    Integer v1 = getServerPort();
    return (v1 != null ? v1 : getDefaultServerPort()).toString();
  }

  /**
   * Gets the name for a Geode Server.
   *
   * @return a String indicating the name for a Geode Server.
   */
  @Override
  public String getServiceName() {
    return SERVER_SERVICE_NAME;
  }

  /**
   * Gets the location of the Spring XML configuration meta-data file used to bootstrap, configure
   * and initialize the Geode Server on start.
   * <p>
   *
   * @return a String indicating the location of the Spring XML configuration file.
   * @see ServerLauncher.Builder#getSpringXmlLocation()
   */
  public String getSpringXmlLocation() {
    return springXmlLocation;
  }

  /**
   * Determines whether this Geode Server was configured and initialized with Spring configuration
   * meta-data.
   * <p>
   *
   * @return a boolean value indicating whether this Geode Server was configured with Spring
   *         configuration meta-data.
   */
  public boolean isSpringXmlLocationSpecified() {
    return isNotBlank(springXmlLocation);
  }

  /**
   * Gets the working directory pathname in which the Server will be run.
   *
   * @return a String value indicating the pathname of the Server's working directory.
   */
  @Override
  public String getWorkingDirectory() {
    return workingDirectory;
  }

  public Float getCriticalHeapPercentage() {
    return criticalHeapPercentage;
  }

  public Float getEvictionHeapPercentage() {
    return evictionHeapPercentage;
  }

  public Float getCriticalOffHeapPercentage() {
    return criticalOffHeapPercentage;
  }

  public Float getEvictionOffHeapPercentage() {
    return evictionOffHeapPercentage;
  }

  public String getHostNameForClients() {
    return hostNameForClients;
  }

  public Integer getMaxConnections() {
    return maxConnections;
  }

  public Integer getMaxMessageCount() {
    return maxMessageCount;
  }

  public Integer getMessageTimeToLive() {
    return messageTimeToLive;
  }

  public Integer getMaxThreads() {
    return maxThreads;
  }

  public Integer getSocketBufferSize() {
    return socketBufferSize;
  }

  private static final String TWO_NEW_LINES = lineSeparator() + lineSeparator();

  /**
   * Displays help for the specified Server launcher command to standard output. If the Server
   * launcher
   * command is unspecified, then usage information is displayed instead.
   *
   * @param command the Server launcher command in which to display help information.
   * @see #usage()
   */
  public void help(final Command command) {
    if (Command.isUnspecified(command)) {
      usage();
    } else {
      info(wrap(helpMap.get(command.getName()), 80, ""));
      info(TWO_NEW_LINES + "usage: " + TWO_NEW_LINES);
      info(wrap("> java ... " + getClass().getName() + ' ' + usageMap.get(command), 80, "\t\t"));
      info(TWO_NEW_LINES + "options: " + TWO_NEW_LINES);

      for (final String option : command.getOptions()) {
        info(wrap("--" + option + ": " + helpMap.get(option) + lineSeparator(), 80, "\t"));
      }

      info(TWO_NEW_LINES);
    }
  }

  /**
   * Displays usage information on the proper invocation of the ServerLauncher from the command-line
   * to standard output.
   *
   * @see #help(ServerLauncher.Command)
   */
  public void usage() {
    info(wrap(helpMap.get("launcher"), 80, "\t"));
    info(TWO_NEW_LINES + "START" + TWO_NEW_LINES);
    help(Command.START);
    info("STATUS" + TWO_NEW_LINES);
    help(Command.STATUS);
    info("STOP" + TWO_NEW_LINES);
    help(Command.STOP);
  }

  /**
   * A Runnable method used to invoke the Geode server (cache server) with the specified command.
   * From run, a user can invoke 'start', 'status', 'stop' and 'version'. Note, that 'version' is
   * also a command-line option, but can be treated as a "command" as well.
   *
   * @see Runnable
   */
  @Override
  public void run() {
    if (!isHelping()) {
      switch (getCommand()) {
        case START:
          info(start());
          waitOnServer();
          break;
        case STATUS:
          info(status());
          break;
        case STOP:
          info(stop());
          break;
        case VERSION:
          info(version());
          break;
        default:
          usage();
      }
    } else {
      help(getCommand());
    }
  }

  /**
   * Gets a File reference with the path to the PID file for the Server.
   *
   * @return a File reference to the path of the Server's PID file.
   */
  protected File getServerPidFile() {
    return new File(getWorkingDirectory(), ProcessType.SERVER.getPidFileName());
  }

  /**
   * Determines whether a Geode Cache Server can be started with this instance of ServerLauncher.
   *
   * @return a boolean indicating whether a Geode Cache Server can be started with this instance
   *         of ServerLauncher, which is true if the ServerLauncher has not already started a Server
   *         or a Server is not already running.
   * @see #start()
   */
  private boolean isStartable() {
    return !isRunning() && starting.compareAndSet(false, true);
  }

  /**
   * Invokes the 'start' command and operation to startup a Geode server (a cache server). Note,
   * this method will cause the JVM to block upon server start, providing the calling Thread is a
   * non-daemon Thread.
   *
   * @return a {@link ServerState} representing the state of the server following the attempt to
   *         start
   * @see #run()
   */
  public ServerState start() {
    long startTime = System.currentTimeMillis();
    if (isStartable()) {
      INSTANCE.compareAndSet(null, this);

      boolean serializationFilterConfigured = configureGlobalSerialFilterIfEnabled();

      try {
        process = getControllableProcess();

        if (!isDisableDefaultServer()) {
          assertPortAvailable(getServerBindAddress(), getServerPort());
        }

        SystemFailure.setExitOK(true);

        ProcessLauncherContext.set(isRedirectingOutput(), getOverriddenDefaults(),
            (String statusMessage) -> {
              debug("Callback setStatus(String) called with message (%1$s)...", statusMessage);
              this.statusMessage = statusMessage;
            });

        try {
          final Properties gemfireProperties = getDistributedSystemProperties(getProperties());
          cache = createCache(gemfireProperties);

          if (serializationFilterConfigured) {
            log.info("Global serial filter is now configured.");
          }

          // Set the resource manager options
          if (criticalHeapPercentage != null) {
            cache.getResourceManager().setCriticalHeapPercentage(getCriticalHeapPercentage());
          }
          if (evictionHeapPercentage != null) {
            cache.getResourceManager().setEvictionHeapPercentage(getEvictionHeapPercentage());
          }
          if (criticalOffHeapPercentage != null) {
            cache.getResourceManager()
                .setCriticalOffHeapPercentage(getCriticalOffHeapPercentage());
          }
          if (evictionOffHeapPercentage != null) {
            cache.getResourceManager()
                .setEvictionOffHeapPercentage(getEvictionOffHeapPercentage());
          }

          cache.setIsServer(true);
          startCacheServer(cache);

          assignBuckets(cache);
          rebalance(cache);
        } finally {
          ProcessLauncherContext.remove();
        }

        awaitStartupTasks(cache, startTime);

        debug("Running Server on (%1$s) in (%2$s) as (%3$s)...", getId(), getWorkingDirectory(),
            getMember());
        running.set(true);

        return new ServerState(this, Status.ONLINE);
      } catch (AuthenticationRequiredException e) {
        failOnStart(e);
