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

import static java.util.Collections.singleton;
import static java.util.concurrent.TimeUnit.MILLISECONDS;
import static org.apache.commons.lang3.exception.ExceptionUtils.getStackTrace;
import static org.apache.geode.distributed.ConfigurationProperties.BIND_ADDRESS;
import static org.apache.geode.distributed.ConfigurationProperties.CACHE_XML_FILE;
import static org.apache.geode.distributed.ConfigurationProperties.LOCATORS;
import static org.apache.geode.util.internal.GeodeGlossary.GEMFIRE_PREFIX;

import java.io.File;
import java.io.IOException;
import java.net.InetAddress;
import java.net.URI;
import java.net.UnknownHostException;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.util.Arrays;
import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;

import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.Nullable;

import org.apache.geode.CancelException;
import org.apache.geode.GemFireConfigException;
import org.apache.geode.annotations.VisibleForTesting;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.cache.client.internal.locator.ClientConnectionRequest;
import org.apache.geode.cache.client.internal.locator.ClientReplacementRequest;
import org.apache.geode.cache.client.internal.locator.GetAllServersRequest;
import org.apache.geode.cache.client.internal.locator.LocatorListRequest;
import org.apache.geode.cache.client.internal.locator.LocatorStatusRequest;
import org.apache.geode.cache.client.internal.locator.QueueConnectionRequest;
import org.apache.geode.cache.client.internal.locator.wan.LocatorMembershipListener;
import org.apache.geode.cache.internal.HttpService;
import org.apache.geode.distributed.ConfigurationProperties;
import org.apache.geode.distributed.DistributedSystem;
import org.apache.geode.distributed.Locator;
import org.apache.geode.distributed.internal.InternalDistributedSystem.ConnectListener;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.distributed.internal.membership.adapter.ServiceConfig;
import org.apache.geode.distributed.internal.membership.api.LocatorConfigurationParser;
import org.apache.geode.distributed.internal.membership.api.MemberIdentifier;
import org.apache.geode.distributed.internal.membership.api.MembershipConfig;
import org.apache.geode.distributed.internal.membership.api.MembershipConfigurationException;
import org.apache.geode.distributed.internal.membership.api.MembershipLocator;
import org.apache.geode.distributed.internal.membership.api.MembershipLocatorBuilder;
import org.apache.geode.distributed.internal.membership.api.QuorumChecker;
import org.apache.geode.distributed.internal.tcpserver.HostAddress;
import org.apache.geode.distributed.internal.tcpserver.HostAndPort;
import org.apache.geode.distributed.internal.tcpserver.InfoRequest;
import org.apache.geode.distributed.internal.tcpserver.TcpHandler;
import org.apache.geode.distributed.internal.tcpserver.TcpServer;
import org.apache.geode.distributed.internal.tcpserver.TcpSocketCreator;
import org.apache.geode.internal.CopyOnWriteHashSet;
import org.apache.geode.internal.GemFireVersion;
import org.apache.geode.internal.InternalDataSerializer;
import org.apache.geode.internal.admin.remote.DistributionLocatorId;
import org.apache.geode.internal.admin.remote.RemoteTransportConfig;
import org.apache.geode.internal.cache.GemFireCacheImpl;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.cache.InternalCacheBuilder;
import org.apache.geode.internal.cache.wan.WANServiceProvider;
import org.apache.geode.internal.config.JAXBService;
import org.apache.geode.internal.inet.LocalHostUtil;
import org.apache.geode.internal.logging.CoreLoggingExecutors;
import org.apache.geode.internal.logging.InternalLogWriter;
import org.apache.geode.internal.logging.LogWriterFactory;
import org.apache.geode.internal.net.SocketCreatorFactory;
import org.apache.geode.internal.security.SecurableCommunicationChannel;
import org.apache.geode.internal.statistics.StatisticsConfig;
import org.apache.geode.logging.internal.InternalSessionContext;
import org.apache.geode.logging.internal.LoggingSession;
import org.apache.geode.logging.internal.NullLoggingSession;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.log4j.api.LogService;
import org.apache.geode.logging.internal.spi.LogConfig;
import org.apache.geode.logging.internal.spi.LogConfigListener;
import org.apache.geode.logging.internal.spi.LogConfigSupplier;
import org.apache.geode.management.ManagementService;
import org.apache.geode.management.api.ClusterManagementService;
import org.apache.geode.management.internal.AgentUtil;
import org.apache.geode.management.internal.JmxManagerLocator;
import org.apache.geode.management.internal.JmxManagerLocatorRequest;
import org.apache.geode.management.internal.api.LocatorClusterManagementService;
import org.apache.geode.management.internal.configuration.domain.SharedConfigurationStatus;
import org.apache.geode.management.internal.configuration.handlers.ClusterManagementServiceInfoRequestHandler;
import org.apache.geode.management.internal.configuration.handlers.SharedConfigurationStatusRequestHandler;
import org.apache.geode.management.internal.configuration.messages.ClusterManagementServiceInfoRequest;
import org.apache.geode.management.internal.configuration.messages.SharedConfigurationStatusRequest;
import org.apache.geode.management.internal.configuration.messages.SharedConfigurationStatusResponse;
import org.apache.geode.metrics.internal.InternalDistributedSystemMetricsService;
import org.apache.geode.security.AuthTokenEnabledComponents;

