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


import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketTimeoutException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Collections;
import java.util.Date;
import java.util.Enumeration;
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.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

import org.apache.logging.log4j.Logger;
import org.jgroups.util.UUID;

import org.apache.geode.annotations.VisibleForTesting;
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.api.MembershipStatistics;
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.HealthMonitor;
import org.apache.geode.distributed.internal.membership.gms.messages.AbstractGMSMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.FinalCheckPassedMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.HeartbeatMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.HeartbeatRequestMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.SuspectMembersMessage;
import org.apache.geode.distributed.internal.membership.gms.messages.SuspectRequest;
import org.apache.geode.distributed.internal.tcpserver.ConnectionWatcher;
import org.apache.geode.distributed.internal.tcpserver.HostAndPort;
import org.apache.geode.distributed.internal.tcpserver.TcpSocketCreator;
import org.apache.geode.internal.lang.utils.JavaWorkarounds;
import org.apache.geode.internal.serialization.KnownVersion;
import org.apache.geode.logging.internal.executors.LoggingExecutors;
import org.apache.geode.util.internal.GeodeGlossary;

/**
 * Failure Detection
 * <p>
 * This class make sure that each member is alive and communicating to this member. To make sure
 * that we create the ring of members based on current view. On this ring, each member make sure
 * that next-member in ring is communicating with it. For that we record last message timestamp from
 * next-member. And if it sees this member has not communicated in last period(member-timeout) then
 * we check whether this member is still alive or not. Based on that we informed probable
 * coordinators to remove that member from view.
 * <p>
 * It has {@link #suspect(MemberIdentifier, String)} api, which can be used to initiate
 * suspect processing for any member. First is checks whether the member is responding or not. Then
 * it informs probable coordinators to remove that member from view.
 * <p>
 * It has {@link HealthMonitor#checkIfAvailable(MemberIdentifier, String, boolean)} api to
 * see if that member is
 * alive. Then based on removal flag it initiates the suspect processing for that member.
 */
@SuppressWarnings({"SynchronizationOnLocalVariableOrMethodParameter"})
public class GMSHealthMonitor<ID extends MemberIdentifier> implements HealthMonitor<ID> {

  private final TcpSocketCreator socketCreator;
  private Services<ID> services;
  private volatile GMSMembershipView<ID> currentView;
  private volatile ID nextNeighbor;

  long memberTimeout;
  private volatile boolean isStopping = false;
  private final AtomicInteger requestId = new AtomicInteger();

  /**
   * membership logger
   */
  private static final Logger logger = Services.getLogger();

  /**
   * The number of recipients of periodic heartbeats. The recipients will be selected from the
   * members that are likely to be monitoring this member.
   */
  private static final int NUM_HEARTBEATS = Integer.getInteger("geode.heartbeat-recipients", 2);

  /**
   * Member activity will be recorded per interval/period. Timer task will set interval's starting
   * time. Each interval will be member-timeout/LOGICAL_INTERVAL. LOGICAL_INTERVAL may be configured
   * via a system property with a default of 2. At least 1 interval is needed.
   */
  public static final int LOGICAL_INTERVAL =
      Integer.getInteger("geode.logical-message-received-interval", 2);

  /**
   * stall time to wait for members leaving concurrently
   */
  public static final long MEMBER_SUSPECT_COLLECTION_INTERVAL =
      Long.getLong("geode.suspect-member-collection-interval", 200);

  /**
   * A millisecond clock reading used to mark the last time a peer made contact.
   */
  @VisibleForTesting
  volatile long currentTimeStamp;

  /**
   * this member's ID
   */
  private ID localAddress;

  /**
   * Timestamp at which we last had contact from a member
   */
  final ConcurrentMap<ID, TimeStamp> memberTimeStamps =
      new ConcurrentHashMap<>();

  /**
   * Members currently being suspected and the view they were suspected in
   */
  private final ConcurrentHashMap<ID, GMSMembershipView<ID>> suspectedMemberIds =
      new ConcurrentHashMap<>();

  /**
   * Members undergoing final checks
   */
  @VisibleForTesting
  final List<ID> membersInFinalCheck =
      Collections.synchronizedList(new ArrayList<>(30));

