/*
 * 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.ratis.grpc.server;

import org.apache.ratis.conf.RaftProperties;
import org.apache.ratis.grpc.GrpcConfigKeys;
import org.apache.ratis.grpc.GrpcUtil;
import org.apache.ratis.grpc.metrics.GrpcServerMetrics;
import org.apache.ratis.metrics.Timekeeper;
import org.apache.ratis.proto.RaftProtos.InstallSnapshotResult;
import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.retry.MultipleLinearRandomRetry;
import org.apache.ratis.retry.RetryPolicy;
import org.apache.ratis.server.RaftServer;
import org.apache.ratis.server.RaftServerConfigKeys;
import org.apache.ratis.server.leader.FollowerInfo;
import org.apache.ratis.server.leader.LeaderState;
import org.apache.ratis.server.leader.LogAppenderBase;
import org.apache.ratis.server.protocol.TermIndex;
import org.apache.ratis.server.raftlog.RaftLog;
import org.apache.ratis.server.util.ServerStringUtils;
import org.apache.ratis.thirdparty.io.grpc.StatusRuntimeException;
import org.apache.ratis.thirdparty.io.grpc.stub.CallStreamObserver;
import org.apache.ratis.thirdparty.io.grpc.stub.StreamObserver;
import org.apache.ratis.proto.RaftProtos.AppendEntriesReplyProto;
import org.apache.ratis.proto.RaftProtos.AppendEntriesReplyProto.AppendResult;
import org.apache.ratis.proto.RaftProtos.AppendEntriesRequestProto;
import org.apache.ratis.proto.RaftProtos.InstallSnapshotReplyProto;
import org.apache.ratis.proto.RaftProtos.InstallSnapshotReplyProto.InstallSnapshotReplyBodyCase;
import org.apache.ratis.proto.RaftProtos.InstallSnapshotRequestProto;
import org.apache.ratis.proto.RaftProtos.InstallSnapshotRequestProto.InstallSnapshotRequestBodyCase;
import org.apache.ratis.statemachine.SnapshotInfo;
import org.apache.ratis.util.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InterruptedIOException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

/**
 * A new log appender implementation using grpc bi-directional stream API.
 */
public class GrpcLogAppender extends LogAppenderBase {
  public static final Logger LOG = LoggerFactory.getLogger(GrpcLogAppender.class);

  private enum BatchLogKey implements BatchLogger.Key {
    RESET_CLIENT,
    APPEND_LOG_RESPONSE_HANDLER_ON_ERROR
  }

  public static final int INSTALL_SNAPSHOT_NOTIFICATION_INDEX = 0;

  private static final Comparator<Long> CALL_ID_COMPARATOR = (left, right) -> {
    // calculate diff in order to take care the possibility of numerical overflow
    final long diff = left - right;
    return diff == 0? 0: diff > 0? 1: -1;
  };

  enum Event {
    APPEND_ENTRIES_REPLY,
    APPEND_ENTRIES_INCONSISTENCY_REPLY,
    SNAPSHOT_REPLY,
    COMPLETE,
    TIMEOUT,
    ERROR;

    boolean updateFirstReplyReceived(boolean firstReplyReceived) {
      switch (this) {
        case APPEND_ENTRIES_REPLY:
        case APPEND_ENTRIES_INCONSISTENCY_REPLY:
        case SNAPSHOT_REPLY:
        case COMPLETE:
          return true;
        case ERROR:
          return false;
        case TIMEOUT:
          return firstReplyReceived;
        default:
          throw new IllegalStateException("Unexpected event: " + this);
      }
    }

    boolean isError() {
      switch (this) {
        case APPEND_ENTRIES_INCONSISTENCY_REPLY:
        case TIMEOUT:
        case ERROR:
          return true;
        case APPEND_ENTRIES_REPLY:
        case SNAPSHOT_REPLY:
        case COMPLETE:
          return false;
        default:
          throw new IllegalStateException("Unexpected event: " + this);
      }
    }
  }

  static class ReplyState {
    private boolean firstReplyReceived = false;
    private int errorCount = 0;

    synchronized boolean isFirstReplyReceived() {
      return firstReplyReceived;
    }

    synchronized int getErrorCount() {
      return errorCount;
    }

    int process(AppendResult result) {
      return process(result == AppendResult.INCONSISTENCY? Event.APPEND_ENTRIES_INCONSISTENCY_REPLY
          : Event.APPEND_ENTRIES_REPLY);
    }

    synchronized int process(Event event) {
      firstReplyReceived = event.updateFirstReplyReceived(firstReplyReceived);
      if (event.isError()) {
        errorCount++;
      } else {
        errorCount = 0;
      }
      return errorCount;
    }
  }

  private final AtomicLong callId = new AtomicLong();