/**
 * Provides the implementation of a distribution {@code Locator} as well as internal-only
 * functionality.
 *
 * <p>
 * This class has APIs that perform essentially three layers of services. At the bottom layer is the
 * JGroups location service. On top of that you can start a distributed system. And then on top of
 * that you can start server location services.
 *
 * <p>
 * Server Location Service DistributedSystem Peer Location Service
 *
 * <p>
 * The startLocator() methods provide a way to start all three services in one call. Otherwise, the
 * services can be started independently {@code  locator = createLocator();
 * locator.startPeerLocation(); locator.startDistributeSystem();}
 *
 * @since GemFire 4.0
 */
public class InternalLocator extends Locator implements ConnectListener, LogConfigSupplier {
  public static final int MAX_POOL_SIZE =
      Integer.getInteger(GEMFIRE_PREFIX + "TcpServer.MAX_POOL_SIZE", 100);
  public static final int POOL_IDLE_TIMEOUT = 60 * 1000;

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

  /**
   * system property name for forcing an locator distribution manager type
   */
  public static final String FORCE_LOCATOR_DM_TYPE = "Locator.forceLocatorDMType";

  /**
   * system property name for inhibiting DM banner
   */
  public static final String INHIBIT_DM_BANNER = "Locator.inhibitDMBanner";

  /**
   * system property name for forcing locators to be preferred as coordinators
   */
  public static final String LOCATORS_PREFERRED_AS_COORDINATORS =
      GEMFIRE_PREFIX + "disable-floating-coordinator";
  static final String IGNORING_RECONNECT_REQUEST =
      "ignoring request to reconnect to the cluster - there is no DistributedSystem available to reconnect.";

  /**
   * the locator hosted by this JVM. As of 7.0 it is a singleton.
   *
   * GuardedBy must synchronize on locatorLock
   */
  @MakeNotStatic
  private static InternalLocator locator;

  private static final Object locatorLock = new Object();

  private final Set<RestartHandler> restartHandlers = new CopyOnWriteHashSet<>();
  private final LocatorMembershipListener locatorListener;
  private final AtomicBoolean shutdownHandled = new AtomicBoolean(false);
  private final LoggingSession loggingSession;
  private final LocatorStats locatorStats;
  private final Path workingDirectory;
  private final MembershipLocator<InternalDistributedMember> membershipLocator;

  /**
   * whether the locator was stopped during forced-disconnect processing but a reconnect will occur
   */
  @VisibleForTesting
  volatile boolean stoppedForReconnect;
  private volatile boolean reconnected;

  /**
   * whether the locator was stopped during forced-disconnect processing
   */
  private volatile boolean forcedDisconnect;
  private volatile boolean isSharedConfigurationStarted;
  private volatile Thread restartThread;

  /**
   * The distributed system owned by this locator, if any. Note that if a ds already exists because
   * the locator is being colocated in a normal member this field will be null.
   */
  private InternalDistributedSystem internalDistributedSystem;