  /**
   * Replies to messages
   */
  private final Map<Integer, Response> requestIdVsResponse = new ConcurrentHashMap<>();

  /**
   * Members suspected in a particular view
   */
  private final Map<GMSMembershipView<ID>, Set<SuspectRequest<ID>>> suspectRequestsInView =
      new HashMap<>();

  private ScheduledExecutorService scheduler;

  private ExecutorService checkExecutor;

  /**
   * to stop check scheduler
   */
  private ScheduledFuture<?> monitorFuture;

  /**
   * test hook
   */
  private volatile boolean playingDead = false;

  /**
   * test hook
   */
  private volatile boolean beingSick = false;

  // For TCP check
  private ExecutorService serverSocketExecutor;
  static final int OK = 0x7B;
  static final int ERROR = 0x00;
  private volatile int socketPort;
  private volatile ServerSocket serverSocket;

  /**
   * Statistics about health monitor
   */
  private MembershipStatistics stats;

  /**
   * Interval to run the Monitor task
   */
  private long monitorInterval;

  /**
   * /**
   * this class is to avoid garbage
   */
  static class TimeStamp {

    private volatile long timeStamp;

    TimeStamp(long timeStamp) {
      this.timeStamp = timeStamp;
    }

    public long getTime() {
      return timeStamp;
    }

    public void setTime(long timeStamp) {
      this.timeStamp = timeStamp;
    }
  }

  /***
   * This class sets start interval timestamp to record the activity of all members. That is used by
   * {@link GMSHealthMonitor#contactedBy(MemberIdentifier)} to record the activity of
   * member.
   *
   * It initiates the suspect processing for next neighbour if it doesn't see any activity from that
   * member in last interval(member-timeout)
   */
  private class Monitor implements Runnable {
    /**
     * Here we use the same threshold for detecting JVM pauses as the StatSampler
     */
    private final long MONITOR_DELAY_THRESHOLD =
        Long.getLong(GeodeGlossary.GEMFIRE_PREFIX + "statSamplerDelayThreshold", 3000);


    final long memberTimeoutInMillis;

    public Monitor(long memberTimeout) {
      memberTimeoutInMillis = memberTimeout;
    }

    @Override
    public void run() {

      ID neighbor = nextNeighbor;
      if (logger.isDebugEnabled()) {
        logger.debug("cluster health monitor invoked with {}", neighbor);
      }
      try {
        if (isStopping) {
          return;
        }

        // TODO - why are we taking two clock readings and setting currentTimeStamp twice?
        long currentTime = System.currentTimeMillis();
        // this is the start of interval to record member activity
        currentTimeStamp = currentTime;

        long oldTimeStamp = currentTimeStamp;
        currentTimeStamp = System.currentTimeMillis();

        GMSMembershipView<ID> myView = currentView;
        if (myView == null) {
          return;
        }

        if (currentTimeStamp - oldTimeStamp > monitorInterval + MONITOR_DELAY_THRESHOLD) {
          // delay in running this task - don't suspect anyone for a while
          logger.info(
              "Failure detector has noticed a JVM pause and is giving all members a heartbeat in view {}",
              currentView);
          for (ID member : myView.getMembers()) {
            contactedBy(member);
          }
          return;
        }

        if (neighbor != null) {
          TimeStamp nextNeighborTS;
          synchronized (GMSHealthMonitor.this) {
            nextNeighborTS = memberTimeStamps.get(neighbor);
          }

          if (nextNeighborTS == null) {
            logger.debug("timestamp for {} was found null - setting current time as timestamp",
                neighbor);
            TimeStamp customTS = new TimeStamp(currentTime);
            memberTimeStamps.put(neighbor, customTS);
            return;
          }

          long interval = memberTimeoutInMillis / GMSHealthMonitor.LOGICAL_INTERVAL;
          long lastTS = currentTime - nextNeighborTS.getTime();
          if (lastTS + interval >= memberTimeoutInMillis) {
            logger.debug("Checking member {} ", neighbor);
            // now do check request for this member;
            checkMember(neighbor);
          }
        }
      } finally {
        if (logger.isDebugEnabled()) {
          logger.debug("cluster health monitor pausing");
        }
      }
    }
  }