  private final RequestMap pendingRequests = new RequestMap();
  private final int maxPendingRequestsNum;
  private final boolean installSnapshotEnabled;

  private final TimeDuration requestTimeoutDuration;
  private final TimeDuration installSnapshotStreamTimeout;
  private final TimeDuration logMessageBatchDuration;
  private final int maxOutstandingInstallSnapshots;
  private final TimeoutExecutor scheduler = TimeoutExecutor.getInstance();
  @SuppressWarnings({"squid:S3077"}) // Suppress volatile for generic type
  private volatile StreamObservers appendLogRequestObserver;
  private final boolean useSeparateHBChannel;

  private final GrpcServerMetrics grpcServerMetrics;

  private final AutoCloseableReadWriteLock lock;
  private final StackTraceElement caller;
  private final RetryPolicy errorRetryWaitPolicy;
  private final ReplyState replyState = new ReplyState();

  public GrpcLogAppender(RaftServer.Division server, LeaderState leaderState, FollowerInfo f) {
    super(server, leaderState, f);

    Preconditions.assertNotNull(getServerRpc(), "getServerRpc()");

    final RaftProperties properties = server.getRaftServer().getProperties();
    this.maxPendingRequestsNum = GrpcConfigKeys.Server.leaderOutstandingAppendsMax(properties);
    this.requestTimeoutDuration = RaftServerConfigKeys.Rpc.requestTimeout(properties);
    this.maxOutstandingInstallSnapshots = GrpcConfigKeys.Server.installSnapshotRequestElementLimit(properties);
    this.installSnapshotStreamTimeout = GrpcConfigKeys.Server.installSnapshotRequestTimeout(properties)
        .multiply(maxOutstandingInstallSnapshots);
    this.logMessageBatchDuration = GrpcConfigKeys.Server.logMessageBatchDuration(properties);
    this.installSnapshotEnabled = RaftServerConfigKeys.Log.Appender.installSnapshotEnabled(properties);
    this.useSeparateHBChannel = GrpcConfigKeys.Server.heartbeatChannel(properties);

    grpcServerMetrics = new GrpcServerMetrics(server.getMemberId().toString());
    grpcServerMetrics.addPendingRequestsCount(getFollowerId().toString(), pendingRequests::logRequestsSize);

    lock = new AutoCloseableReadWriteLock(this);
    caller = LOG.isTraceEnabled()? JavaUtils.getCallerStackTraceElement(): null;
    errorRetryWaitPolicy = MultipleLinearRandomRetry.parseCommaSeparated(
        RaftServerConfigKeys.Log.Appender.retryPolicy(properties));
  }

  @Override
  public GrpcService getServerRpc() {
    return (GrpcService)super.getServerRpc();
  }

  private GrpcServerProtocolClient getClient() throws IOException {
    return getServerRpc().getProxies().getProxy(getFollowerId());
  }

  private void resetClient(AppendEntriesRequest request, Event event) {
    try (AutoCloseableLock writeLock = lock.writeLock(caller, LOG::trace)) {
      getClient().resetConnectBackoff();
      if (appendLogRequestObserver != null) {
        appendLogRequestObserver.stop();
        appendLogRequestObserver = null;
      }
      final int errorCount = replyState.process(event);
      // clear the pending requests queue and reset the next index of follower
      pendingRequests.clear();
      final FollowerInfo f = getFollower();
      final long nextIndex = 1 + Optional.ofNullable(request)
          .map(AppendEntriesRequest::getPreviousLog)
          .map(TermIndex::getIndex)
          .orElseGet(f::getMatchIndex);
      if (event.isError() && request == null) {
        final long followerNextIndex = f.getNextIndex();
        BatchLogger.warn(BatchLogKey.RESET_CLIENT, f.getId() + "-" + followerNextIndex, suffix ->
            LOG.warn("{}: Follower failed (request=null, errorCount={}); keep nextIndex ({}) unchanged and retry.{}",
                this, errorCount, followerNextIndex, suffix), logMessageBatchDuration);
        return;
      }
      if (request != null && request.isHeartbeat()) {
        return;
      }
      getFollower().computeNextIndex(getNextIndexForError(nextIndex));
    } catch (IOException ie) {
      LOG.warn(this + ": Failed to getClient for " + getFollowerId(), ie);
    }
  }

  private boolean isFollowerCommitBehindLastCommitIndex() {
    return getRaftLog().getLastCommittedIndex() > getFollower().getCommitIndex();
  }

  private boolean installSnapshot() {
    if (installSnapshotEnabled) {
      final SnapshotInfo snapshot = shouldInstallSnapshot();
      if (snapshot != null) {
        installSnapshot(snapshot);
        return true;
      }
    } else {
      // check installSnapshotNotification
      final TermIndex firstAvailable = shouldNotifyToInstallSnapshot();
      if (firstAvailable != null) {
        notifyInstallSnapshot(firstAvailable);
        return true;
      }
    }
    return false;
  }