  /**
   * The cache owned by this locator, if any. Note that if a cache already exists because the
   * locator is being colocated in a normal member this field will be null.
   */
  private InternalCache internalCache;

  private boolean peerLocator;
  private ServerLocator serverLocator;
  private Properties env;

  private final DistributionConfigImpl distributionConfig;
  private WanLocatorDiscoverer locatorDiscoverer;
  private InternalConfigurationPersistenceService configurationPersistenceService;
  private ClusterManagementService clusterManagementService;
  // synchronization lock that ensures we only have one thread performing location services
  // restart at a time
  private final Object servicesRestartLock = new Object();

  public static @Nullable InternalLocator getLocator() {
    synchronized (locatorLock) {
      return locator;
    }
  }

  public static boolean hasLocator() {
    synchronized (locatorLock) {
      return locator != null;
    }
  }

  private static void removeLocator(InternalLocator locator) {
    if (locator == null) {
      return;
    }
    synchronized (locatorLock) {
      if (locator.loggingSession.getState() != InternalSessionContext.State.STOPPED) {
        locator.loggingSession.stopSession();
        locator.loggingSession.shutdown();
      }
      if (locator.equals(InternalLocator.locator)) {
        InternalLocator.locator = null;
      }
    }
  }

  /**
   * Create a locator that listens on a given port. This locator will not have peer or server
   * location services available until they are started by calling startServerLocation or
   * startPeerLocation on the locator object.
   *
   * @param port the tcp/ip port to listen on
   * @param loggingSession the LoggingSession to use, may be a NullLoggingSession which does
   *        nothing
   * @param logFile the file that log messages should be written to
   * @param logWriter a log writer that should be used (logFile parameter is ignored)
   * @param securityLogWriter the logWriter to be used for security related log messages
   * @param distributedSystemProperties optional properties to configure the distributed system
   *        (e.g., mcast addr/port, other locators)
   * @param startDistributedSystem if true then this locator will also start its own ds
   *
   * @deprecated Please use
   *             {@link #createLocator(int, LoggingSession, File, InternalLogWriter, InternalLogWriter, HostAddress, String, Properties, Path)}
   *             instead.
   */
  @Deprecated
  public static InternalLocator createLocator(int port, LoggingSession loggingSession, File logFile,
      InternalLogWriter logWriter,
      InternalLogWriter securityLogWriter,
      HostAddress bindAddress,
      String hostnameForClients,
      Properties distributedSystemProperties,
      boolean startDistributedSystem) {
    return createLocator(port, loggingSession, logFile, logWriter, securityLogWriter, bindAddress,
        hostnameForClients, distributedSystemProperties,
        Paths.get(System.getProperty("user.dir")));
  }

  /**
   * Create a locator that listens on a given port. This locator will not have peer or server
   * location services available until they are started by calling startServerLocation or
   * startPeerLocation on the locator object.
   *
   * @param port the tcp/ip port to listen on
   * @param loggingSession the LoggingSession to use, may be a NullLoggingSession which does
   *        nothing
   * @param logFile the file that log messages should be written to
   * @param logWriter a log writer that should be used (logFile parameter is ignored)
   * @param securityLogWriter the logWriter to be used for security related log messages
   * @param bindAddress the bind address for the locator
   * @param distributedSystemProperties optional properties to configure the distributed system
   *        (e.g., mcast addr/port, other locators)
   * @param workingDirectory the working directory to use for any files
   */
  public static InternalLocator createLocator(int port, LoggingSession loggingSession, File logFile,
      InternalLogWriter logWriter,
      InternalLogWriter securityLogWriter,
      HostAddress bindAddress,
      String hostnameForClients,
      Properties distributedSystemProperties,
      Path workingDirectory) {
    synchronized (locatorLock) {
      if (hasLocator()) {
        throw new IllegalStateException(
            "A locator can not be created because one already exists in this JVM.");
      }
      InternalLocator locator =
          new InternalLocator(port, loggingSession, logFile, logWriter, securityLogWriter,
              bindAddress, hostnameForClients, distributedSystemProperties, null,
              workingDirectory);
      InternalLocator.locator = locator;
      return locator;
    }
  }