  /***
   * Check thread waits on this object for response. It puts requestId in requestIdVsResponse map.
   * Response will have requestId, which is used to get ResponseObject. Then it is used to notify
   * waiting thread.
   */
  private class Response {

    private AbstractGMSMessage<ID> responseMsg;

    public AbstractGMSMessage<ID> getResponseMsg() {
      return responseMsg;
    }

    public void setResponseMsg(AbstractGMSMessage<ID> responseMsg) {
      this.responseMsg = responseMsg;
    }

  }

  class ClientSocketHandler implements Runnable {

    private final Socket socket;

    public ClientSocketHandler(Socket socket) {
      this.socket = socket;
    }

    @Override
    public void run() {
      try (DataInputStream in = new DataInputStream(socket.getInputStream())) {
        socket.setTcpNoDelay(true);
        OutputStream out = socket.getOutputStream();
        @SuppressWarnings("UnusedVariable")
        short version = in.readShort();
        int vmViewId = in.readInt();
        long uuidLSBs = in.readLong();
        long uuidMSBs = in.readLong();
        stats.incFinalCheckRequestsReceived();
        stats.incTcpFinalCheckRequestsReceived();
        ID gmbr = localAddress;
        UUID myUUID = gmbr.getUUID();
        // during reconnect or rapid restart we will have a zero viewId but there may still
        // be an old ID in the membership view that we do not want to respond to
        int myVmViewId = gmbr.getVmViewId();
        if (playingDead) {
          logger.debug("HealthMonitor: simulating sick member in health check");
        } else if (uuidLSBs == myUUID.getLeastSignificantBits()
            && uuidMSBs == myUUID.getMostSignificantBits()
            && (vmViewId == myVmViewId || myVmViewId < 0)) {
          logger.debug("HealthMonitor: sending OK reply");
          out.write(OK);
          out.flush();
          socket.shutdownOutput();
          stats.incFinalCheckResponsesSent();
          stats.incTcpFinalCheckResponsesSent();
          logger.debug("HealthMonitor: server replied OK.");
        } else {
          if (logger.isDebugEnabled()) {
            logger.debug(
                "HealthMonitor: sending ERROR reply - my UUID is {},{} received is {},{}.  My viewID is {} received is {}",
                Long.toHexString(myUUID.getMostSignificantBits()),
                Long.toHexString(myUUID.getLeastSignificantBits()), Long.toHexString(uuidMSBs),
                Long.toHexString(uuidLSBs), myVmViewId, vmViewId);
          }
          out.write(ERROR);
          out.flush();
          socket.shutdownOutput();
          stats.incFinalCheckResponsesSent();
          stats.incTcpFinalCheckResponsesSent();
          logger.debug("HealthMonitor: server replied ERROR.");
        }
      } catch (IOException e) {
        // this is expected if it is a connection-timeout or other failure
        // to connect
      } catch (RuntimeException e) {
        logger.debug("Unexpected runtime exception", e);
        throw e;
      } catch (Error e) {
        logger.debug("Unexpected error", e);
        throw e;
      } finally {
        if (socket != null) {
          try {
            socket.close();
          } catch (IOException e) {
            // expected if the socket is already closed
          }
        }
      }
    }
  }

  public GMSHealthMonitor(final TcpSocketCreator socketCreator) {
    Objects.requireNonNull(socketCreator);
    this.socketCreator = socketCreator;
  }

  /*
   * Record the member activity for current time interval.
   */
  @Override
  public void contactedBy(ID sender) {
    contactedBy(sender, currentTimeStamp);
  }


  /**
   * Record member activity at a specified time
   */
  private void contactedBy(ID sender, long timeStamp) {
    final TimeStamp cTS = JavaWorkarounds.computeIfAbsent(memberTimeStamps, sender, (s) -> {
      return new TimeStamp(timeStamp);
    });
    if (cTS != null && cTS.getTime() < timeStamp) {
      cTS.setTime(timeStamp);
    }
    if (suspectedMemberIds.containsKey(sender)) {
      memberUnsuspected(sender);
      setNextNeighbor(currentView, null);
    }
  }


