/*
 * 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 java.io.NotSerializableException;
import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.Comparator;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.stream.Collectors;

import org.apache.logging.log4j.Logger;
import org.jetbrains.annotations.NotNull;
import org.jetbrains.annotations.Nullable;

import org.apache.geode.CancelCriterion;
import org.apache.geode.CancelException;
import org.apache.geode.ForcedDisconnectException;
import org.apache.geode.IncompatibleSystemException;
import org.apache.geode.InternalGemFireError;
import org.apache.geode.InternalGemFireException;
import org.apache.geode.InvalidDeltaException;
import org.apache.geode.SystemConnectException;
import org.apache.geode.SystemFailure;
import org.apache.geode.ToDataException;
import org.apache.geode.admin.GemFireHealthConfig;
import org.apache.geode.alerting.internal.api.AlertingService;
import org.apache.geode.annotations.Immutable;
import org.apache.geode.annotations.internal.MakeNotStatic;
import org.apache.geode.cache.CacheClosedException;
import org.apache.geode.distributed.DistributedMember;
import org.apache.geode.distributed.DistributedSystemDisconnectedException;
import org.apache.geode.distributed.Locator;
import org.apache.geode.distributed.Role;
import org.apache.geode.distributed.internal.locks.ElderState;
import org.apache.geode.distributed.internal.membership.InternalDistributedMember;
import org.apache.geode.distributed.internal.membership.api.MemberData;
import org.apache.geode.distributed.internal.membership.api.MemberDisconnectedException;
import org.apache.geode.distributed.internal.membership.api.MemberIdentifier;
import org.apache.geode.distributed.internal.membership.api.MemberIdentifierFactory;
import org.apache.geode.distributed.internal.membership.api.Membership;
import org.apache.geode.distributed.internal.membership.api.MembershipLocator;
import org.apache.geode.distributed.internal.membership.api.MembershipView;
import org.apache.geode.distributed.internal.membership.api.Message;
import org.apache.geode.internal.Assert;
import org.apache.geode.internal.NanoTimer;
import org.apache.geode.internal.admin.remote.AdminConsoleDisconnectMessage;
import org.apache.geode.internal.admin.remote.RemoteGfManagerAgent;
import org.apache.geode.internal.admin.remote.RemoteTransportConfig;
import org.apache.geode.internal.cache.InitialImageOperation;
import org.apache.geode.internal.cache.InternalCache;
import org.apache.geode.internal.inet.LocalHostUtil;
import org.apache.geode.internal.logging.log4j.LogMarker;
import org.apache.geode.internal.monitoring.ThreadsMonitoring;
import org.apache.geode.internal.sequencelog.MembershipLogger;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.internal.tcp.ConnectionTable;
import org.apache.geode.internal.tcp.ReenteredConnectException;
import org.apache.geode.logging.internal.OSProcess;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.logging.internal.executors.LoggingUncaughtExceptionHandler;
import org.apache.geode.logging.internal.log4j.api.LogService;

/**
 * The <code>DistributionManager</code> uses a {@link Membership} to distribute {@link
 * DistributionMessage messages}. It also reports on who is currently in the distributed system and
 * tracks the elder member for the distributed lock service. You may also register a membership
 * listener with the DistributionManager to receive notification of changes in membership.
 *
 * <p>
 * <p>
 * Code that wishes to send a {@link DistributionMessage} must get the
 * <code>DistributionManager</code> and invoke {@link #putOutgoing}.
 *
 * <p>
 *
 * @see DistributionMessage#process
 * @see IgnoredByManager
 */
public class ClusterDistributionManager implements DistributionManager {

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

  private static final boolean DEBUG_NO_ACKNOWLEDGEMENTS =
      Boolean.getBoolean("DistributionManager.DEBUG_NO_ACKNOWLEDGEMENTS");

  /**
   * Maximum number of interrupt attempts to stop a thread
   */
  private static final int MAX_STOP_ATTEMPTS = 10;


  private static final boolean SYNC_EVENTS = Boolean.getBoolean("DistributionManager.syncEvents");


  /**
   * The DM type for regular distribution managers
   */
  public static final int NORMAL_DM_TYPE = MemberIdentifier.NORMAL_DM_TYPE;