  private static void setLocator(InternalLocator locator) {
    synchronized (locatorLock) {
      if (InternalLocator.locator != null && InternalLocator.locator != locator) {
        throw new IllegalStateException(
            "A locator can not be created because one already exists in this JVM.");
      }

      InternalLocator.locator = locator;
    }
  }

  /**
   * Creates a distribution locator that runs in this VM on the given port and bind address in the
   * default working directory (Java "user.dir" System Property).
   *
   * <p>
   * This is for internal use only as it does not create a distributed system unless told to do so.
   *
   * @param port the tcp/ip port to listen on
   * @param logFile the file that log messages should be written to
   * @param logWriter a log writer that should be used (logFile parameter is ignored)
   * @param securityLogWriter the logWriter to be used for security related log messages
   * @param startDistributedSystem if true, a distributed system is started
   * @param distributedSystemProperties optional properties to configure the distributed system
   *        (e.g., mcast
   *        addr/port, other locators)
   * @param hostnameForClients the name to give to clients for connecting to this locator
   */
  public static InternalLocator startLocator(int port, File logFile, InternalLogWriter logWriter,
      InternalLogWriter securityLogWriter, HostAddress bindAddress, boolean startDistributedSystem,
      Properties distributedSystemProperties, String hostnameForClients)
      throws IOException {
    return startLocator(port, logFile, logWriter, securityLogWriter, bindAddress,
        startDistributedSystem, distributedSystemProperties, hostnameForClients,
        Paths.get(System.getProperty("user.dir")));
  }

  /**
   * Creates a distribution locator that runs in this VM on the given port and bind address in the
   * specified working directory.
   *
   * <p>
   * This is for internal use only as it does not create a distributed system unless told to do so.
   *
   * @param port the tcp/ip port to listen on
   * @param logFile the file that log messages should be written to
   * @param logWriter a log writer that should be used (logFile parameter is ignored)
   * @param securityLogWriter the logWriter to be used for security related log messages
   * @param bindAddress the bind address for the locator
   * @param startDistributedSystem if true, a distributed system is started
   * @param distributedSystemProperties optional properties to configure the distributed system
   *        (e.g., mcast
   *        addr/port, other locators)
   * @param hostnameForClients the name to give to clients for connecting to this locator
   * @param workingDirectory the working directory to use for any files
   */
  public static InternalLocator startLocator(int port, File logFile, InternalLogWriter logWriter,
      InternalLogWriter securityLogWriter,
      HostAddress bindAddress,
      boolean startDistributedSystem,
      Properties distributedSystemProperties,
      String hostnameForClients, Path workingDirectory)
      throws IOException {
    System.setProperty(FORCE_LOCATOR_DM_TYPE, "true");
    InternalLocator newLocator = null;

    boolean startedLocator = false;
    try {

      // if startDistributedSystem is true then Locator uses a NullLoggingSession (does nothing)
      LoggingSession loggingSession =
          startDistributedSystem ? NullLoggingSession.create() : LoggingSession.create();

      newLocator = createLocator(port, loggingSession, logFile, logWriter, securityLogWriter,
          bindAddress, hostnameForClients, distributedSystemProperties,
          workingDirectory);

      loggingSession.createSession(newLocator);
      loggingSession.startSession();

      try {
        newLocator.startPeerLocation();

        if (startDistributedSystem) {

          try {
            newLocator.startDistributedSystem();
          } catch (RuntimeException e) {
            newLocator.stop();
            throw e;
          }

          InternalDistributedSystem system = newLocator.internalDistributedSystem;
          if (system != null) {
            system.getDistributionManager().addHostedLocators(system.getDistributedMember(),
                getLocatorStrings(), newLocator.isSharedConfigurationEnabled());
          }
        }
      } catch (IllegalStateException e) {
        newLocator.stop();
        throw e;
      }

      InternalDistributedSystem system = InternalDistributedSystem.getConnectedInstance();
      if (system != null) {
        try {
          newLocator.startServerLocation(system);
        } catch (RuntimeException e) {
          newLocator.stop();
          throw e;
        }
      }

      newLocator.endStartLocator(null);
      startedLocator = true;
      return newLocator;

    } finally {
      System.clearProperty(FORCE_LOCATOR_DM_TYPE);

      if (!startedLocator) {
        removeLocator(newLocator);
      }
    }
  }