  private HeartbeatRequestMessage<ID> constructHeartbeatRequestMessage(
      final ID mbr) {
    final int reqId = requestId.getAndIncrement();
    final HeartbeatRequestMessage<ID> hrm = new HeartbeatRequestMessage<>(mbr, reqId);
    hrm.setRecipient(mbr);

    return hrm;
  }

  private void checkMember(final ID mbr) {
    final GMSMembershipView<ID> cv = currentView;

    // as check may take time
    setNextNeighbor(cv, mbr);

    // we need to check this member
    doAndIgnoreRejectedExecutionExceptionIfStopping(() -> checkExecutor.execute(() -> {
      boolean pinged;
      try {
        pinged = doCheckMember(mbr, true);
      } catch (MembershipClosedException e) {
        return;
      }

      if (!pinged) {
        String reason = "Member isn't responding to heartbeat requests";
        memberSuspected(localAddress, mbr, reason);
        initiateSuspicion(mbr, reason);
        setNextNeighbor(currentView, null);
      } else {
        logger.trace("Setting next neighbor as member {} has responded.", mbr);
        memberUnsuspected(mbr);
        // back to previous one
        setNextNeighbor(currentView, null);
      }
    }));
  }

  private void initiateSuspicion(ID mbr, String reason) {
    if (services.getJoinLeave().isMemberLeaving(mbr)) {
      return;
    }
    sendSuspectRequest(Collections.singletonList(new SuspectRequest<>(mbr, reason)));
  }

  /**
   * This method sends heartbeat request to other member and waits for member-timeout time for
   * response. If it doesn't see response then it returns false.
   */
  private boolean doCheckMember(ID member, boolean waitForResponse) {
    if (playingDead || beingSick) {
      // a member playingDead should not be sending messages to other
      // members, so we avoid sending heartbeat requests or suspect
      // messages by returning true.
      return true;
    }
    long startTime = System.currentTimeMillis();
    logger.debug("Requesting heartbeat from {}", member);
    final HeartbeatRequestMessage<ID> hrm = constructHeartbeatRequestMessage(member);
    Response pingResp = null;
    if (waitForResponse) {
      pingResp = new Response();
      requestIdVsResponse.put(hrm.getRequestId(), pingResp);
    } else {
      hrm.clearRequestId();
    }
    try {
      Set<ID> membersNotReceivedMsg = services.getMessenger().send(hrm);
      stats.incHeartbeatRequestsSent();
      if (membersNotReceivedMsg != null && membersNotReceivedMsg.contains(member)) {
        // member is not part of current view.
        logger.trace("Member {} is not part of current view.", member);
      } else if (waitForResponse) {
        synchronized (pingResp) {
          if (pingResp.getResponseMsg() == null) {
            pingResp.wait(memberTimeout);
          }
          TimeStamp ts = memberTimeStamps.get(member);
          if (ts != null && ts.getTime() > startTime) {
            return true;
          }
          if (pingResp.getResponseMsg() == null) {
            if (isStopping) {
              return true;
            }
            logger.debug("no heartbeat response received from {} and no recent activity", member);
            return false;
          } else {
            logger.trace("received heartbeat from {}", member);
            stats.incHeartbeatsReceived();
            if (ts != null) {
              ts.setTime(System.currentTimeMillis());
            }
            return true;
          }
        }
      }
    } catch (InterruptedException e) {
      logger.debug(
          "GMSHealthMonitor checking thread interrupted, while waiting for response from member: {} .",
          member);
    } finally {
      if (waitForResponse) {
        requestIdVsResponse.remove(hrm.getRequestId());
      }
    }
    return false;
  }