  /**
   * The DM type for locator distribution managers
   */
  public static final int LOCATOR_DM_TYPE = MemberIdentifier.LOCATOR_DM_TYPE;

  /**
   * The DM type for Console (admin-only) distribution managers
   */
  public static final int ADMIN_ONLY_DM_TYPE = MemberIdentifier.ADMIN_ONLY_DM_TYPE;

  /**
   * The DM type for stand-alone members
   */
  public static final int LONER_DM_TYPE = MemberIdentifier.LONER_DM_TYPE;


  /**
   * Is this node running an AdminDistributedSystem?
   */
  @MakeNotStatic
  private static volatile boolean isDedicatedAdminVM = false;

  @MakeNotStatic
  private static final ThreadLocal<Boolean> isStartupThread = new ThreadLocal<>();

  ///////////////////// Instance Fields //////////////////////

  private final Stopper stopper = new Stopper(this);


  /**
   * The id of this distribution manager
   */
  private final InternalDistributedMember localAddress;

  /**
   * The distribution manager type of this dm; set in its constructor.
   */
  private final int dmType;

  /**
   * The <code>MembershipListener</code>s that are registered on this manager.
   */
  private final ConcurrentMap<MembershipListener, Boolean> membershipListeners;
  private final ClusterElderManager clusterElderManager = new ClusterElderManager(this);
  private final Distribution distribution;
  private ClusterOperationExecutors executors;

  /**
   * Membership failure listeners - for testing
   */
  private List<MembershipTestHook> membershipTestHooks;


  /**
   * The <code>MembershipListener</code>s that are registered on this manager for ALL members.
   *
   * @since GemFire 5.7
   */
  private volatile Set<MembershipListener> allMembershipListeners = Collections.emptySet();

  /**
   * A lock to hold while adding and removing all membership listeners.
   *
   * @since GemFire 5.7
   */
  private final Object allMembershipListenersLock = new Object();

  /**
   * A queue of MemberEvent instances
   */
  private final BlockingQueue<MemberEvent> membershipEventQueue = new LinkedBlockingQueue<>();

  /**
   * Used to invoke registered membership listeners in the background.
   */
  private Thread memberEventThread;


  /**
   * A brief description of this DistributionManager
   */
  protected final String description;

  /**
   * Statistics about distribution
   */
  protected final DistributionStats stats;

  /**
   * Did an exception occur in one of the DM threads?
   */
  private boolean exceptionInThreads;

  private volatile boolean shutdownMsgSent = false;

  /**
   * Set to true when this manager is being shutdown
   */
  private volatile boolean closeInProgress = false;

  private volatile boolean receivedStartupResponse = false;

  private volatile String rejectionMessage = null;

  /**
   * Map of all locator members of the distributed system. The value is a collection of locator
   * strings that are hosted in that member. All accesses to this field must be synchronized on
   * {@link #membersLock}.
   */
  private Map<InternalDistributedMember, Collection<String>> hostedLocatorsAll =
      Collections.emptyMap();

  /**
   * Map of all locator members of the distributed system which have the shared configuration. The
   * value is a collection of locator strings that are hosted in that member. All accesses to this
   * field must be synchronized on {@link #membersLock}.
   */
  private Map<InternalDistributedMember, Collection<String>> hostedLocatorsWithSharedConfiguration =
      Collections.emptyMap();

  private volatile boolean readyForMessages = false;

  /**
   * Set to true once this DM is ready to send messages. Note that it is always ready to send the
   * startup message.
   */
  private volatile boolean readyToSendMsgs = false;
  private final Object readyToSendMsgsLock = new Object();

  /**
   * Is this distribution manager closed?
   */
  private volatile boolean closed = false;

  /**
   * The distributed system to which this distribution manager is connected.
   */
  private final InternalDistributedSystem system;

  /**
   * The remote transport configuration for this dm
   */
  private final RemoteTransportConfig transport;

  /**
   * The administration agent associated with this distribution manager.
   */
  private volatile RemoteGfManagerAgent agent;

  /**
   * TODO why does the distribution manager arbitrate GII operations? That should be a Cache
   * function
   */
  private final Semaphore parallelGIIs = new Semaphore(InitialImageOperation.MAX_PARALLEL_GIIS);

