commit d2f1c57fe60e3abe1bbfa1820297f9d60736bb42
author Potato <tanxinyu@apache.org> Wed Jan 31 00:12:23 2024 +0800
committer GitHub <noreply@github.com> Tue Jan 30 08:12:23 2024 -0800
tree 717517ee515bee5869b3b5d985a7802297eaf77e
parent cf897f02957bce7ab587a895143405c914adbf91

RATIS-2019 Fixed abnormal exit of StateMachineUpdater (#1033)

ratis-common/src/main/java/org/apache/ratis/util/JavaUtils.java

diff --git a/ratis-common/src/main/java/org/apache/ratis/util/JavaUtils.java b/ratis-common/src/main/java/org/apache/ratis/util/JavaUtils.java
index d3f899a..f689006 100644
--- a/ratis-common/src/main/java/org/apache/ratis/util/JavaUtils.java
+++ b/ratis-common/src/main/java/org/apache/ratis/util/JavaUtils.java
@@ -269,6 +269,10 @@
     return future;
   }
 
+  static boolean isCompletedNormally(CompletableFuture<?> future) {
+    return future.isDone() && !future.isCancelled() && !future.isCompletedExceptionally();
+  }
+
   static Throwable unwrapCompletionException(Throwable t) {
     Objects.requireNonNull(t, "t == null");
     return t instanceof CompletionException && t.getCause() != null? t.getCause(): t;

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

diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java
index 043c731..4f313a4 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/LeaderStateImpl.java
@@ -311,19 +311,16 @@
       return appliedIndexFuture;
     }
 
-    boolean isApplied(LogEntryProto logEntry) {
-      if (appliedIndexFuture.isDone()) {
-        return true;
+    boolean checkStartIndex(LogEntryProto logEntry) {
+      final boolean completed = logEntry.getIndex() == startIndex && appliedIndexFuture.complete(startIndex);
+      if (completed) {
+        LOG.info("Leader {} is ready since appliedIndex == startIndex == {}", LeaderStateImpl.this, startIndex);
       }
-      final long appliedIndex = logEntry != null? logEntry.getIndex(): server.getState().getLastAppliedIndex();
-      if (appliedIndex >= startIndex) {
-        appliedIndexFuture.complete(appliedIndex);
-        LOG.info("leader is ready since appliedIndex == {} >= startIndex == {}",
-            appliedIndex, startIndex);
-        return true;
-      } else {
-        return false;
-      }
+      return completed;
+    }
+
+    boolean isApplied() {
+      return JavaUtils.isCompletedNormally(appliedIndexFuture);
     }
   }
 
@@ -422,12 +419,11 @@
   }
 
   boolean isReady() {
-    return startupLogEntry.isInitialized() && startupLogEntry.get().isApplied(null);
+    return startupLogEntry.isInitialized() && startupLogEntry.get().isApplied();
   }
 
   void checkReady(LogEntryProto entry) {
-    Preconditions.assertTrue(startupLogEntry.isInitialized());
-    if (entry.getTerm() == getCurrentTerm() && startupLogEntry.get().isApplied(entry)) {
+    if (entry.getTerm() == server.getState().getCurrentTerm() && startupLogEntry.get().checkStartIndex(entry)) {
       server.getStateMachine().leaderEvent().notifyLeaderReady();
     }
   }

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

diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadIndexHeartbeats.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadIndexHeartbeats.java
index 3f31a25..d08a1ea 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadIndexHeartbeats.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/ReadIndexHeartbeats.java
@@ -22,6 +22,7 @@
 import org.apache.ratis.server.leader.LogAppender;
 import org.apache.ratis.server.raftlog.RaftLog;
 import org.apache.ratis.server.raftlog.RaftLogIndex;
+import org.apache.ratis.util.JavaUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -99,7 +100,7 @@
 
     boolean receive(LogAppender logAppender, AppendEntriesReplyProto proto,
                     Predicate<Predicate<RaftPeerId>> hasMajority) {
-      if (isCompletedNormally()) {
+      if (JavaUtils.isCompletedNormally(future)) {
         return true;
       }
 
@@ -112,16 +113,12 @@
         }
       }
 
-      return isCompletedNormally();
+      return JavaUtils.isCompletedNormally(future);
     }
 
     boolean isAcknowledged(RaftPeerId id) {
       return Optional.ofNullable(replies.get(id)).filter(HeartbeatAck::isAcknowledged).isPresent();
     }
-
-    boolean isCompletedNormally() {
-      return future.isDone() && !future.isCancelled() && !future.isCompletedExceptionally();
-    }
   }
 
   class AppendEntriesListeners {

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

diff --git a/ratis-server/src/main/java/org/apache/ratis/server/impl/RetryCacheImpl.java b/ratis-server/src/main/java/org/apache/ratis/server/impl/RetryCacheImpl.java
index a8bac4e..50d238b 100644
--- a/ratis-server/src/main/java/org/apache/ratis/server/impl/RetryCacheImpl.java
+++ b/ratis-server/src/main/java/org/apache/ratis/server/impl/RetryCacheImpl.java
@@ -67,7 +67,7 @@
     }
 
     boolean isCompletedNormally() {
-      return !failed && replyFuture.isDone() && !replyFuture.isCompletedExceptionally() && !replyFuture.isCancelled();
+      return !failed && JavaUtils.isCompletedNormally(replyFuture);
     }
 
     void updateResult(RaftClientReply reply) {