  @Override
  public void run() throws IOException {
    for(; isRunning(); mayWait()) {
      //HB period is expired OR we have messages OR follower is behind with commit index
      if (shouldSendAppendEntries() || isFollowerCommitBehindLastCommitIndex()) {
        final boolean installingSnapshot = installSnapshot();
        appendLog(installingSnapshot || haveTooManyPendingRequests());
      }
      getLeaderState().checkHealth(getFollower());
    }

    Optional.ofNullable(appendLogRequestObserver).ifPresent(StreamObservers::onCompleted);
  }

  public long getWaitTimeMs() {
    if (haveTooManyPendingRequests()) {
      return getHeartbeatWaitTimeMs(); // Should wait for a short time
    } else if (shouldSendAppendEntries() && !isSlowFollower()) {
      // For normal nodes, new entries should be sent ASAP
      // however for slow followers (especially when the follower is down),
      // keep sending without any wait time only ends up in high CPU load
      return TimeDuration.max(getRemainingWaitTime(), TimeDuration.ZERO).toLong(TimeUnit.MILLISECONDS);
    }
    return getHeartbeatWaitTimeMs();
  }

  private boolean isSlowFollower() {
    final TimeDuration elapsedTime = getFollower().getLastRpcResponseTime().elapsedTime();
    return elapsedTime.compareTo(getServer().properties().rpcSlownessTimeout()) > 0;
  }

  private void mayWait() {
    // use lastSend time instead of lastResponse time
    try {
      getEventAwaitForSignal().await(getWaitTimeMs() + errorWaitTimeMs(),
          TimeUnit.MILLISECONDS);
    } catch (InterruptedException ie) {
      LOG.warn(this + ": Wait interrupted by " + ie);
      Thread.currentThread().interrupt();
    }
  }

  private long errorWaitTimeMs() {
    return errorRetryWaitPolicy.handleAttemptFailure(replyState::getErrorCount)
        .getSleepTime().toLong(TimeUnit.MILLISECONDS);
  }

  @Override
  public CompletableFuture<LifeCycle.State> stopAsync() {
    grpcServerMetrics.unregister();
    return super.stopAsync();
  }

  @Override
  public boolean shouldSendAppendEntries() {
    return appendLogRequestObserver == null || super.shouldSendAppendEntries();
  }

  @Override
  public boolean hasPendingDataRequests() {
    return pendingRequests.logRequestsSize() > 0;
  }

  /** @return true iff either (1) queue is full, or (2) queue is non-empty and not received first response. */
  private boolean haveTooManyPendingRequests() {
    final int size = pendingRequests.logRequestsSize();
    if (size == 0) {
      return false;
    } else if (size >= maxPendingRequestsNum) {
      return true;
    } else {
      // queue is non-empty and non-full
      return !replyState.isFirstReplyReceived();
    }
  }

  static class StreamObservers {
    private final CallStreamObserver<AppendEntriesRequestProto> appendLog;
    private final CallStreamObserver<AppendEntriesRequestProto> heartbeat;
    private final TimeDuration waitForReady;
    private volatile boolean running = true;

    StreamObservers(GrpcServerProtocolClient client, AppendLogResponseHandler handler, boolean separateHeartbeat,
        TimeDuration waitTimeMin) {
      this.appendLog = client.appendEntries(handler, false);
      this.heartbeat = separateHeartbeat? client.appendEntries(handler, true): null;
      this.waitForReady = waitTimeMin.isPositive()? waitTimeMin: TimeDuration.ONE_MILLISECOND;
    }

    void onNext(AppendEntriesRequestProto proto)
        throws InterruptedIOException {
      CallStreamObserver<AppendEntriesRequestProto> stream;
      boolean isHeartBeat = heartbeat != null && proto.getEntriesCount() == 0;
      if (isHeartBeat) {
        stream = heartbeat;
      } else {
        stream = appendLog;
      }
      // stall for stream to be ready.
      while (!stream.isReady() && running) {
        sleep(waitForReady, isHeartBeat);
      }
      stream.onNext(proto);
    }

    void stop() {
      running = false;
    }

    void onCompleted() {
      appendLog.onCompleted();
      Optional.ofNullable(heartbeat).ifPresent(StreamObserver::onCompleted);
    }
  }

  @Override
  public long getCallId() {
    return callId.get();
  }

  @Override
  public Comparator<Long> getCallIdComparator() {
    return CALL_ID_COMPARATOR;
  }