  /**
   * Map of InetAddress to HashSets of InetAddress, to define equivalences between network interface
   * cards and hosts.
   */
  private final HashMap<InetAddress, Set<InetAddress>> equivalentHosts = new HashMap<>();

  private int distributedSystemId;


  private final Map<InternalDistributedMember, String> redundancyZones =
      Collections.synchronizedMap(new HashMap<>());

  private boolean enforceUniqueZone = false;

  /**
   * root cause of forcibly shutting down the distribution manager
   */
  private volatile Throwable rootCause = null;

  /**
   * @see #closeInProgress
   */
  private final Object shutdownMutex = new Object();

  private final AlertingService alertingService;

  private final Object membersLock = new Object();

  ////////////////////// Static Methods //////////////////////

  /**
   * Creates a new distribution manager and discovers the other members of the distributed system.
   * Note that it does not check to see whether or not this VM already has a distribution manager.
   *
   * @param system The distributed system to which this distribution manager will send messages.
   */
  static ClusterDistributionManager create(InternalDistributedSystem system,
      final MembershipLocator<InternalDistributedMember> membershipLocator) {

    ClusterDistributionManager distributionManager = null;
    boolean beforeJoined = true;

    try {

      int vmKind;

      if (Boolean.getBoolean(InternalLocator.FORCE_LOCATOR_DM_TYPE)) {
        // if this DM is starting for a locator, set it to be a locator DM
        vmKind = LOCATOR_DM_TYPE;

      } else if (isDedicatedAdminVM()) {
        vmKind = ADMIN_ONLY_DM_TYPE;

      } else {
        vmKind = NORMAL_DM_TYPE;
      }

      RemoteTransportConfig transport = new RemoteTransportConfig(system.getConfig(), vmKind);
      transport.setIsReconnectingDS(system.isReconnectingDS());
      transport.setOldDSMembershipInfo(system.oldDSMembershipInfo());

      long start = System.currentTimeMillis();

      distributionManager =
          new ClusterDistributionManager(system, transport, system.getAlertingService(),
              membershipLocator);
      distributionManager.assertDistributionManagerType();

      beforeJoined = false; // we have now joined the system

      {
        InternalDistributedMember id = distributionManager.getDistributionManagerId();
        if (!"".equals(id.getName())) {
          for (InternalDistributedMember m : distributionManager
              .getViewMembers()) {
            if (m.equals(id)) {
              // I'm counting on the members returned by getViewMembers being ordered such that
              // members that joined before us will precede us AND members that join after us
              // will succeed us.
              // SO once we find ourselves break out of this loop.
              break;
            }
            if (id.getName().equals(m.getName())) {
              if (distributionManager.getDistribution().verifyMember(m,
                  "member is using the name of " + id)) {
                throw new IncompatibleSystemException("Member " + id
                    + " could not join this distributed system because the existing member " + m
                    + " used the same name. Set the \"name\" gemfire property to a unique value.");
              }
            }
          }
        }
        distributionManager.addNewMember(id); // add ourselves
      }

      // Send out a StartupMessage to the other members.
      StartupOperation op = new StartupOperation(distributionManager, transport);

      try {
        if (!distributionManager.sendStartupMessage(op)) {
          // Well we didn't hear back from anyone else. We assume that
          // we're the first one.
          if (distributionManager.getOtherDistributionManagerIds().size() == 0) {
            logger.info("Did not hear back from any other system. I am the first one.");
          } else if (transport.isMcastEnabled()) {
            // perform a multicast ping test
            if (!distributionManager.testMulticast()) {
              logger.warn(
                  "Did not receive a startup response but other members exist.  Multicast does not seem to be working.");
            }
          }
        }
      } catch (InterruptedException ex) {
        Thread.currentThread().interrupt();
        // This is ALWAYS bad; don't consult a CancelCriterion.
        throw new InternalGemFireException(
            "Interrupted while waiting for first StartupResponseMessage",
            ex);
      } catch (IncompatibleSystemException ex) {
        logger.fatal(ex.getMessage(), ex);
        throw ex;
      } finally {
        distributionManager.readyToSendMsgs();
      }

      if (logger.isInfoEnabled()) {
        long delta = System.currentTimeMillis() - start;
        logger.info(LogMarker.DM_MARKER,
            "DistributionManager {} started on {}. There were {} other DMs. others: {} {} {}",
            distributionManager.getDistributionManagerId(), transport,
            distributionManager.getOtherDistributionManagerIds().size(),
            distributionManager.getOtherDistributionManagerIds(),
            (logger.isInfoEnabled(LogMarker.DM_MARKER) ? " (took " + delta + " ms)" : ""),
            ((distributionManager.getDMType() == ADMIN_ONLY_DM_TYPE) ? " (admin only)"
                : (distributionManager.getDMType() == LOCATOR_DM_TYPE) ? " (locator)" : ""));

        MembershipLogger.logStartup(distributionManager.getDistributionManagerId());
      }
      return distributionManager;
    } catch (RuntimeException r) {
      if (distributionManager != null) {
        if (logger.isDebugEnabled()) {
          logger.debug("cleaning up incompletely started DistributionManager due to exception", r);
        }
        distributionManager.uncleanShutdown(beforeJoined);
      }
      throw r;
    }
  }