  /**
   * Determines if this VM is a locator which must ignore a shutdown.
   *
   * @return true if this VM is a locator which should ignore a shutdown, false if it is a normal
   *         member.
   */
  public static boolean isDedicatedLocator() {
    InternalLocator internalLocator = getLocator();
    if (internalLocator == null) {
      return false;
    }

    InternalDistributedSystem system = internalLocator.internalDistributedSystem;
    if (system == null) {
      return false;
    }
    DistributionManager distributionManager = system.getDistributionManager();
    if (distributionManager.isLoner()) {
      return false;
    }
    ClusterDistributionManager clusterDistributionManager =
        (ClusterDistributionManager) system.getDistributionManager();
    return clusterDistributionManager.getDMType() == ClusterDistributionManager.LOCATOR_DM_TYPE;
  }

  /**
   * Creates a new {@code Locator} with the given port, log file, logWriter, and bind address.
   *
   * @param port the tcp/ip port to listen on
   * @param logFile the file that log messages should be written to
   * @param logWriter a log writer that should be used (logFile parameter is ignored)
   * @param securityLogWriter the log writer to be used for security related log messages
   * @param hostnameForClients the name to give to clients for connecting to this locator
   * @param distributedSystemProperties optional properties to configure the distributed system
   *        (e.g., mcast addr/port, other locators)
   * @param distributionConfig the config if being called from a distributed system; otherwise null.
   * @param workingDirectory the working directory to use for files
   */
  @VisibleForTesting
  InternalLocator(int port, LoggingSession loggingSession, File logFile,
      InternalLogWriter logWriter, InternalLogWriter securityLogWriter,
      HostAddress hostAddress, String hostnameForClients,
      Properties distributedSystemProperties,
      DistributionConfigImpl distributionConfig, Path workingDirectory) {
    this.logFile = logFile;
    this.hostAddress = hostAddress;
    // bindAddress is superceded by hostAddress but must be kept for Locator API backward
    // compatibility
    if (hostAddress != null) {
      bindAddress = hostAddress.getAddress();
    }
    this.hostnameForClients = hostnameForClients;

    this.workingDirectory = workingDirectory;


    env = new Properties();

    // set bind-address explicitly only if not wildcard and let any explicit
    // value in distributedSystemProperties take precedence
    if (hostAddress != null) {
      env.setProperty(BIND_ADDRESS, hostAddress.getHostName());
    }

    if (distributedSystemProperties != null) {
      env.putAll(distributedSystemProperties);
    }
    env.setProperty(CACHE_XML_FILE, "");

    // create a DC so that all of the lookup rules, gemfire.properties, etc,
    // are considered and we have a config object we can trust
    if (distributionConfig == null) {
      distributionConfig = new DistributionConfigImpl(env);
      env.clear();
      env.putAll(distributionConfig.getProps());
    }
    this.distributionConfig = distributionConfig;

    boolean hasLogFileButConfigDoesNot =
        this.logFile != null && this.distributionConfig.getLogFile()
            .toString().equals(DistributionConfig.DEFAULT_LOG_FILE.toString());
    if (logWriter == null && hasLogFileButConfigDoesNot) {
      // LOG: this is(was) a hack for when logFile and config don't match -- if config specifies a
      // different log-file things will break!
      this.distributionConfig.unsafeSetLogFile(this.logFile);
    }

    if (loggingSession == null) {
      throw new Error("LoggingSession must not be null");
    }
    this.loggingSession = loggingSession;

    // LOG: create LogWriters for GemFireTracer (or use whatever was passed in)
    if (logWriter == null) {
      LogWriterFactory.createLogWriterLogger(this.distributionConfig, false);
      if (logger.isDebugEnabled()) {
        logger.debug("LogWriter for locator is created.");
      }
    }

    if (securityLogWriter == null) {
      securityLogWriter = LogWriterFactory.createLogWriterLogger(this.distributionConfig, true);
      securityLogWriter.fine("SecurityLogWriter for locator is created.");
    }

    SocketCreatorFactory.setDistributionConfig(this.distributionConfig);

    locatorListener = WANServiceProvider.createLocatorMembershipListener();
    if (locatorListener != null) {
      // We defer setting the port until the handler is init'd - that way we'll have an actual port
      // in the case where we're starting with port = 0.
      locatorListener.setConfig(getConfig());
      locatorListener.start();
    }

    locatorStats = new LocatorStats();

    InternalLocatorTcpHandler handler = new InternalLocatorTcpHandler();
    try {
      MembershipConfig config = new ServiceConfig(
          new RemoteTransportConfig(distributionConfig, MemberIdentifier.LOCATOR_DM_TYPE),
          distributionConfig);
      Supplier<ExecutorService> executor = () -> CoreLoggingExecutors
          .newThreadPoolWithSynchronousFeed(MAX_POOL_SIZE, POOL_IDLE_TIMEOUT, MILLISECONDS,
              "locator request thread ", new ThreadPoolExecutor.CallerRunsPolicy(),
              new DelayedPoolStatHelper());
      final TcpSocketCreator socketCreator = SocketCreatorFactory
          .getSocketCreatorForComponent(SecurableCommunicationChannel.LOCATOR);
      membershipLocator =
          MembershipLocatorBuilder.<InternalDistributedMember>newLocatorBuilder(
              socketCreator,
              InternalDataSerializer.getDSFIDSerializer(),
              workingDirectory,
              executor)
              .setConfig(config)
              .setPort(port)
              .setBindAddress(hostAddress)
              .setProtocolChecker(new ProtocolCheckerImpl())
              .setFallbackHandler(handler)
              .setLocatorsAreCoordinators(shouldLocatorsBeCoordinators())
              .setLocatorStats(locatorStats)
              .create();
    } catch (MembershipConfigurationException | UnknownHostException e) {
      throw new GemFireConfigException(e.getMessage());
    }

    membershipLocator.addHandler(InfoRequest.class, new InfoRequestHandler());
    restartHandlers.add((ds, cache, sharedConfig) -> {
      final InternalDistributedSystem ids = (InternalDistributedSystem) ds;
      // let old locator know about new membership object
      membershipLocator.setMembership(ids.getDM().getDistribution().getMembership());
    });
  }