  private void appendLog(boolean heartbeat) throws IOException {
    ReferenceCountedObject<AppendEntriesRequestProto> pending = null;
    final AppendEntriesRequest request;
    try (AutoCloseableLock writeLock = lock.writeLock(caller, LOG::trace)) {
      // Prepare and send the append request.
      // Note changes on follower's nextIndex and ops on pendingRequests should always be done under the write-lock
      pending = nextAppendEntriesRequest(callId.getAndIncrement(), heartbeat);
      if (pending == null) {
        return;
      }
      request = new AppendEntriesRequest(pending.get(), getFollowerId(), grpcServerMetrics);
      pendingRequests.put(request);
      increaseNextIndex(pending.get());
      if (appendLogRequestObserver == null) {
        appendLogRequestObserver = new StreamObservers(
            getClient(), new AppendLogResponseHandler(), useSeparateHBChannel, getWaitTimeMin());
      }
    } catch(Exception e) {
      if (pending != null) {
        pending.release();
      }
      throw e;
    }

    try {
      final TimeDuration remaining = getRemainingWaitTime();
      if (remaining.isPositive()) {
        sleep(remaining, heartbeat);
      }
      if (isRunning()) {
        sendRequest(request, pending.get());
      }
    } finally {
      pending.release();
    }
  }

  private static void sleep(TimeDuration waitTime, boolean heartbeat)
      throws InterruptedIOException {
    try {
      waitTime.sleep();
    } catch (InterruptedException e) {
      Thread.currentThread().interrupt();
      throw IOUtils.toInterruptedIOException(
          "Interrupted appendLog, heartbeat? " + heartbeat, e);
    }
  }

  private void sendRequest(AppendEntriesRequest request,
      AppendEntriesRequestProto proto) throws InterruptedIOException {
    CodeInjectionForTesting.execute(GrpcService.GRPC_SEND_SERVER_REQUEST,
        getServer().getId(), null, proto);
    resetHeartbeatTrigger();

    StreamObservers observers = appendLogRequestObserver;
    if (observers != null) {
      request.startRequestTimer();
      observers.onNext(proto);
      getFollower().updateLastRpcSendTime(request.isHeartbeat());
      scheduler.onTimeout(requestTimeoutDuration,
          () -> timeoutAppendRequest(request.getCallId(), request.isHeartbeat()),
          LOG, () -> "Timeout check failed for append entry request: " + request);
    }
  }

  private void timeoutAppendRequest(long cid, boolean heartbeat) {
    final AppendEntriesRequest pending = pendingRequests.handleTimeout(cid, heartbeat);
    if (pending != null) {
      final int errorCount = replyState.process(Event.TIMEOUT);
      LOG.warn("{}: Timed out {}appendEntries, errorCount={}, request={}",
          this, heartbeat ? "HEARTBEAT " : "", errorCount, pending);
      grpcServerMetrics.onRequestTimeout(getFollowerId().toString(), heartbeat);
      pending.stopRequestTimer();
    }
  }

  private void increaseNextIndex(AppendEntriesRequestProto request) {
    final int count = request.getEntriesCount();
    if (count > 0) {
      getFollower().increaseNextIndex(request.getEntries(count - 1).getIndex() + 1);
    }
  }

  private void increaseNextIndex(final long installedSnapshotIndex, Object reason) {
    final long newNextIndex = installedSnapshotIndex + 1;
    LOG.info("{}: updateNextIndex {} for {}", this, newNextIndex, reason);
    getFollower().updateNextIndex(newNextIndex);
  }

  /**
   * StreamObserver for handling responses from the follower
   */
  private class AppendLogResponseHandler implements StreamObserver<AppendEntriesReplyProto> {
    private final String name = getFollower().getName() + "-" + JavaUtils.getClassSimpleName(getClass());

    /**
     * After receiving a appendEntries reply, do the following:
     * 1. If the reply is success, update the follower's match index and submit
     *    an event to leaderState
     * 2. If the reply is NOT_LEADER, step down
     * 3. If the reply is INCONSISTENCY, increase/ decrease the follower's next
     *    index based on the response
     */
    @Override
    public void onNext(AppendEntriesReplyProto reply) {
      AppendEntriesRequest request = pendingRequests.remove(reply);
      if (request != null) {
        request.stopRequestTimer(); // Update completion time
        getFollower().updateLastRespondedAppendEntriesSendTime(request.getSendTime());
      }
      getFollower().updateLastRpcResponseTime();

      if (LOG.isDebugEnabled()) {
        LOG.debug("{}: received {} reply {}, request={}",
            this, replyState.isFirstReplyReceived()? "a": "the first",
            ServerStringUtils.toAppendEntriesReplyString(reply), request);
      }

      try {
        onNextImpl(request, reply);
      } catch(Exception t) {
        LOG.error("Failed onNext request=" + request
            + ", reply=" + ServerStringUtils.toAppendEntriesReplyString(reply), t);
      }
    }