  public OperationExecutors getExecutors() {
    return executors;
  }

  @Override
  public ThreadsMonitoring getThreadMonitoring() {
    return executors.getThreadMonitoring();
  }

  /////////////////////// Constructors ///////////////////////

  /**
   * Creates a new <code>DistributionManager</code> by initializing itself, creating the membership
   * manager and executors
   *
   * @param transport The configuration for the communications transport
   */
  private ClusterDistributionManager(RemoteTransportConfig transport,
      InternalDistributedSystem system,
      AlertingService alertingService,
      MembershipLocator<InternalDistributedMember> locator) {

    this.system = system;
    this.transport = transport;
    this.alertingService = alertingService;

    dmType = transport.getVmKind();
    membershipListeners = new ConcurrentHashMap<>();
    distributedSystemId = system.getConfig().getDistributedSystemId();

    long statId = OSProcess.getId();
    stats = new DistributionStats(system, statId);
    DistributionStats.enableClockStats = system.getConfig().getEnableTimeStatistics();

    exceptionInThreads = false;
    boolean finishedConstructor = false;
    try {

      executors = new ClusterOperationExecutors(stats, system);

      if (!SYNC_EVENTS) {
        memberEventThread =
            new LoggingThread("DM-MemberEventInvoker", new MemberEventInvoker());
      }

      StringBuilder sb = new StringBuilder(" (took ");

      // connect to the cluster
      long start = System.currentTimeMillis();

      DMListener listener = new DMListener(this);
      distribution = DistributionImpl
          .createDistribution(this, transport, system, listener,
              this::handleIncomingDMsg, locator);

      sb.append(System.currentTimeMillis() - start);

      localAddress = distribution.getLocalMember();

      sb.append(" ms)");

      logger.info("Starting DistributionManager {}. {}",
          new Object[] {localAddress,
              (logger.isInfoEnabled(LogMarker.DM_MARKER) ? sb.toString() : "")});

      description = "Distribution manager on " + localAddress + " started at "
          + (new Date(System.currentTimeMillis()));

      finishedConstructor = true;
    } finally {
      if (!finishedConstructor && executors != null) {
        askThreadsToStop(); // fix for bug 42039
      }
    }
  }

  /**
   * Creates a new distribution manager
   *
   * @param system The distributed system to which this distribution manager will send messages.
   */
  private ClusterDistributionManager(InternalDistributedSystem system,
      RemoteTransportConfig transport,
      AlertingService alertingService,
      final MembershipLocator<InternalDistributedMember> membershipLocator) {
    this(transport, system, alertingService, membershipLocator);

    boolean finishedConstructor = false;
    try {

      setIsStartupThread();

      startThreads();

      // Allow events to start being processed.
      distribution.startEventProcessing();
      for (;;) {
        getCancelCriterion().checkCancelInProgress(null);
        boolean interrupted = Thread.interrupted();
        try {
          distribution.waitForEventProcessing();
          break;
        } catch (InterruptedException e) {
          interrupted = true;
        } finally {
          if (interrupted) {
            Thread.currentThread().interrupt();
          }
        }
      }

      finishedConstructor = true;
    } finally {
      if (!finishedConstructor) {
        askThreadsToStop(); // fix for bug 42039
      }
    }
  }