  /**
   * During final check, establish TCP connection between current member and suspect member. And
   * exchange PING/PONG message to see if the suspect member is still alive.
   *
   * @param suspectMember member that does not respond to HeartbeatRequestMessage
   * @return true if successfully exchanged PING/PONG with TCP connection, otherwise false.
   */
  boolean doTCPCheckMember(ID suspectMember, int port,
      boolean retryIfConnectFails) {
    Socket clientSocket = null;
    // make sure we try to check on the member for the contracted memberTimeout period
    // in case a timed socket.connect() returns immediately. Use milliseconds to be in
    // sync with the socket timeout parameter unit of measure
    long giveupTime = System.currentTimeMillis() + services.getConfig().getMemberTimeout();
    boolean passed = false;
    int iteration = 0;
    do {
      iteration++;
      if (iteration > 1) {
        try {
          Thread.sleep(100);
        } catch (InterruptedException e) {
          Thread.currentThread().interrupt();
          return false;
        }
      }
      try {
        logger.debug("Checking member {} with TCP socket connection {}:{}.", suspectMember,
            suspectMember.getInetAddress(), port);
        clientSocket =
            socketCreator.forAdvancedUse()
                .connect(new HostAndPort(suspectMember.getHostName(), port), (int) memberTimeout,
                    new ConnectTimeoutTask(services.getTimer(), memberTimeout), false, -1, false);
        clientSocket.setTcpNoDelay(true);
        passed = doTCPCheckMember(suspectMember, clientSocket);
      } catch (IOException e) {
        // this is expected if it is a connection-timeout or other failure
        // to connect
      } catch (IllegalStateException e) {
        // This is expected if ConnectTimeoutTask is scheduled on a timer that was already
        // cancelled.
        // Once a timer has been terminated, no more tasks may be scheduled on it.
        if (!isStopping) {
          logger.trace("Unexpected exception", e);
        }
      } finally {
        try {
          if (clientSocket != null) {
            clientSocket.setSoLinger(true, 0); // abort the connection
            clientSocket.close();
          }
        } catch (IOException e) {
          // expected
        }
      }
    } while (retryIfConnectFails && !passed && !isShutdown()
        && System.currentTimeMillis() < giveupTime);
    return passed;
  }

  // Package protected for testing purposes
  boolean doTCPCheckMember(ID suspectMember, Socket clientSocket) {
    try {
      if (clientSocket.isConnected()) {
        clientSocket.setSoTimeout((int) services.getConfig().getMemberTimeout());
        InputStream in = clientSocket.getInputStream();
        DataOutputStream out = new DataOutputStream(clientSocket.getOutputStream());
        ID gmbr = suspectMember;
        writeMemberToStream(gmbr, out);
        stats.incFinalCheckRequestsSent();
        stats.incTcpFinalCheckRequestsSent();
        logger.debug("Connected to suspect member - reading response");
        int b = in.read();
        if (logger.isDebugEnabled()) {
          logger.debug("Received {}",
              (b == OK ? "OK" : (b == ERROR ? "ERROR" : "unknown response: " + b)));
        }
        if (b >= 0) {
          stats.incFinalCheckResponsesReceived();
          stats.incTcpFinalCheckResponsesReceived();
        }
        if (b == OK) {
          TimeStamp ts = memberTimeStamps.get(suspectMember);
          if (ts != null) {
            ts.setTime(System.currentTimeMillis());
          }
          return true;
        } else {
          // received ERROR
          return false;
        }
      } else {// cannot establish TCP connection with suspect member
        return false;
      }
    } catch (SocketTimeoutException e) {
      logger.debug("Availability check TCP/IP connection timed out for suspect member {}",
          suspectMember);
      return false;
    } catch (IOException e) {
      logger.trace("Unexpected exception", e);
    }
    return false;
  }

  void writeMemberToStream(ID gmbr, DataOutputStream out) throws IOException {
    out.writeShort(KnownVersion.getCurrentVersion().ordinal());
    out.writeInt(gmbr.getVmViewId());
    out.writeLong(gmbr.getUuidLeastSignificantBits());
    out.writeLong(gmbr.getUuidMostSignificantBits());
    out.flush();
  }

  @Override
  public void suspect(ID mbr, String reason) {
    initiateSuspicion(mbr, reason);
  }

  @Override
  public boolean checkIfAvailable(ID mbr, String reason,
      boolean initiateRemoval) {
    return checkIfAvailable(mbr, reason, initiateRemoval, true);
  }

  @Override
  public boolean checkIfAvailable(ID mbr, String reason,
      boolean initiateRemoval, boolean assumeMembersInFinalCheckAreAvailable) {
    if (assumeMembersInFinalCheckAreAvailable && membersInFinalCheck.contains(mbr)) {
      return true; // status unknown for now but someone is checking
    }
    return inlineCheckIfAvailable(localAddress, currentView, initiateRemoval,
        mbr, reason);
  }