    private void onNextImpl(AppendEntriesRequest request, AppendEntriesReplyProto reply) {
      final int errorCount = replyState.process(reply.getResult());

      switch (reply.getResult()) {
        case SUCCESS:
          grpcServerMetrics.onRequestSuccess(getFollowerId().toString(), reply.getIsHearbeat());
          getLeaderState().onFollowerCommitIndex(getFollower(), reply.getFollowerCommit());
          if (getFollower().updateMatchIndex(reply.getMatchIndex())) {
            getFollower().updateNextIndex(reply.getMatchIndex() + 1);
            getLeaderState().onFollowerSuccessAppendEntries(getFollower());
          }
          break;
        case NOT_LEADER:
          grpcServerMetrics.onRequestNotLeader(getFollowerId().toString());
          LOG.warn("{}: received {} reply with term {}", this, reply.getResult(), reply.getTerm());
          if (onFollowerTerm(reply.getTerm())) {
            return;
          }
          break;
        case INCONSISTENCY:
          grpcServerMetrics.onRequestInconsistency(getFollowerId().toString());
          LOG.warn("{}: received {} reply with nextIndex {}, errorCount={}, request={}",
              this, reply.getResult(), reply.getNextIndex(), errorCount, request);
          final long requestFirstIndex = request != null? request.getFirstIndex(): RaftLog.INVALID_LOG_INDEX;
          updateNextIndex(getNextIndexForInconsistency(requestFirstIndex, reply.getNextIndex()));
          break;
        default:
          throw new IllegalStateException("Unexpected reply result: " + reply.getResult());
      }
      getLeaderState().onAppendEntriesReply(GrpcLogAppender.this, reply);
      notifyLogAppender();
    }

    /**
     * for now we simply retry the first pending request
     */
    @Override
    public void onError(Throwable t) {
      if (!isRunning()) {
        LOG.info("{} is already stopped", GrpcLogAppender.this);
        return;
      }
      BatchLogger.warn(BatchLogKey.APPEND_LOG_RESPONSE_HANDLER_ON_ERROR, AppendLogResponseHandler.this.name,
          suffix -> GrpcUtil.warn(LOG, () -> this + ": Failed appendEntries" + suffix, t),
          logMessageBatchDuration, t instanceof StatusRuntimeException);
      grpcServerMetrics.onRequestRetry(); // Update try counter
      AppendEntriesRequest request = pendingRequests.remove(GrpcUtil.getCallId(t), GrpcUtil.isHeartbeat(t));
      resetClient(request, Event.ERROR);
    }

    @Override
    public void onCompleted() {
      LOG.info("{}: follower responses appendEntries COMPLETED", this);
      resetClient(null, Event.COMPLETE);
    }

    @Override
    public String toString() {
      return name;
    }
  }

  private void updateNextIndex(long replyNextIndex) {
    try (AutoCloseableLock writeLock = lock.writeLock(caller, LOG::trace)) {
      pendingRequests.clear();
      getFollower().setNextIndex(replyNextIndex);
    }
  }

  private class InstallSnapshotResponseHandler implements StreamObserver<InstallSnapshotReplyProto> {
    private final String name = getFollower().getName() + "-" + JavaUtils.getClassSimpleName(getClass());
    private final Queue<Integer> pending;
    private final AtomicBoolean done = new AtomicBoolean(false);
    private final boolean isNotificationOnly;

    InstallSnapshotResponseHandler() {
      this(false);
    }

    InstallSnapshotResponseHandler(boolean notifyOnly) {
      pending = new LinkedList<>();
      this.isNotificationOnly = notifyOnly;
    }

    void addPending(InstallSnapshotRequestProto request) {
      try (AutoCloseableLock writeLock = lock.writeLock(caller, LOG::trace)) {
        final int index;
        if (isNotificationOnly) {
          Preconditions.assertSame(InstallSnapshotRequestBodyCase.NOTIFICATION,
                  request.getInstallSnapshotRequestBodyCase(), "request case");
          index = INSTALL_SNAPSHOT_NOTIFICATION_INDEX;
        } else {
          Preconditions.assertSame(InstallSnapshotRequestBodyCase.SNAPSHOTCHUNK,
                  request.getInstallSnapshotRequestBodyCase(), "request case");
          index = request.getSnapshotChunk().getRequestIndex();
        }
        if (index == 0) {
          Preconditions.assertTrue(pending.isEmpty(), "pending queue is non-empty before offer for index 0");
        }
        pending.offer(index);
      }
    }

