RATIS-2235. Allow only one thread to perform appendLog (#1206)

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

@@ -132,6 +132,7 @@
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
@@ -249,6 +250,8 @@ public long[] getFollowerNextIndices() {
   private final AtomicBoolean firstElectionSinceStartup = new AtomicBoolean(true);
   private final ThreadGroup threadGroup;
 
+  private final AtomicReference<CompletableFuture<Void>> appendLogFuture;
+
   RaftServerImpl(RaftGroup group, StateMachine stateMachine, RaftServerProxy proxy, RaftStorage.StartupOption option)
       throws IOException {
     final RaftPeerId id = proxy.getId();
@@ -282,6 +285,7 @@ public long[] getFollowerNextIndices() {
     this.transferLeadership = new TransferLeadership(this, properties);
     this.snapshotRequestHandler = new SnapshotManagementRequestHandler(this);
     this.snapshotInstallationHandler = new SnapshotInstallationHandler(this, properties);
+    this.appendLogFuture = new AtomicReference<>(CompletableFuture.completedFuture(null));
 
     this.serverExecutor = ConcurrentUtils.newThreadPoolWithMax(
         RaftServerConfigKeys.ThreadPool.serverCached(properties),
@@ -630,6 +634,15 @@ synchronized void changeToLeader() {
 
   @Override
   public Collection<CommitInfoProto> getCommitInfos() {
+    try {
+      return getCommitInfosImpl();
+    } catch (Throwable t) {
+      LOG.warn("{} Failed to getCommitInfos", getMemberId(), t);
+      return Collections.emptyList();
+    }
+  }
+
+  private Collection<CommitInfoProto> getCommitInfosImpl() {
     final List<CommitInfoProto> infos = new ArrayList<>();
     // add the commit info of this server
     final long commitIndex = updateCommitInfoCache();
@@ -803,8 +816,6 @@ private CompletableFuture<RaftClientReply> appendTransaction(
     CodeInjectionForTesting.execute(APPEND_TRANSACTION, getId(),
         request.getClientId(), request, context, cacheEntry);
 
-    assertLifeCycleState(LifeCycle.States.RUNNING);
-
     final PendingRequest pending;
     synchronized (this) {
       final CompletableFuture<RaftClientReply> reply = checkLeaderState(request, cacheEntry);
@@ -823,6 +834,7 @@ private CompletableFuture<RaftClientReply> appendTransaction(
         return cacheEntry.getReplyFuture();
       }
       try {
+        assertLifeCycleState(LifeCycle.States.RUNNING);
         state.appendLog(context);
       } catch (StateMachineException e) {
         // the StateMachineException is thrown by the SM in the preAppend stage.
@@ -834,6 +846,9 @@ private CompletableFuture<RaftClientReply> appendTransaction(
           leaderState.submitStepDownEvent(LeaderState.StepDownReason.STATE_MACHINE_EXCEPTION);
         }
         return CompletableFuture.completedFuture(exceptionReply);
+      } catch (ServerNotReadyException e) {
+        final RaftClientReply exceptionReply = newExceptionReply(request, e);
+        return CompletableFuture.completedFuture(exceptionReply);
       }
 
       // put the request into the pending queue
@@ -1585,9 +1600,9 @@ leaderId, getMemberId(), currentTerm, followerCommit, inconsistencyReplyNextInde
       state.updateConfiguration(entries);
     }
     future.join();
+    final CompletableFuture<Void> appendLog = entries.isEmpty()? CompletableFuture.completedFuture(null)
+        : appendLog(entries);
 
-    final List<CompletableFuture<Long>> futures = entries.isEmpty() ? Collections.emptyList()
-        : state.getLog().append(entries);
     proto.getCommitInfosList().forEach(commitInfoCache::update);
 
     CodeInjectionForTesting.execute(LOG_SYNC, getId(), null);
@@ -1601,7 +1616,7 @@ 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 JavaUtils.allOf(futures).whenCompleteAsync((r, t) -> {
+    return appendLog.whenCompleteAsync((r, t) -> {
       followerState.ifPresent(fs -> fs.updateLastRpcTime(FollowerState.UpdateType.APPEND_COMPLETE));
       timer.stop();
     }, getServerExecutor()).thenApply(v -> {
@@ -1618,6 +1633,10 @@ leaderId, getMemberId(), currentTerm, followerCommit, inconsistencyReplyNextInde
       return reply;
     });
   }
+  private CompletableFuture<Void> appendLog(List<LogEntryProto> entries) {
+    return appendLogFuture.updateAndGet(f -> f.thenCompose(
+            ignored -> JavaUtils.allOf(state.getLog().append(entries))));
+  }
 
   private long checkInconsistentAppendEntries(TermIndex previous, List<LogEntryProto> entries) {
     // Check if a snapshot installation through state machine is in progress.