  @Override
  public void start() throws MemberStartupException {
    scheduler = LoggingExecutors.newScheduledThreadPool(1, "Geode Failure Detection Scheduler");
    checkExecutor = LoggingExecutors.newCachedThreadPool("Geode Failure Detection thread ", true);
    Monitor m = new Monitor(memberTimeout);
    monitorInterval = memberTimeout / LOGICAL_INTERVAL;
    monitorFuture =
        scheduler.scheduleAtFixedRate(m, monitorInterval, monitorInterval, TimeUnit.MILLISECONDS);
    serverSocketExecutor =
        LoggingExecutors.newCachedThreadPool("Geode Failure Detection Server thread ", true);
  }

  ServerSocket createServerSocket(InetAddress socketAddress, int[] portRange) throws IOException {
    ServerSocket newSocket = socketCreator.forAdvancedUse()
        .createServerSocketUsingPortRange(socketAddress, 50/* backlog */, true/* isBindAddress */,
            false/* useNIO */, 65536/* tcpBufferSize */, portRange, false);
    socketPort = newSocket.getLocalPort();
    return newSocket;
  }

  /**
   * start the thread that listens for tcp/ip connections and responds to connection attempts
   */
  private void startTcpServer(ServerSocket ssocket) {
    // allocate a socket here so there are no race conditions between knowing the FD
    // socket port and joining the system

    serverSocketExecutor.execute(() -> {
      logger.info("Started failure detection server thread on {}:{}.", ssocket.getInetAddress(),
          socketPort);
      Socket socket = null;
      try {
        while (!services.getCancelCriterion().isCancelInProgress()
            && !isStopping) {
          try {
            socket = ssocket.accept();
            if (playingDead) {
              continue;
            }
            serverSocketExecutor.execute(new ClientSocketHandler(socket));
          } catch (RejectedExecutionException e) {
            // this can happen during shutdown

          } catch (IOException e) {
            if (!isStopping) {
              logger.trace("Unexpected exception", e);
            }
            try {
              if (socket != null) {
                socket.close();
              }
            } catch (IOException ioe) {
              logger.trace("Unexpected exception", ioe);
            }
          }
        }
        logger.info("GMSHealthMonitor server thread exiting");
      } finally {
        // close the server socket
        if (!ssocket.isClosed()) {
          try {
            ssocket.close();
          } catch (IOException e) {
            logger.debug("Unexpected exception", e);
          }
        }
      }
    });
  }

  /**
   * start the thread that periodically sends a message to processes that might be watching this
   * process
   */
  private void startHeartbeatThread() {
    checkExecutor.execute(new Heart());
  }

  @Override
  public synchronized void installView(GMSMembershipView<ID> newView) {
    synchronized (suspectRequestsInView) {
      suspectRequestsInView.clear();
    }
    for (Iterator<ID> it = memberTimeStamps.keySet().iterator(); it
        .hasNext();) {
      if (!newView.contains(it.next())) {
        it.remove();
      }
    }
    for (Iterator<ID> it = suspectedMemberIds.keySet().iterator(); it
        .hasNext();) {
      if (!newView.contains(it.next())) {
        it.remove();
      }
    }
    currentView = newView;
    setNextNeighbor(newView, null);
  }

  /**
   * this method is primarily for tests. The current view should be pulled from JoinLeave or the
   * Membership (which includes surprise members)
   */
  public synchronized GMSMembershipView<ID> getView() {
    return currentView;
  }