    void removePending(InstallSnapshotReplyProto reply) {
      try (AutoCloseableLock writeLock = lock.writeLock(caller, LOG::trace)) {
        final int index = Objects.requireNonNull(pending.poll(), "index == null");
        if (isNotificationOnly) {
          Preconditions.assertSame(InstallSnapshotReplyBodyCase.SNAPSHOTINDEX,
                  reply.getInstallSnapshotReplyBodyCase(), "reply case");
          Preconditions.assertSame(INSTALL_SNAPSHOT_NOTIFICATION_INDEX, (int) index, "poll index");
        } else {
          Preconditions.assertSame(InstallSnapshotReplyBodyCase.REQUESTINDEX,
                  reply.getInstallSnapshotReplyBodyCase(), "reply case");
          Preconditions.assertSame(reply.getRequestIndex(), (int) index, "poll index");
        }
      }
    }

    //compare follower's latest installed snapshot index with leader's start index
    void onFollowerCatchup(long followerSnapshotIndex) {
      final long leaderStartIndex = getRaftLog().getStartIndex();
      final long followerNextIndex = followerSnapshotIndex + 1;
      if (followerNextIndex >= leaderStartIndex) {
        LOG.info("{}: Follower can catch up leader after install the snapshot, as leader's start index is {}",
            this, followerNextIndex);
        notifyInstallSnapshotFinished(InstallSnapshotResult.SUCCESS, followerSnapshotIndex);
      }
    }

    void notifyInstallSnapshotFinished(InstallSnapshotResult result, long snapshotIndex) {
      getServer().getStateMachine().event().notifySnapshotInstalled(result, snapshotIndex, getFollower().getPeer());
    }

    boolean isDone() {
      return done.get();
    }

    void close() {
      done.set(true);
      notifyLogAppender();
    }

    boolean hasAllResponse() {
      try (AutoCloseableLock readLock = lock.readLock(caller, LOG::trace)) {
        return pending.isEmpty();
      }
    }

    @Override
    public void onNext(InstallSnapshotReplyProto reply) {
      if (LOG.isInfoEnabled()) {
        LOG.info("{}: received {} reply {}", this, replyState.isFirstReplyReceived()? "a" : "the first",
            ServerStringUtils.toInstallSnapshotReplyString(reply));
      }

      // update the last rpc time
      getFollower().updateLastRpcResponseTime();
      replyState.process(Event.SNAPSHOT_REPLY);

      final long followerSnapshotIndex;
      switch (reply.getResult()) {
        case SUCCESS:
          LOG.info("{}: Completed InstallSnapshot. Reply: {}", this, reply);
          getFollower().setAttemptedToInstallSnapshot();
          removePending(reply);
          break;
        case IN_PROGRESS:
          LOG.info("{}: InstallSnapshot in progress.", this);
          removePending(reply);
          break;
        case ALREADY_INSTALLED:
          followerSnapshotIndex = reply.getSnapshotIndex();
          LOG.info("{}: Follower snapshot is already at index {}.", this, followerSnapshotIndex);
          getFollower().setSnapshotIndex(followerSnapshotIndex);
          getFollower().setAttemptedToInstallSnapshot();
          getLeaderState().onFollowerCommitIndex(getFollower(), followerSnapshotIndex);
          increaseNextIndex(followerSnapshotIndex, reply.getResult());
          removePending(reply);
          break;
        case NOT_LEADER:
          onFollowerTerm(reply.getTerm());
          break;
        case CONF_MISMATCH:
          LOG.error("{}: Configuration Mismatch ({}): Leader {} has it set to {} but follower {} has it set to {}",
              this, RaftServerConfigKeys.Log.Appender.INSTALL_SNAPSHOT_ENABLED_KEY,
              getServer().getId(), installSnapshotEnabled, getFollowerId(), !installSnapshotEnabled);
          break;
        case SNAPSHOT_INSTALLED:
          followerSnapshotIndex = reply.getSnapshotIndex();
          LOG.info("{}: Follower installed snapshot at index {}", this, followerSnapshotIndex);
          getFollower().setSnapshotIndex(followerSnapshotIndex);
          getFollower().setAttemptedToInstallSnapshot();
          getLeaderState().onFollowerCommitIndex(getFollower(), followerSnapshotIndex);
          increaseNextIndex(followerSnapshotIndex, reply.getResult());
          onFollowerCatchup(followerSnapshotIndex);
          removePending(reply);
          break;
        case SNAPSHOT_UNAVAILABLE:
          LOG.info("{}: Follower could not install snapshot as it is not available.", this);
          getFollower().setAttemptedToInstallSnapshot();
          notifyInstallSnapshotFinished(InstallSnapshotResult.SNAPSHOT_UNAVAILABLE, RaftLog.INVALID_LOG_INDEX);
          removePending(reply);
          break;
        case UNRECOGNIZED:
          LOG.error("Unrecognized the reply result {}: Leader is {}, follower is {}",
              reply.getResult(), getServer().getId(), getFollowerId());
          break;
        default:
          break;
      }
    }