  /**
   * Is this VM dedicated to administration (like a GUI console or a JMX agent)? If so, then it
   * creates {@link #ADMIN_ONLY_DM_TYPE} type distribution managers.
   *
   * @since GemFire 4.0
   */
  public static boolean isDedicatedAdminVM() {
    return isDedicatedAdminVM;
  }

  public static void setIsDedicatedAdminVM(boolean isDedicatedAdminVM) {
    ClusterDistributionManager.isDedicatedAdminVM = isDedicatedAdminVM;
  }

  private static Boolean isStartupThread() {
    return isStartupThread.get();
  }

  private static void setIsStartupThread() {
    ClusterDistributionManager.isStartupThread.set(Boolean.TRUE);
  }

  //////////////////// Instance Methods /////////////////////

  /**
   * Returns true if the two members are on the same equivalent host based on overlapping IP
   * addresses collected for all NICs during exchange of startup messages.
   *
   * @param member1 First member
   * @param member2 Second member
   */
  @Override
  public boolean areOnEquivalentHost(InternalDistributedMember member1,
      InternalDistributedMember member2) {
    Set<InetAddress> equivalents1 = getEquivalents(member1.getInetAddress());
    return equivalents1.contains(member2.getInetAddress());
  }

  /**
   * Set the host equivalencies for a given host. This overrides any previous information in the
   * tables.
   *
   * @param equivs list of InetAddress's that all point at same host
   */
  void setEquivalentHosts(Set<InetAddress> equivs) {
    Iterator<InetAddress> it = equivs.iterator();
    synchronized (equivalentHosts) {
      while (it.hasNext()) {
        equivalentHosts.put(it.next(), Collections.unmodifiableSet(equivs));
      }
    }
  }


  /**
   * Return all of the InetAddress's that are equivalent to the given one (same host)
   *
   * @param in host to match up
   * @return all the addresses thus equivalent
   */
  @Override
  public Set<InetAddress> getEquivalents(InetAddress in) {
    Set<InetAddress> result;
    synchronized (equivalentHosts) {
      result = equivalentHosts.get(in);
    }
    // DS 11/25/08 - It appears that when using VPN, the distributed member
    // id is the vpn address, but that doesn't show up in the equivalents.
    if (result == null) {
      result = Collections.singleton(in);
    }
    return result;
  }

  void setRedundancyZone(InternalDistributedMember member, String redundancyZone) {
    if (redundancyZone != null && !redundancyZone.equals("")) {
      redundancyZones.put(member, redundancyZone);
    }
    if (member != getDistributionManagerId()) {
      String relationship = areInSameZone(getDistributionManagerId(), member) ? "" : "not ";
      logger.info("Member {} is {}equivalent or in the same redundancy zone.",
          member, relationship);
    }
  }

  /**
   * Set the flag indicating that we should enforce unique zones. If we are already enforcing unique
   * zones, keep it that way.
   */
  void setEnforceUniqueZone(boolean enforceUniqueZone) {
    this.enforceUniqueZone |= enforceUniqueZone;
  }

  @Override
  public boolean enforceUniqueZone() {
    return enforceUniqueZone;
  }

  @Override
  public String getRedundancyZone(InternalDistributedMember member) {
    return redundancyZones.get(member);
  }

  /**
   * Asserts that distributionManagerType is LOCAL, GEMFIRE, or ADMIN_ONLY. Also asserts that the
   * distributionManagerId (jgroups DistributedMember) has a VmKind that matches.
   */
  private void assertDistributionManagerType() {
    // Assert that dmType is one of the three DM types...
    int theDmType = getDMType();
    switch (theDmType) {
      case NORMAL_DM_TYPE:
      case LONER_DM_TYPE:
      case ADMIN_ONLY_DM_TYPE:
      case LOCATOR_DM_TYPE:
        break;
      default:
        Assert.assertTrue(false, "unknown distribution manager type");
    }

    // Assert InternalDistributedMember VmKind matches this DistributionManagerType...
    final InternalDistributedMember theId = getDistributionManagerId();
    final int vmKind = theId.getVmKind();
    if (theDmType != vmKind) {
      Assert.assertTrue(false,
          "InternalDistributedMember has a vmKind of " + vmKind + " instead of " + theDmType);
    }
  }

