RATIS-2506. Delay consecutive GrpcLogAppender restart.

Author: szetszwo

ratis-grpc/src/main/java/org/apache/ratis/grpc/GrpcUtil.java

@@ -97,7 +97,7 @@ static Throwable unwrapThrowable(Throwable t) {
         return unwrapped;
       }
     }
-    return t;
+    return JavaUtils.unwrapCompletionException(t);
   }
 
   static IOException unwrapException(StatusRuntimeException se) {

ratis-grpc/src/main/java/org/apache/ratis/grpc/server/GrpcLogAppender.java

@@ -63,6 +63,8 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicLong;
 
+import static org.apache.ratis.server.raftlog.LogProtoUtils.toLogEntryTermIndexString;
+
 /**
  * A new log appender implementation using grpc bi-directional stream API.
  */
@@ -127,31 +129,21 @@ boolean isError() {
     }
   }
 
-  static class ReplyState {
+  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;
+      return getFollower().getErrorState().updateErrorCount(event.isError());
     }
   }
 
@@ -298,17 +290,25 @@ private boolean isSlowFollower() {
   private void mayWait() {
     // use lastSend time instead of lastResponse time
     try {
-      getEventAwaitForSignal().await(getWaitTimeMs() + errorWaitTimeMs(),
-          TimeUnit.MILLISECONDS);
+      // For errors, enforce sleep which cannot be waked up by signal
+      sleepForErrors();
+      // Normal await can be waked up by signal
+      getEventAwaitForSignal().await(getWaitTimeMs(), TimeUnit.MILLISECONDS);
     } catch (InterruptedException ie) {
-      LOG.warn(this + ": Wait interrupted by " + ie);
       Thread.currentThread().interrupt();
+      LOG.warn("{} is interrupted", this, ie);
     }
   }
 
-  private long errorWaitTimeMs() {
-    return errorRetryWaitPolicy.handleAttemptFailure(replyState::getErrorCount)
-        .getSleepTime().toLong(TimeUnit.MILLISECONDS);
+  private void sleepForErrors() throws InterruptedException {
+    final int errorCount = getFollower().getErrorState().getErrorCountToDelay();
+    if (errorCount < 1) {
+      return;
+    }
+
+    final TimeDuration sleepTime = errorRetryWaitPolicy.handleAttemptFailure(() -> errorCount).getSleepTime();
+    LOG.debug("{}: sleepForError {}, errorCount={}", this, sleepTime, errorCount);
+    sleepTime.sleep();
   }
 
   @Override
@@ -889,13 +889,9 @@ boolean isHeartbeat() {
 
     @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;
+          + "," + toLogEntryTermIndexString(entriesCount, firstEntry, lastEntry);
     }
   }
 

ratis-server-api/src/main/java/org/apache/ratis/server/leader/FollowerInfo.java

@@ -112,4 +112,33 @@ public interface FollowerInfo {
 
   /** Update lastRpcResponseTime and LastRespondedAppendEntriesSendTime */
   void updateLastRespondedAppendEntriesSendTime(Timestamp sendTime);
+
+  /** @return the error state. */
+  ErrorState getErrorState();
+
+  /** Error state such as the count for consecutive errors. */
+  interface ErrorState {
+    /**
+     * If it is an error, increment the count; otherwise, reset the count to 0.
+     *
+     * @return the updated error count.
+     */
+    int updateErrorCount(boolean isError);
+
+    /**
+     * Each error count is returned only once.
+     * The subsequent calls for the same error count, it returns 0.
+     * <p>
+     * For example,
+     * 1. Error count is 3
+     * 2. Calling getErrorCountToDelay() returns 3
+     * 3. Calling getErrorCountToDelay() again returns 0
+     * 4. Error count is incremented to 4
+     * 5. Calling getErrorCountToDelay() returns 4
+     * 6. Calling getErrorCountToDelay() again returns 0
+     *
+     * @return each error count only once.
+     */
+    int getErrorCountToDelay();
+  }
 }

ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerInfoImpl.java

@@ -42,6 +42,7 @@ class FollowerInfoImpl implements FollowerInfo {
   private final RaftLogIndex matchIndex = new RaftLogIndex("matchIndex", RaftLog.INVALID_LOG_INDEX);
   private final RaftLogIndex commitIndex = new RaftLogIndex("commitIndex", RaftLog.INVALID_LOG_INDEX);
   private final RaftLogIndex snapshotIndex = new RaftLogIndex("snapshotIndex", 0L);
+  private final ErrorState errorState = new ErrorStateImpl();
   private volatile boolean caughtUp;
   private volatile boolean ackInstallSnapshotAttempt = false;
 
@@ -240,4 +241,35 @@ public Timestamp getLastRespondedAppendEntriesSendTime() {
   public void updateLastRespondedAppendEntriesSendTime(Timestamp sendTime) {
     lastRespondedAppendEntriesSendTime.set(sendTime);
   }
+
+  @Override
+  public ErrorState getErrorState() {
+    return errorState;
+  }
+
+  static class ErrorStateImpl implements ErrorState {
+    /** The number of consecutive errors without getting a successful reply for a particular follower. */
+    private int errorCount = 0;
+    private int lastReturnedErrorCount = 0;
+
+    @Override
+    public synchronized int updateErrorCount(boolean isError) {
+      if (isError) {
+        errorCount++;
+      } else {
+        errorCount = 0;
+        lastReturnedErrorCount = 0;
+      }
+      return errorCount;
+    }
+
+    @Override
+    public synchronized int getErrorCountToDelay() {
+      if (errorCount == lastReturnedErrorCount) {
+        return 0;
+      }
+      lastReturnedErrorCount = errorCount;
+      return errorCount;
+    }
+  }
 }

ratis-server/src/main/java/org/apache/ratis/server/impl/FollowerState.java

@@ -158,8 +158,8 @@ private void runImpl() {
         }
         synchronized (server) {
           if (roleChangeChecking(electionTimeout)) {
-            LOG.info("{}: change to CANDIDATE, lastRpcElapsedTime:{}, electionTimeout:{}",
-                this, lastRpcTime.elapsedTime(), electionTimeout);
+            LOG.info("{}: change to CANDIDATE, lastRpcElapsedTime:{}ms, electionTimeout:{}",
+                this, lastRpcTime.elapsedTimeMs(), electionTimeout);
             server.getLeaderElectionMetrics().onLeaderElectionTimeout(); // Update timeout metric counters.
             // election timeout, should become a candidate
             server.changeToCandidate(false);

ratis-server/src/main/java/org/apache/ratis/server/impl/RaftServerImpl.java

@@ -150,6 +150,7 @@
 import static org.apache.ratis.server.impl.ServerProtoUtils.toReadIndexRequestProto;
 import static org.apache.ratis.server.impl.ServerProtoUtils.toRequestVoteReplyProto;
 import static org.apache.ratis.server.impl.ServerProtoUtils.toStartLeaderElectionReplyProto;
+import static org.apache.ratis.server.raftlog.LogProtoUtils.toLogEntryTermIndexString;
 import static org.apache.ratis.server.util.ServerStringUtils.toAppendEntriesReplyString;
 import static org.apache.ratis.server.util.ServerStringUtils.toAppendEntriesRequestString;
 import static org.apache.ratis.server.util.ServerStringUtils.toRequestVoteReplyString;
@@ -1703,6 +1704,11 @@ leaderId, getMemberId(), currentTerm, followerCommit, inconsistencyReplyNextInde
     final long commitIndex = effectiveCommitIndex(proto.getLeaderCommit(), previous, entries.size());
     final long matchIndex = isHeartbeat? RaftLog.INVALID_LOG_INDEX: entries.get(entries.size() - 1).getIndex();
     return appendFuture.whenCompleteAsync((r, t) -> {
+      if  (t != null) {
+        LOG.warn("{}: appendEntries* failed: {}", getMemberId(), toLogEntryTermIndexString(entries), t);
+      } else if (LOG.isDebugEnabled()) {
+        LOG.debug("{}: appendEntries* succeeded {}", getMemberId(), toLogEntryTermIndexString(entries));
+      }
       followerState.ifPresent(fs -> fs.updateLastRpcTime(FollowerState.UpdateType.APPEND_COMPLETE));
       timer.stop();
     }, getServerExecutor()).thenApply(v -> {
@@ -1753,7 +1759,7 @@ private long checkInconsistentAppendEntries(TermIndex previous, List<LogEntryPro
         && !(appendLogTermIndices != null && appendLogTermIndices.contains(previous))
         && !state.containsTermIndex(previous)) {
       final long replyNextIndex = Math.min(state.getNextIndex(), previous.getIndex());
-      LOG.info("{}: Failed appendEntries as previous log entry ({}) is not found", getMemberId(), previous);
+      LOG.info("{}: Failed appendEntries as previous log entry {} is not found", getMemberId(), previous);
       return replyNextIndex;
     }
 

ratis-server/src/main/java/org/apache/ratis/server/raftlog/LogProtoUtils.java

@@ -92,6 +92,18 @@ public static String toLogEntriesShortString(List<LogEntryProto> entries,
         : "size=" + entries.size() + ", first=" + toLogEntryString(entries.get(0), stateMachineToString);
   }
 
+  public static String toLogEntryTermIndexString(List<LogEntryProto> entries) {
+    final int n = entries.size();
+    return n == 0 ? toLogEntryTermIndexString(n, null, null)
+        : toLogEntryTermIndexString(n, TermIndex.valueOf(entries.get(0)), TermIndex.valueOf(entries.get(n - 1)));
+  }
+
+  public static String toLogEntryTermIndexString(int n, TermIndex first, TermIndex last) {
+    return n == 0 ? "HEARTBEAT"
+        : n == 1 ? "entry" + first
+        : n + " entries:" + first + "..." + last;
+  }
+
   public static LogEntryProto toLogEntryProto(RaftConfiguration conf, Long term, long index) {
     final LogEntryProto.Builder b = LogEntryProto.newBuilder();
     Optional.ofNullable(term).ifPresent(b::setTerm);