    @Override
    public void onError(Throwable t) {
      if (!isRunning()) {
        LOG.info("{} is stopped", GrpcLogAppender.this);
        return;
      }
      GrpcUtil.warn(LOG, () -> this + ": Failed InstallSnapshot", t);
      grpcServerMetrics.onRequestRetry(); // Update try counter
      resetClient(null, Event.ERROR);
      close();
    }

    @Override
    public void onCompleted() {
      if (!isNotificationOnly || LOG.isDebugEnabled()) {
        LOG.info("{}: follower responded installSnapshot COMPLETED", this);
      }
      replyState.process(Event.COMPLETE);
      close();
    }

    @Override
    public String toString() {
      return name;
    }
  }

  /**
   * Send installSnapshot request to Follower with a snapshot.
   * @param snapshot the snapshot to be sent to Follower
   */
  private void installSnapshot(SnapshotInfo snapshot) {
    LOG.info("{}: followerNextIndex = {} but logStartIndex = {}, send snapshot {} to follower",
        this, getFollower().getNextIndex(), getRaftLog().getStartIndex(), snapshot);

    final InstallSnapshotResponseHandler responseHandler = new InstallSnapshotResponseHandler();
    StreamObserver<InstallSnapshotRequestProto> snapshotRequestObserver = null;
    final String requestId = UUID.randomUUID().toString();
    try {
      snapshotRequestObserver = getClient().installSnapshot(
          getFollower().getName() + "-installSnapshot-" + requestId,
          installSnapshotStreamTimeout, maxOutstandingInstallSnapshots, responseHandler);
      for (InstallSnapshotRequestProto request : newInstallSnapshotRequests(requestId, snapshot)) {
        if (isRunning()) {
          snapshotRequestObserver.onNext(request);
          getFollower().updateLastRpcSendTime(false);
          responseHandler.addPending(request);
        } else {
          break;
        }
      }
      snapshotRequestObserver.onCompleted();
      grpcServerMetrics.onInstallSnapshot();
    } catch (Exception e) {
      LOG.warn(this + ": failed to installSnapshot " + snapshot, e);
      if (snapshotRequestObserver != null) {
        snapshotRequestObserver.onError(e);
      }
      return;
    }

    while (isRunning() && !responseHandler.isDone()) {
      try {
        getEventAwaitForSignal().await();
      } catch (InterruptedException ignored) {
        Thread.currentThread().interrupt();
      }
    }

    if (responseHandler.hasAllResponse()) {
      getFollower().setSnapshotIndex(snapshot.getTermIndex().getIndex());
      LOG.info("{}: installed snapshot {} successfully", this, snapshot);
    }
  }

  /**
   * Send an installSnapshot notification request to the Follower.
   * @param firstAvailable the first available log's index on the Leader
   */
  private void notifyInstallSnapshot(TermIndex firstAvailable) {
    LOG.info("{}: notifyInstallSnapshot with firstAvailable={}, followerNextIndex={}",
        this, firstAvailable, getFollower().getNextIndex());

    final InstallSnapshotResponseHandler responseHandler = new InstallSnapshotResponseHandler(true);
    StreamObserver<InstallSnapshotRequestProto> snapshotRequestObserver = null;
    // prepare and enqueue the notify install snapshot request.
    final InstallSnapshotRequestProto request = newInstallSnapshotNotificationRequest(firstAvailable);
    if (LOG.isInfoEnabled()) {
      LOG.info("{}: send {}", this, ServerStringUtils.toInstallSnapshotRequestString(request));
    }
    try {
      snapshotRequestObserver = getClient().installSnapshot(getFollower().getName() + "-notifyInstallSnapshot",
          requestTimeoutDuration, 0, responseHandler);

      snapshotRequestObserver.onNext(request);
      getFollower().updateLastRpcSendTime(false);
      responseHandler.addPending(request);
      snapshotRequestObserver.onCompleted();
    } catch (Exception e) {
      GrpcUtil.warn(LOG, () -> this + ": Failed to notify follower to install snapshot.", e);
      if (snapshotRequestObserver != null) {
        snapshotRequestObserver.onError(e);
      }
      return;
    }

    while (isRunning() && !responseHandler.isDone()) {
      try {
        getEventAwaitForSignal().await();
      } catch (InterruptedException ignored) {
        Thread.currentThread().interrupt();
      }
    }
  }