  @Override
  public int getDMType() {
    return dmType;
  }

  @Override
  public List<InternalDistributedMember> getViewMembers() {
    return distribution.getView().getMembers();
  }

  private boolean testMulticast() {
    return distribution.testMulticast();
  }

  /**
   * Need to do this outside the constructor so that the child constructor can finish.
   */
  private void startThreads() {
    system.setDM(this); // fix for bug 33362
    if (memberEventThread != null) {
      memberEventThread.start();
    }
    try {

      // And the distinguished guests today are...
      MembershipView<InternalDistributedMember> v = distribution.getView();
      logger.info("Initial (distribution manager) view, {}",
          String.valueOf(v));

      // Add them all to our view
      for (InternalDistributedMember internalDistributedMember : v.getMembers()) {
        addNewMember(internalDistributedMember);
      }

    } catch (Exception ex) {
      throw new InternalGemFireException(
          "Could not process initial view",
          ex);
    }
    try {
      executors.getWaitingThreadPool().execute(() -> {
        // call in background since it might need to send a reply
        // and we are not ready to send messages until startup is finished
        setIsStartupThread();
        readyForMessages();
      });
    } catch (VirtualMachineError err) {
      SystemFailure.initiateFailure(err);
      // If this ever returns, rethrow the error. We're poisoned
      // now, so don't let this thread continue.
      throw err;
    } catch (Throwable t) {
      // Whenever you catch Error or Throwable, you must also
      // catch VirtualMachineError (see above). However, there is
      // _still_ a possibility that you are dealing with a cascading
      // error condition, so you also need to check to see if the JVM
      // is still usable:
      SystemFailure.checkFailure();
      logger.fatal("Uncaught exception calling readyForMessages", t);
    }
  }

  private void readyForMessages() {
    synchronized (this) {
      readyForMessages = true;
      notifyAll();
    }
    distribution.startEventProcessing();
  }

  private void waitUntilReadyForMessages() {
    if (readyForMessages) {
      return;
    }
    synchronized (this) {
      while (!readyForMessages) {
        stopper.checkCancelInProgress(null);
        boolean interrupted = Thread.interrupted();
        try {
          wait();
        } catch (InterruptedException e) {
          interrupted = true;
          stopper.checkCancelInProgress(e);
        } finally {
          if (interrupted) {
            Thread.currentThread().interrupt();
          }
        }
      }
    } // synchronized
  }

  /**
   * Call when the DM is ready to send messages.
   */
  private void readyToSendMsgs() {
    synchronized (readyToSendMsgsLock) {
      readyToSendMsgs = true;
      readyToSendMsgsLock.notifyAll();
    }
  }

  /**
   * Return when DM is ready to send out messages.
   *
   * @param msg the messsage that is currently being sent
   */
  private void waitUntilReadyToSendMsgs(DistributionMessage msg) {
    if (readyToSendMsgs) {
      return;
    }
    // another process may have been started in the same view, so we need
    // to be responsive to startup messages and be able to send responses
    if (msg instanceof AdminMessageType) {
      return;
    }
    if (isStartupThread() == Boolean.TRUE) {
      // let the startup thread send messages
      // the only case I know of that does this is if we happen to log a
      // message during startup and an alert listener has registered.
      return;
    }

    synchronized (readyToSendMsgsLock) {
      while (!readyToSendMsgs) {
        stopper.checkCancelInProgress(null);
        boolean interrupted = Thread.interrupted();
        try {
          readyToSendMsgsLock.wait();
        } catch (InterruptedException e) {
          interrupted = true;
          stopper.checkCancelInProgress(e);
        } finally {
          if (interrupted) {
            Thread.currentThread().interrupt();
          }
        }
      } // for
    } // synchronized
  }


  @Override
  public void forceUDPMessagingForCurrentThread() {
    distribution.forceUDPMessagingForCurrentThread();
  }


  @Override
  public void releaseUDPMessagingForCurrentThread() {
    distribution.releaseUDPMessagingForCurrentThread();
  }

  /**
   * Did an exception occur in one of the threads launched by this distribution manager?
   */
  @Override
  public boolean exceptionInThreads() {
    return exceptionInThreads
        || LoggingUncaughtExceptionHandler.getUncaughtExceptionsCount() > 0;
  }