  public boolean isSharedConfigurationEnabled() {
    return distributionConfig.getEnableClusterConfiguration();
  }

  private boolean loadFromSharedConfigDir() {
    return distributionConfig.getLoadClusterConfigFromDir();
  }

  public boolean isSharedConfigurationRunning() {
    return configurationPersistenceService != null
        && configurationPersistenceService.getStatus() == SharedConfigurationStatus.RUNNING;
  }

  public LocatorMembershipListener getLocatorMembershipListener() {
    return locatorListener;
  }

  /**
   * @deprecated Please use {@link #getLocatorMembershipListener()} instead.
   */
  @Deprecated
  public LocatorMembershipListener getlocatorMembershipListener() {
    return getLocatorMembershipListener();
  }

  private void startTcpServer() throws IOException {
    logger.info("Starting {}", this);
    membershipLocator.start();
  }

  public InternalConfigurationPersistenceService getConfigurationPersistenceService() {
    return configurationPersistenceService;
  }

  public DistributionConfigImpl getConfig() {
    return distributionConfig;
  }

  public InternalCache getCache() {
    if (internalCache == null) {
      return GemFireCacheImpl.getInstance();
    }
    return internalCache;
  }

  /**
   * Start peer location in this locator. If you plan on starting a distributed system later, this
   * method should be called first so that the distributed system can use this locator.
   *
   * @return returns the port that the locator to which the locator is bound
   * @since GemFire 5.7
   */
  int startPeerLocation() throws IOException {
    if (isPeerLocator()) {
      throw new IllegalStateException(
          String.format("Peer location is already running for %s", this));
    }
    logger.info("Starting peer location for {}", this);

    peerLocator = true;
    return membershipLocator.start();
  }

