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

import static org.apache.geode.distributed.internal.membership.api.MembershipConfig.MEMBER_REQUEST_COLLECTION_INTERVAL;
import static org.apache.geode.internal.serialization.DataSerializableFixedID.JOIN_REQUEST;
import static org.apache.geode.internal.serialization.DataSerializableFixedID.LEAVE_REQUEST_MESSAGE;
import static org.apache.geode.internal.serialization.DataSerializableFixedID.REMOVE_MEMBER_REQUEST;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Objects;
import java.util.Properties;
import java.util.Set;
import java.util.TimerTask;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Future;

import org.apache.commons.lang3.StringUtils;
import org.apache.logging.log4j.Logger;

import org.apache.geode.annotations.VisibleForTesting;
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.MemberStartupException;
import org.apache.geode.distributed.internal.membership.api.MembershipClosedException;
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.gms.GMSMembershipView;
import org.apache.geode.distributed.internal.membership.gms.Services;
import org.apache.geode.distributed.internal.membership.gms.interfaces.JoinLeave;
import org.apache.geode.distributed.internal.membership.gms.locator.FindCoordinatorRequest;
import org.apache.geode.distributed.internal.membership.gms.locator.FindCoordinatorResponse;
import org.apache.geode.distributed.internal.membership.gms.messages.AbstractGMSMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.HasMemberID;
import org.apache.geode.distributed.internal.membership.gms.messages.InstallViewMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.JoinRequestMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.JoinResponseMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.LeaveRequestMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.NetworkPartitionMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.RemoveMemberMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.ViewAckMessage;
import org.apache.geode.distributed.internal.tcpserver.HostAndPort;
import org.apache.geode.distributed.internal.tcpserver.TcpClient;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.logging.internal.executors.LoggingExecutors;
import org.apache.geode.logging.internal.executors.LoggingThread;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * GMSJoinLeave is responsible for finding the membership coordinator and joining the
 * cluster. It is also responsible for informing other members of the cluster when
 * Membership is shutting down.
 * <p>
 * One member of the cluster must also take on the role of coordinator. This is typically
 * the oldest member of the cluster. The coordinator is responsible for sending out new
 * membership "views" that announce cluster membership.
 */
public class GMSJoinLeave<ID extends MemberIdentifier> implements JoinLeave<ID> {

  public static final String BYPASS_DISCOVERY_PROPERTY =
      GeodeGlossary.GEMFIRE_PREFIX + "bypass-discovery";