  /**
   * Clears the boolean that determines whether or not an exception occurred in one of the worker
   * threads. This method should be used for testing purposes only!
   */
  @Override
  public void clearExceptionInThreads() {
    exceptionInThreads = false;
    LoggingUncaughtExceptionHandler.clearUncaughtExceptionsCount();
  }

  /**
   * Returns the current "cache time" in milliseconds since the epoch. The "cache time" takes into
   * account skew among the local clocks on the various machines involved in the cache.
   */
  @Override
  public long cacheTimeMillis() {
    return system.getClock().cacheTimeMillis();
  }


  @Override
  public DistributedMember getMemberWithName(String name) {
    for (DistributedMember id : getViewMembers()) {
      if (Objects.equals(id.getName(), name)) {
        return id;
      }
    }
    if (Objects.equals(localAddress.getName(), name)) {
      return localAddress;
    }
    return null;
  }

  /**
   * Returns the id of this distribution manager.
   */
  @Override
  public InternalDistributedMember getDistributionManagerId() {
    return localAddress;
  }

  /**
   * Returns an unmodifiable set containing the identities of all of the known (non-admin-only)
   * distribution managers.
   */
  @Override
  public Set<InternalDistributedMember> getDistributionManagerIds() {
    return distribution.getMembersNotShuttingDown();
  }

  /**
   * Adds the entry in {@link #hostedLocatorsAll} for a member with one or more hosted locators. The
   * value is a collection of host[port] strings. If a bind-address was used for a locator then the
   * form is bind-addr[port].
   *
   * @since GemFire 6.6.3
   */
  @Override
  public void addHostedLocators(InternalDistributedMember member, Collection<String> locators,
      boolean isSharedConfigurationEnabled) {
    synchronized (membersLock) {
      if (locators == null || locators.isEmpty()) {
        throw new IllegalArgumentException("Cannot use empty collection of locators");
      }
      if (hostedLocatorsAll.isEmpty()) {
        hostedLocatorsAll = new HashMap<>();
      }
      Map<InternalDistributedMember, Collection<String>> tmp =
          new HashMap<>(hostedLocatorsAll);
      tmp.remove(member);
      tmp.put(member, locators);
      tmp = Collections.unmodifiableMap(tmp);
      hostedLocatorsAll = tmp;

      if (isSharedConfigurationEnabled) {
        if (hostedLocatorsWithSharedConfiguration.isEmpty()) {
          hostedLocatorsWithSharedConfiguration = new HashMap<>();
        }
        tmp = new HashMap<>(hostedLocatorsWithSharedConfiguration);
        tmp.remove(member);
        tmp.put(member, locators);
        tmp = Collections.unmodifiableMap(tmp);
        hostedLocatorsWithSharedConfiguration = tmp;
      }

    }
  }


  private void removeHostedLocators(InternalDistributedMember member) {
    synchronized (membersLock) {
      if (hostedLocatorsAll.containsKey(member)) {
        Map<InternalDistributedMember, Collection<String>> tmp =
            new HashMap<>(hostedLocatorsAll);
        tmp.remove(member);
        if (tmp.isEmpty()) {
          tmp = Collections.emptyMap();
        } else {
          tmp = Collections.unmodifiableMap(tmp);
        }
        hostedLocatorsAll = tmp;
      }
      if (hostedLocatorsWithSharedConfiguration.containsKey(member)) {
        Map<InternalDistributedMember, Collection<String>> tmp =
            new HashMap<>(
                hostedLocatorsWithSharedConfiguration);
        tmp.remove(member);
        if (tmp.isEmpty()) {
          tmp = Collections.emptyMap();
        } else {
          tmp = Collections.unmodifiableMap(tmp);
        }
        hostedLocatorsWithSharedConfiguration = tmp;
      }
    }
  }


  /**
   * Gets the value in {@link #hostedLocatorsAll} for a member with one or more hosted locators. The
   * value is a collection of host[port] strings. If a bind-address was used for a locator then the
   * form is bind-addr[port].
   *
   * @since GemFire 6.6.3
   */
  @Override
  public Collection<String> getHostedLocators(InternalDistributedMember member) {
    synchronized (membersLock) {
      return hostedLocatorsAll.get(member);
    }
  }

  /**