  /***
   * This method sets next neighbour which it needs to watch in current view.
   *
   * if nextTo == null then it watches member next to it.
   *
   * It becomes null when we suspect current neighbour, during that time it watches member next to
   * suspect member.
   */
  protected synchronized void setNextNeighbor(GMSMembershipView<ID> newView,
      ID nextTo) {
    if (newView == null) {
      return;
    }
    if (nextTo == null) {
      nextTo = localAddress;
    }

    List<ID> allMembers = newView.getMembers();

    if (allMembers.size() > 1 && suspectedMemberIds.size() >= allMembers.size() - 1) {
      boolean nonSuspectFound = false;
      for (ID member : allMembers) {
        if (member.equals(localAddress)) {
          continue;
        }
        if (!suspectedMemberIds.containsKey(member)) {
          nonSuspectFound = true;
          break;
        }
      }
      if (!nonSuspectFound) {
        logger.info("All other members are suspect at this point");
        nextNeighbor = null;
        return;
      }
    }

    int index = allMembers.indexOf(nextTo);
    if (index != -1) {
      int nextNeighborIndex = (index + 1) % allMembers.size();
      ID newNeighbor = allMembers.get(nextNeighborIndex);
      if (suspectedMemberIds.containsKey(newNeighbor)) {
        setNextNeighbor(newView, newNeighbor);
        return;
      }
      ID oldNeighbor = nextNeighbor;
      if (oldNeighbor != newNeighbor) {
        logger.debug("Failure detection is now watching " + newNeighbor);
        nextNeighbor = newNeighbor;
      }
    }

    if (nextNeighbor != null && nextNeighbor.equals(localAddress)) {
      if (logger.isDebugEnabled()) {
        logger.debug("Health monitor is unable to find a neighbor to watch.  "
            + "Current suspects are {}", suspectedMemberIds.keySet());
      }
      nextNeighbor = null;
    }

  }

  /** test method */
  public ID getNextNeighbor() {
    return nextNeighbor;
  }

  @Override
  public void init(Services<ID> s) throws MembershipConfigurationException {
    isStopping = false;
    services = s;
    memberTimeout = s.getConfig().getMemberTimeout();
    stats = services.getStatistics();

    services.getMessenger().addHandler(HeartbeatRequestMessage.class,
        this::processMessage);
    services.getMessenger().addHandler(HeartbeatMessage.class,
        this::processMessage);
    services.getMessenger().addHandler(SuspectMembersMessage.class,
        this::processMessage);
    services.getMessenger().addHandler(FinalCheckPassedMessage.class,
        this::processMessage);
  }

  @Override
  public void started() throws MemberStartupException {
    setLocalAddress(services.getMessenger().getMemberID());
    try {
      serverSocket = createServerSocket(localAddress.getInetAddress(),
          services.getConfig().getMembershipPortRange());
    } catch (IOException e) {
      throw new MemberStartupException("Problem creating HealthMonitor socket", e);
    }
    startTcpServer(serverSocket);
    startHeartbeatThread();
  }

  @Override
  public void stop() {
    stopServices();
  }

  private void stopServices() {
    logger.debug("Stopping HealthMonitor");
    isStopping = true;
    if (monitorFuture != null) {
      monitorFuture.cancel(true);
    }
    if (scheduler != null) {
      scheduler.shutdown();
    }

    Collection<Response> val = requestIdVsResponse.values();
    for (Response r : val) {
      synchronized (r) {
        r.notify();
      }
    }

    if (checkExecutor != null) {
      checkExecutor.shutdown();
    }

    stopServer();
  }

  void stopServer() {
    if (serverSocketExecutor != null) {
      if (serverSocket != null && !serverSocket.isClosed()) {
        try {
          serverSocket.close();
        } catch (IOException e) {
          logger.trace("Unexpected exception", e);
        }
      }
      serverSocketExecutor.shutdownNow();
      try {
        serverSocketExecutor.awaitTermination(2000, TimeUnit.MILLISECONDS);
      } catch (InterruptedException e) {
        Thread.currentThread().interrupt();
      }
    }
  }

  /***
   * test method
   */
  public boolean isShutdown() {
    return scheduler.isShutdown() && checkExecutor.isShutdown()
        && serverSocketExecutor.isShutdown();
  }

  /**
   * Test method - check to see if a member is under suspicion
   */
  public boolean isSuspectMember(ID m) {
    return suspectedMemberIds.containsKey(m);
  }

  @Override
  public void stopped() {

  }

  @Override
  public void memberSuspected(ID initiator,
      ID suspect, String reason) {
    synchronized (suspectRequestsInView) {
      suspectedMemberIds.put(suspect, currentView);
      Collection<SuspectRequest<ID>> requests = suspectRequestsInView.get(currentView);
      boolean found = false;
      if (requests == null) {