  /**
   * amount of time to wait for responses to FindCoordinatorRequests
   */
  private static final int DISCOVERY_TIMEOUT =
      Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "discovery-timeout", 3000);

  /**
   * amount of time to sleep before trying to join after a failed attempt
   */
  public static final int JOIN_RETRY_SLEEP =
      Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "join-retry-sleep", 1000);

  /**
   * amount of time to sleep before trying to contact a locator after a failed attempt
   */
  public static final int FIND_LOCATOR_RETRY_SLEEP = 1_000;

  /**
   * time to wait for a broadcast message to be transmitted by jgroups
   */
  private static final long BROADCAST_MESSAGE_SLEEP_TIME =
      Long.getLong(GeodeGlossary.GEMFIRE_PREFIX + "broadcast-message-sleep-time", 1000);

  /**
   * if the locators don't know who the coordinator is we send find-coord requests to this many
   * nodes
   */
  private static final int MAX_DISCOVERY_NODES =
      Integer.getInteger(GeodeGlossary.GEMFIRE_PREFIX + "max-discovery-nodes", 30);

  /**
   * interval for broadcasting the current view to members in case they didn't get it the first time
   */
  private static final long VIEW_BROADCAST_INTERVAL =
      Long.getLong(GeodeGlossary.GEMFIRE_PREFIX + "view-broadcast-interval", 60000);

  /**
   * membership logger
   */
  private static final Logger logger = Services.getLogger();
  private static final boolean ALLOW_OLD_VERSION_FOR_TESTING = Boolean
      .getBoolean(GeodeGlossary.GEMFIRE_PREFIX + "allow_old_members_to_join_for_testing");

  /**
   * the view ID where I entered into membership
   */
  private int birthViewId;

  /**
   * my address
   */
  private ID localAddress;

  private Services<ID> services;

  /**
   * have I connected to the distributed system?
   */
  private volatile boolean isJoined;

  /**
   * guarded by viewInstallationLock
   */
  @VisibleForTesting
  volatile boolean isCoordinator;

  /**
   * a synch object that guards view installation
   */
  private final Object viewInstallationLock = new Object();

  /**
   * the currently installed view. Guarded by viewInstallationLock
   */
  @VisibleForTesting
  volatile GMSMembershipView<ID> currentView;

  /**
   * the previous view
   **/
  private volatile GMSMembershipView<ID> previousView;

  /**
   * members who we have been declared dead in the current view
   */
  private final Set<ID> removedMembers = new HashSet<>();

  /**
   * members who we've received a leave message from
   **/
  private final Set<ID> leftMembers = new HashSet<>();

  /**
   * a new view being installed
   */
  private volatile GMSMembershipView<ID> preparedView;

  /**
   * the last view that conflicted with view preparation
   */
  private GMSMembershipView<ID> lastConflictingView;

  private List<HostAndPort> locators;

  /**
   * a list of join/leave/crashes
   */
  private final List<AbstractGMSMessage<ID>> viewRequests = new LinkedList<>();

  /**
   * the established request collection jitter. This can be overridden for testing with
   * delayViewCreationForTest
   */
  long requestCollectionInterval = MEMBER_REQUEST_COLLECTION_INTERVAL;

  /**
   * collects the response to a join request
   */
  private final JoinResponseMessage<ID>[] joinResponse = new JoinResponseMessage[1];

  /**
   * collects responses to new views
   */
  ViewReplyProcessor viewProcessor = new ViewReplyProcessor(false);

  /**
   * collects responses to view preparation messages
   */
  ViewReplyProcessor prepareProcessor = new ViewReplyProcessor(true);

  /**
   * whether quorum checks can cause a forced-disconnect
   */
  private boolean quorumRequired = false;

  /**
   * timeout in receiving view acknowledgement
   */
  private long viewAckTimeout;

  /**
   * background thread that creates new membership views
   */
  private ViewCreator viewCreator;

  /**
   * am I shutting down?
   */
  private volatile boolean isStopping;

  /**
   * state of collected artifacts during discovery
   */
  final SearchState<ID> searchState = new SearchState<>();

  /**
   * a collection used to detect unit testing
   */
  Set<String> unitTesting = new HashSet<>();

  /**
   * a test hook to make this member unresponsive
   */
  private volatile boolean playingDead;

  /**
   * the view where quorum was most recently lost
   */
  GMSMembershipView<ID> quorumLostView;

  /**
   * for messaging locator
   */
  private final TcpClient locatorClient;

  public GMSJoinLeave(final TcpClient locatorClient) {
    this.locatorClient = locatorClient;
  }

  public List<HostAndPort> getLocators() {
    return locators;
  }

  static class SearchState<ID extends MemberIdentifier> {
    public int joinedMembersContacted;
    Set<ID> alreadyTried = new HashSet<>();
    Set<ID> registrants = new HashSet<>();
    ID possibleCoordinator;
    int viewId = -100;
    int locatorsContacted = 0;
    boolean hasContactedAJoinedLocator;
    GMSMembershipView<ID> view;
    int lastFindCoordinatorInViewId = -1000;
    final Set<FindCoordinatorResponse<ID>> responses = new HashSet<>();
    public int responsesExpected;
    Exception lastLocatorException;

    void cleanup() {
      alreadyTried.clear();
      possibleCoordinator = null;
      view = null;
      lastLocatorException = null;
      synchronized (responses) {
        responses.clear();
      }
    }

    public String toString() {
      return "locatorsContacted=" + locatorsContacted
          + "; findInViewResponses=" + joinedMembersContacted
          + "; alreadyTried=" + alreadyTried + "; registrants="
          + registrants + "; possibleCoordinator=" + possibleCoordinator
          + "; viewId=" + viewId + "; hasContactedAJoinedLocator="
          + hasContactedAJoinedLocator + "; view=" + view + "; responses="
          + responses;
    }
  }

  Object getViewInstallationLock() {
    return viewInstallationLock;
  }

  @VisibleForTesting
  public static int getMinimumRetriesBeforeBecomingCoordinator(int locatorsSize) {
    return locatorsSize * 2;
  }

  /**
   * attempt to join the distributed system loop send a join request to a locator & get a response
   * <p>
   * If the response indicates there's no coordinator it will contain a set of members that have
   * recently contacted it. The "oldest" member is selected as the coordinator based on ID sort
   * order.
   */
  @Override
  public void join() throws MemberStartupException {

    try {
      if (Boolean.getBoolean(BYPASS_DISCOVERY_PROPERTY)) {
        synchronized (viewInstallationLock) {
          becomeCoordinator();
        }
        return;
      }

      SearchState<ID> state = searchState;

      long locatorWaitTime = ((long) services.getConfig().getLocatorWaitTime()) * 1000L;
      long timeout = services.getConfig().getJoinTimeout();
      logger.debug("join timeout is set to {}", timeout);
      long startTime = System.currentTimeMillis();
      long locatorGiveUpTime = startTime + locatorWaitTime;
      long giveupTime = startTime + timeout;
      int minimumRetriesBeforeBecomingCoordinator =
          getMinimumRetriesBeforeBecomingCoordinator(locators.size());

      for (int tries = 0; !isJoined && !isStopping; tries++) {
        logger.debug("searching for the membership coordinator");
        boolean found = findCoordinator();
        logger.info("Discovery state after looking for membership coordinator is {}",
            state);
        long now = System.currentTimeMillis();
        if (found) {
          logger.info("found possible coordinator {}", state.possibleCoordinator);
          if (localAddress.preferredForCoordinator()
              && state.possibleCoordinator.equals(localAddress)) {
            // if we haven't contacted a member of a cluster maybe this node should
            // become the coordinator.
            if (state.joinedMembersContacted <= 0 && ((now >= locatorGiveUpTime &&
                tries >= minimumRetriesBeforeBecomingCoordinator) ||
                state.locatorsContacted >= locators.size())) {
              synchronized (viewInstallationLock) {
                becomeCoordinator();
              }
              return;
            }
          } else {
            if (attemptToJoin()) {
              return;
            }
            if (isStopping) {
              break;
            }
            if (!state.possibleCoordinator.equals(localAddress)) {
              state.alreadyTried.add(state.possibleCoordinator);
            }
            if (System.currentTimeMillis() > giveupTime) {
              break;
            }
          }
        } else {
          if (state.locatorsContacted <= 0) {
            if (now > locatorGiveUpTime) {
              // break out of the loop and return false
              break;
            }
            tries = 0;
            giveupTime = now + timeout;
          } else if (now > giveupTime) {
            break;
          }
        }
        if (found && !state.hasContactedAJoinedLocator) {
          try {
            if (pauseIfThereIsNoCoordinator(state.possibleCoordinator.getVmViewId(),
                JOIN_RETRY_SLEEP)) {
              // since we were given a coordinator that couldn't be used we should keep trying
              tries = 0;
              giveupTime = System.currentTimeMillis() + timeout;
            }
          } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new MembershipConfigurationException(
                "Retry sleep interrupted. Giving up on joining the distributed system.");
          }
        }
      } // for

      if (!isJoined) {
        logger.debug("giving up attempting to join the distributed system after "
            + (System.currentTimeMillis() - startTime) + "ms");

        // to preserve old behavior we need to throw a MemberStartupException if
        // unable to contact any of the locators
        if (state.hasContactedAJoinedLocator) {
          throw new MemberStartupException("Unable to join the distributed system in "
              + (System.currentTimeMillis() - startTime) + "ms");
        }

        if (state.locatorsContacted == 0) {
          throw new MembershipConfigurationException(
              "Unable to join the distributed system. Could not contact any of the locators: "
                  + locators,
              state.lastLocatorException);
        }

        if (System.currentTimeMillis() > giveupTime) {
          throw new MembershipConfigurationException(
              "Unable to join the distributed system. Operation timed out");
        }
      }
      return;
    } finally {
      // notify anyone waiting on the address to be completed
      if (isJoined) {
        synchronized (localAddress) {
          localAddress.notifyAll();
        }
      }
      searchState.cleanup();
    }
  }

  boolean pauseIfThereIsNoCoordinator(int viewId, long retrySleep)
      throws InterruptedException {
    // if locators are restarting they may be handing out IDs from a stale view that
    // we should go through quickly. Otherwise we should sleep a bit to let failure
    // detection select a new coordinator
    if (viewId < 0) {
      // the process hasn't finished joining the cluster.
      logger.debug("sleeping for {} before making another attempt to find the coordinator",
          retrySleep);
      Thread.sleep(retrySleep);
    } else {
      // the member has joined the cluster.
      return true;
    }

    return false;
  }

  /**
   * send a join request and wait for a reply. Process the reply. This may throw a
   * MemberStartupException or an exception from the authenticator, if present.
   *
   * @return true if the attempt succeeded, false if it timed out
   */
  @edu.umd.cs.findbugs.annotations.SuppressWarnings(value = "WA_NOT_IN_LOOP")
  boolean attemptToJoin() throws MemberStartupException {
    SearchState<ID> state = searchState;

    // send a join request to the coordinator and wait for a response
    ID coord = state.possibleCoordinator;
    if (state.alreadyTried.contains(coord)) {
      logger.info("Probable coordinator is still {} - waiting for a join-response", coord);
    } else {
      logger.info("Attempting to join the distributed system through coordinator " + coord
          + " using address " + localAddress);
      int port = services.getHealthMonitor().getFailureDetectionPort();
      JoinRequestMessage<ID> req = new JoinRequestMessage<>(coord, localAddress,
          services.getAuthenticator().getCredentials(coord), port,
          services.getMessenger().getRequestId());
      services.getMessenger().send(req);
    }

    JoinResponseMessage<ID> response;
    try {
      response = waitForJoinResponse();
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      return false;
    }

    if (response == null) {
      if (!isJoined) {
        logger.debug("received no join response");
      }
      return isJoined;
    }

    logger.info("received join response {}", response);
    joinResponse[0] = null;
    String failReason = response.getRejectionMessage();
    if (failReason != null) {
      if (failReason.contains("Rejecting the attempt of a member using an older version")
          || failReason.contains("15806")
          || failReason.contains("ForcedDisconnectException")) {
        throw new MemberStartupException(failReason);
      }
      throw new SecurityException(failReason);
    }

    throw new RuntimeException("Join Request Failed with response " + response);
  }

  private JoinResponseMessage<ID> waitForJoinResponse() throws InterruptedException {
    JoinResponseMessage<ID> response;
    synchronized (joinResponse) {
      if (joinResponse[0] == null && !isJoined) {
        // Note that if we give up waiting but a response is on
        // the way we will get the new view and join that way.
        // See installView()
        long timeout = Math.max(services.getConfig().getMemberTimeout(),
            services.getConfig().getJoinTimeout() / 5);
        joinResponse.wait(timeout);
      }
      response = joinResponse[0];

      if (services.getConfig().getSecurityUDPDHAlgo().length() > 0) {
        if (response != null && response.getCurrentView() != null && !isJoined) {
          // reset joinResponse[0]
          joinResponse[0] = null;
          // we got view here that means either we have to wait for
          GMSMembershipView<ID> v = response.getCurrentView();
          ID coord = v.getCoordinator();
          if (searchState.alreadyTried.contains(coord)) {
            searchState.view = response.getCurrentView();
            // we already sent join request to it..so lets wait some more time here
            // assuming we got this response immediately, so wait for same timeout here..
            long timeout = Math.max(services.getConfig().getMemberTimeout(),
                services.getConfig().getJoinTimeout() / 5);
            joinResponse.wait(timeout);
            response = joinResponse[0];
          } else {
            // try on this coordinator
            searchState.view = response.getCurrentView();
            response = null;
          }
          searchState.view = v;
        }
        if (isJoined) {
          return null;
        }
      }
    }
    return response;
  }

  @Override
  public boolean isMemberLeaving(ID mbr) {
    if (getPendingRequestIDs(LEAVE_REQUEST_MESSAGE).contains(mbr)
        || getPendingRequestIDs(REMOVE_MEMBER_REQUEST).contains(mbr)
        || !currentView.contains(mbr)) {
      return true;
    }
    synchronized (removedMembers) {
      if (removedMembers.contains(mbr)) {
        return true;
      }
    }
    synchronized (leftMembers) {
      if (leftMembers.contains(mbr)) {
        return true;
      }
    }
    return false;
  }

  /**
   * process a join request from another member. If this is the coordinator this method will enqueue
   * the request for processing in another thread. If this is not the coordinator but the
   * coordinator is known, the message is forwarded to the coordinator.
   *
   * @param incomingRequest the request to be processed
   */
  void processJoinRequestMessage(JoinRequestMessage<ID> incomingRequest) {
    if (isStopping) {
      return;
    }

    logger.info("Received a join request from {}", incomingRequest.getMemberID());

    if (!ALLOW_OLD_VERSION_FOR_TESTING
        && incomingRequest.getMemberID().getVersionOrdinal() < KnownVersion.getCurrentVersion()
            .ordinal()) {
      logger.warn("detected an attempt to start a peer using an older version of the product {}",
          incomingRequest.getMemberID());
      JoinResponseMessage<ID> m =
          new JoinResponseMessage<>(
              "Rejecting the attempt of a member using an older version of the "
                  + "product to join the distributed system",
              incomingRequest.getRequestId());
      m.setRecipient(incomingRequest.getMemberID());
      services.getMessenger().send(m);
      return;
    }

    Object creds = incomingRequest.getCredentials();
    String rejection;
    try {
      rejection = services.getAuthenticator().authenticate(incomingRequest.getMemberID(),
          (Properties) creds);
    } catch (Exception e) {
      rejection = e.getMessage();
    }
    if (rejection != null && rejection.length() > 0) {
      JoinResponseMessage<ID> m = new JoinResponseMessage<>(rejection, 0);
      m.setRecipient(incomingRequest.getMemberID());
      services.getMessenger().send(m);
      return;
    }

    recordViewRequest(incomingRequest);
  }

  /**
   * Process a Leave request from another member. This may cause this member to become the new
   * membership coordinator. If this is the coordinator a new view will be triggered.
   *
   * @param incomingRequest the request to be processed
   */
  void processLeaveRequestMessage(LeaveRequestMessage<ID> incomingRequest) {
    if (isStopping) {
      return;
    }

    logger.info("received leave request from {} for {}", incomingRequest.getSender(),
        incomingRequest.getMemberID());

    GMSMembershipView<ID> v = currentView;
    if (v == null) {
      recordViewRequest(incomingRequest);
      return;
    }

    ID mbr = incomingRequest.getMemberID();

    logger.info(
        () -> "JoinLeave.processLeaveRequestMessage(LeaveRequestMessage) invoked.  isCoordinator="
            + isCoordinator
            + "; isStopping=" + isStopping + "; cancelInProgress="
            + services.getCancelCriterion().isCancelInProgress());

    if (!v.contains(mbr) && mbr.getVmViewId() < v.getViewId()) {
      logger.info("ignoring leave request from old member");
      return;
    }

    if (incomingRequest.getMemberID().equals(localAddress)) {
      logger.info("I am being told to leave the distributed system by {}",
          incomingRequest.getSender());
      forceDisconnect(incomingRequest.getReason());
      return;
    }

    if (!isCoordinator && !isStopping && !services.getCancelCriterion().isCancelInProgress()) {
      logger.info("Checking to see if I should become coordinator.  My address is {}",
          localAddress);
      GMSMembershipView<ID> check = new GMSMembershipView<>(v, v.getViewId() + 1);
      check.remove(mbr);
      synchronized (removedMembers) {
        check.removeAll(removedMembers);
        check.addCrashedMembers(removedMembers);
      }
      synchronized (leftMembers) {
        leftMembers.add(mbr);
        check.removeAll(leftMembers);
      }
      ID coordinator = check.getCoordinator();
      logger.info("View with removed and left members removed is {} and coordinator would be {}",
          check, coordinator);
      if (coordinator.equals(localAddress)) {
        synchronized (viewInstallationLock) {
          becomeCoordinator(mbr);
        }
      }
    } else {
      if (!isStopping && !services.getCancelCriterion().isCancelInProgress()) {
        recordViewRequest(incomingRequest);
        viewProcessor.processLeaveRequest(incomingRequest.getMemberID());
        prepareProcessor.processLeaveRequest(incomingRequest.getMemberID());
      }
    }
  }

  /**
   * Process a Remove request from another member. This may cause this member to become the new
   * membership coordinator. If this is the coordinator a new view will be triggered.
   *
   * @param incomingRequest the request to process
   */
  void processRemoveMemberMessage(RemoveMemberMessage<ID> incomingRequest) {
    if (isStopping) {
      return;
    }

    GMSMembershipView<ID> v = currentView;
    boolean fromMe =
        incomingRequest.getSender() == null || incomingRequest.getSender().equals(localAddress);

    ID mbr = incomingRequest.getMemberID();

    if (v != null && !v.contains(incomingRequest.getSender())) {
      logger.info("Membership ignoring removal request for " + mbr + " from non-member "
          + incomingRequest.getSender());
      return;
    }

    if (v == null) {
      // not yet a member
      return;
    }

    if (!fromMe) {
      logger.info("Membership received a request to remove " + mbr + " from "
          + incomingRequest.getSender() + " reason=" + incomingRequest.getReason());
    }

    if (mbr.equals(localAddress)) {
      // oops - I've been kicked out
      forceDisconnect(incomingRequest.getReason());
      return;
    }

    if (getPendingRequestIDs(REMOVE_MEMBER_REQUEST).contains(mbr)) {
      logger.debug("ignoring removal request as I already have a removal request for this member");
      return;
    }

    if (!isCoordinator && !isStopping && !services.getCancelCriterion().isCancelInProgress()) {
      logger.debug("Checking to see if I should become coordinator");
      GMSMembershipView<ID> check = new GMSMembershipView<>(v, v.getViewId() + 1);
      synchronized (removedMembers) {
        removedMembers.add(mbr);
        check.addCrashedMembers(removedMembers);
        check.removeAll(removedMembers);
      }
      synchronized (leftMembers) {
        check.removeAll(leftMembers);
      }
      if (check.getCoordinator().equals(localAddress)) {
        synchronized (viewInstallationLock) {
          becomeCoordinator(mbr);
        }
      }
    } else {
      if (!isStopping && !services.getCancelCriterion().isCancelInProgress()) {
        // suspect processing tends to get carried away sometimes during
        // shutdown (especially shutdownAll), so we check for a scheduled shutdown
        // message
        if (!getPendingRequestIDs(LEAVE_REQUEST_MESSAGE).contains(mbr)) {
          recordViewRequest(incomingRequest);
          viewProcessor.processRemoveRequest(mbr);
          prepareProcessor.processRemoveRequest(mbr);
        }
      }
      if (isCoordinator) {
        if (!v.contains(mbr)) {
          // removing a rogue process
          RemoveMemberMessage<ID> removeMemberMessage = new RemoveMemberMessage<>(mbr, mbr,
              incomingRequest.getReason());
          services.getMessenger().send(removeMemberMessage);
        }
      }
    }
  }

  @VisibleForTesting
  void recordViewRequest(AbstractGMSMessage<ID> request) {
    try {
      synchronized (viewRequests) {
        logger.debug("Recording the request to be processed in the next membership view");
        viewRequests.add(request);
        viewRequests.notifyAll();
      }
    } catch (RuntimeException | Error t) {
      logger.warn("unable to record a membership view request due to this exception", t);
      throw t;
    }
  }

  // for testing purposes, returns a copy of the view requests for verification
  List<AbstractGMSMessage<ID>> getViewRequests() {
    synchronized (viewRequests) {
      return new LinkedList<>(viewRequests);
    }
  }

  // for testing purposes, returns the view-creation thread
  ViewCreator getViewCreator() {
    return viewCreator;
  }

  /**
   * Yippeee - I get to be the coordinator
   */
  void becomeCoordinator() { // package access for unit testing
    becomeCoordinator(null);
  }

  /**
   * Test hook for delaying the creation of new views. This should be invoked before this member
   * becomes coordinator and creates its ViewCreator thread.
   *
   */
  public void delayViewCreationForTest(int millis) {
    requestCollectionInterval = millis;
  }

  /**
   * Transitions this member into the coordinator role. This must be invoked under a synch on
   * viewInstallationLock that was held at the time the decision was made to become coordinator so
   * that the decision is atomic with actually becoming coordinator.
   *
   * @param oldCoordinator may be null
   */
  private void becomeCoordinator(ID oldCoordinator) {

    assert Thread.holdsLock(viewInstallationLock);

    if (isCoordinator) {
      return;
    }

    logger.info("This member is becoming the membership coordinator with address {}", localAddress);
    isCoordinator = true;
    org.apache.geode.distributed.internal.membership.gms.interfaces.Locator<ID> locator =
        services.getLocator();
    if (locator != null) {
      locator.setIsCoordinator(true);
    }
    if (currentView == null) {
      // create the initial membership view
      GMSMembershipView<ID> newView = new GMSMembershipView<>(localAddress);
      newView.setFailureDetectionPort(localAddress,
          services.getHealthMonitor().getFailureDetectionPort());
      localAddress.setVmViewId(0);
      installView(newView);
      isJoined = true;
      createAndStartViewCreator(newView);
      startViewBroadcaster();
    } else {
      // create and send out a new view
      GMSMembershipView<ID> newView = copyCurrentViewAndAddMyAddress(oldCoordinator);
      createAndStartViewCreator(newView);
      startViewBroadcaster();
    }
  }

  private void createAndStartViewCreator(GMSMembershipView<ID> newView) {
    if (viewCreator == null || viewCreator.isShutdown()) {
      services.getMessenger().initClusterKey();
      viewCreator = new ViewCreator("Geode Membership View Creator");
      if (newView != null) {
        viewCreator.setInitialView(newView, newView.getNewMembers(),
            newView.getShutdownMembers(),
            newView.getCrashedMembers());
      }
      logger.info("ViewCreator starting on:" + localAddress);
      viewCreator.start();
    }
  }

  private GMSMembershipView<ID> copyCurrentViewAndAddMyAddress(ID oldCoordinator) {
    boolean testing = unitTesting.contains("noRandomViewChange");
    GMSMembershipView<ID> newView;
    Set<ID> leaving = new HashSet<>();
    Set<ID> removals;
    synchronized (viewInstallationLock) {
      int rand = testing ? 0 : GMSMembershipView.RANDOM.nextInt(10);
      int viewNumber = currentView.getViewId() + 5 + rand;
      if (localAddress.getVmViewId() < 0) {
        localAddress.setVmViewId(viewNumber);
      }
      List<ID> mbrs = new ArrayList<>(currentView.getMembers());
      if (!mbrs.contains(localAddress)) {
        mbrs.add(localAddress);
      }
      synchronized (removedMembers) {
        removals = new HashSet<>(removedMembers);
      }
      synchronized (leftMembers) {
        leaving.addAll(leftMembers);
      }
      if (oldCoordinator != null && !removals.contains(oldCoordinator)) {
        leaving.add(oldCoordinator);
      }
      mbrs.removeAll(removals);
      mbrs.removeAll(leaving);
      newView = new GMSMembershipView<>(localAddress, viewNumber, mbrs, leaving, removals);
      newView.setFailureDetectionPorts(currentView);
      newView.setPublicKeys(currentView);
      newView.setFailureDetectionPort(localAddress,
          services.getHealthMonitor().getFailureDetectionPort());
    }
    return newView;
  }

  private void sendRemoveMessages(List<ID> removals, List<String> reasons,
      Set<ID> oldIds) {
    Iterator<String> reason = reasons.iterator();
    for (ID mbr : removals) {
      // if olds not contains mbr then send remove request
      if (!oldIds.contains(mbr)) {
        RemoveMemberMessage<ID> response = new RemoveMemberMessage<>(mbr, mbr, reason.next());
        services.getMessenger().send(response);
      } else {
        reason.next();
      }
    }
  }

  boolean isShuttingDown() {