  /**
   * Should the Leader notify the Follower to install the snapshot through
   * its own State Machine.
   * @return the first available log's start term index
   */
  private TermIndex shouldNotifyToInstallSnapshot() {
    final FollowerInfo follower = getFollower();
    final long leaderNextIndex = getRaftLog().getNextIndex();
    final boolean isFollowerBootstrapping = getLeaderState().isFollowerBootstrapping(follower);
    final long leaderStartIndex = getRaftLog().getStartIndex();
    final TermIndex firstAvailable = Optional.ofNullable(getRaftLog().getTermIndex(leaderStartIndex))
        .orElseGet(() -> TermIndex.valueOf(getServer().getInfo().getCurrentTerm(), leaderNextIndex));
    if (isFollowerBootstrapping && !follower.hasAttemptedToInstallSnapshot()) {
      // If the follower is bootstrapping and has not yet installed any snapshot from leader, then the follower should
      // be notified to install a snapshot. Every follower should try to install at least one snapshot during
      // bootstrapping, if available.
      LOG.debug("{}: follower is bootstrapping, notify to install snapshot to {}.", this, firstAvailable);
      return firstAvailable;
    }

    final long followerNextIndex = follower.getNextIndex();
    if (followerNextIndex >= leaderNextIndex) {
      return null;
    }

    if (followerNextIndex < leaderStartIndex) {
      // The Leader does not have the logs from the Follower's last log
      // index onwards. And install snapshot is disabled. So the Follower
      // should be notified to install the latest snapshot through its
      // State Machine.
      return firstAvailable;
    } else if (leaderStartIndex == RaftLog.INVALID_LOG_INDEX) {
      // Leader has no logs to check from, hence return next index.
      return firstAvailable;
    }

    return null;
  }

  static class AppendEntriesRequest {
    private final Timekeeper timer;
    @SuppressWarnings({"squid:S3077"}) // Suppress volatile for generic type
    private volatile Timekeeper.Context timerContext;

    private final long callId;
    private final TermIndex previousLog;
    private final int entriesCount;

    private final TermIndex firstEntry;
    private final TermIndex lastEntry;
    @SuppressWarnings({"squid:S3077"}) // Suppress volatile for generic type
    private volatile Timestamp sendTime;

    AppendEntriesRequest(AppendEntriesRequestProto proto, RaftPeerId followerId, GrpcServerMetrics grpcServerMetrics) {
      this.callId = proto.getServerRequest().getCallId();
      this.previousLog = proto.hasPreviousLog()? TermIndex.valueOf(proto.getPreviousLog()): null;
      this.entriesCount = proto.getEntriesCount();
      this.firstEntry = entriesCount > 0? TermIndex.valueOf(proto.getEntries(0)): null;
      this.lastEntry = entriesCount > 0? TermIndex.valueOf(proto.getEntries(entriesCount - 1)): null;

      this.timer = grpcServerMetrics.getGrpcLogAppenderLatencyTimer(followerId.toString(), isHeartbeat());
      grpcServerMetrics.onRequestCreate(isHeartbeat());
    }

    long getCallId() {
      return callId;
    }

    TermIndex getPreviousLog() {
      return previousLog;
    }

    long getFirstIndex() {
      return Optional.ofNullable(firstEntry).map(TermIndex::getIndex).orElse(RaftLog.INVALID_LOG_INDEX);
    }

    Timestamp getSendTime() {
      return sendTime;
    }

    void startRequestTimer() {
      timerContext = timer.time();
      sendTime = Timestamp.currentTime();
    }

    void stopRequestTimer() {
      timerContext.stop();
    }

    boolean isHeartbeat() {
      return entriesCount == 0;
    }

    @Override
    public String toString() {
      final String entries = entriesCount == 0? ""
          : entriesCount == 1? ",entry=" + firstEntry
          : ",entries=" + firstEntry + "..." + lastEntry;
      return JavaUtils.getClassSimpleName(getClass())
          + ":cid=" + callId
          + ",entriesCount=" + entriesCount
          + entries;
    }
  }

  static class RequestMap {
    private final Map<Long, AppendEntriesRequest> logRequests = new ConcurrentHashMap<>();
    private final Map<Long, AppendEntriesRequest> heartbeats = new ConcurrentHashMap<>();

    int logRequestsSize() {
      return logRequests.size();
    }

    void clear() {
      logRequests.clear();
      heartbeats.clear();
    }

    void put(AppendEntriesRequest request) {
      if (request.isHeartbeat()) {
        heartbeats.put(request.getCallId(), request);
      } else {
        logRequests.put(request.getCallId(), request);
      }
    }

    AppendEntriesRequest remove(AppendEntriesReplyProto reply) {
      return remove(reply.getServerReply().getCallId(), reply.getIsHearbeat());
    }

    AppendEntriesRequest remove(long cid, boolean isHeartbeat) {
      return isHeartbeat ? heartbeats.remove(cid): logRequests.remove(cid);
    }

    public AppendEntriesRequest handleTimeout(long callId, boolean heartbeat) {
      return heartbeat ? heartbeats.remove(callId) : logRequests.get(callId);
    }
  }
}