  private boolean shouldLocatorsBeCoordinators() {
    // check for settings that would require only locators to hold the
    // coordinator - e.g., security and network-partition detection
    boolean locatorsAreCoordinators;
    boolean networkPartitionDetectionEnabled =
        distributionConfig.getEnableNetworkPartitionDetection();
    if (networkPartitionDetectionEnabled) {
      locatorsAreCoordinators = true;
    } else {
      // check if security is enabled
      String prop = distributionConfig.getSecurityPeerAuthInit();
      locatorsAreCoordinators = prop != null && !prop.isEmpty();
      if (!locatorsAreCoordinators) {
        locatorsAreCoordinators = Boolean.getBoolean(LOCATORS_PREFERRED_AS_COORDINATORS);
      }
    }
    return locatorsAreCoordinators;
  }

  /**
   * @return the TcpHandler for peer to peer discovery
   */
  public MembershipLocator<InternalDistributedMember> getMembershipLocator() {
    return membershipLocator;
  }

  /**
   * For backward-compatibility we retain this method
   *
   * @deprecated use a form of the method that does not have peerLocator/serverLocator parameters
   */
  @Deprecated
  public static InternalLocator startLocator(int locatorPort, File logFile,
      InternalLogWriter logWriter, InternalLogWriter securityLogWriter, HostAddress bindAddress,
      Properties distributedSystemProperties, boolean peerLocator, boolean serverLocator,
      String hostnameForClients, boolean b1) throws IOException {
    return startLocator(locatorPort, logFile, logWriter, securityLogWriter, bindAddress, true,
        distributedSystemProperties, hostnameForClients);
  }

  /**
   * Start a distributed system whose life cycle is managed by this locator. When the locator is
   * stopped, this distributed system will be disconnected. If a distributed system already exists,
   * this method will have no affect.
   *
   * @since GemFire 5.7
   */
  private void startDistributedSystem() throws IOException {
    InternalDistributedSystem existing = InternalDistributedSystem.getConnectedInstance();
    if (existing != null) {
      // LOG: changed from config to info
      logger.info("Using existing distributed system: {}", existing);
      startCache(existing);
      return;
    }

    HostAddress thisLocator = getHostAddress(hostAddress);
    if (peerLocator) {
      String existingLocators = distributionConfig.getLocators();
      String newLocators = addLocatorIfMissing(existingLocators, thisLocator, getPort());
      if (!newLocators.equals(existingLocators)) {
        setLocatorProperties(newLocators);
      }
    }

    Properties distributedSystemProperties = new Properties();
    // LogWriterAppender is now shared via that class
    // using a DistributionConfig earlier in this method
    distributedSystemProperties.put(DistributionConfig.DS_CONFIG_NAME, distributionConfig);

    logger.info("Starting distributed system");

    internalDistributedSystem =
        InternalDistributedSystem
            .connectInternal(distributedSystemProperties, null,
                new InternalDistributedSystemMetricsService.Builder(),
                membershipLocator);

    if (peerLocator) {
      // We've created a peer location message handler - it needs to be connected to
      // the membership service in order to get membership view notifications
      membershipLocator
          .setMembership(internalDistributedSystem.getDM()
              .getDistribution().getMembership());
    }

    internalDistributedSystem.addDisconnectListener(sys -> stop(false, false, false));

    startCache(internalDistributedSystem);

    logger.info("Locator started on {}[{}]", thisLocator, getPort());

  }

  HostAddress getHostAddress(HostAddress hostAddress) throws UnknownHostException {
    HostAddress thisLocator;
    if (hostAddress != null && !StringUtils.isEmpty(hostAddress.getHostName())) {
      thisLocator = hostAddress;
    } else {
      thisLocator = new HostAddress(LocalHostUtil.getLocalHost());
    }
    return thisLocator;
  }

  /**
   * add this locator to the locators list if not already in, note the locator's
   * hostname is used, but if it resoves to be the same as what's already in the
   * list, what's specified by the user is preserved
   *
   * @param existingLocators the existing configured locators
   * @param thisLocator this locator's address
   * @param port this locator's port
