RATIS-2283. GrpcLogAppender Thread Restart Leaves catchup=false, Blocking Reconfiguration Progress (apache#1250)

Author: RkGrit

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

@@ -143,7 +143,7 @@ default SnapshotInfo shouldInstallSnapshot() {
     // we should install snapshot if the follower needs to catch up and:
     // 1. there is no local log entry but there is snapshot
     // 2. or the follower's next index is smaller than the log start index
-    // 3. or the follower is bootstrapping and has not installed any snapshot yet
+    // 3. or the follower is bootstrapping (i.e. not yet caught up) and has not installed any snapshot yet
     final FollowerInfo follower = getFollower();
     final boolean isFollowerBootstrapping = getLeaderState().isFollowerBootstrapping(follower);
     final SnapshotInfo snapshot = getServer().getStateMachine().getLatestSnapshot();

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

@@ -818,7 +818,9 @@ public void onFollowerSuccessAppendEntries(FollowerInfo follower) {
 
   @Override
   public boolean isFollowerBootstrapping(FollowerInfo follower) {
-    return isBootStrappingPeer(follower.getId());
+    // It is better to check caught up than staging state
+    // since a follower may have already caught up but still in the staging state.
+    return !isCaughtUp(follower);
   }
 
   private void checkStaging() {
@@ -850,7 +852,12 @@ private void checkStaging() {
   }
 
   boolean isBootStrappingPeer(RaftPeerId peerId) {
-    return Optional.ofNullable(stagingState).map(s -> s.contains(peerId)).orElse(false);
+    final Optional<LogAppender> info = getLogAppender(peerId);
+    if (info.isPresent()) {
+      return !isCaughtUp(info.get().getFollower());
+    }
+    final ConfigurationStagingState staging = stagingState;
+    return staging != null && staging.contains(peerId);
   }
 
   void submitUpdateCommitEvent() {