RATIS-2278. Follower Fails to Append Entries Due to Index Validation Race Condition in NavigableIndices (apache#1248)

Author: RkGrit

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

@@ -1634,7 +1634,10 @@ leaderId, getMemberId(), currentTerm, followerCommit, inconsistencyReplyNextInde
   }
   private CompletableFuture<Void> appendLog(List<LogEntryProto> entries) {
     final List<ConsecutiveIndices> entriesTermIndices = ConsecutiveIndices.convert(entries);
-    appendLogTermIndices.append(entriesTermIndices);
+    if (!appendLogTermIndices.append(entriesTermIndices)) {
+      // index already exists, return the last future
+      return appendLogFuture.get();
+    }
 
     return appendLogFuture.updateAndGet(f -> f.thenCompose(
             ignored -> JavaUtils.allOf(state.getLog().append(entries))))

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

@@ -136,15 +136,20 @@ synchronized Long getTerm(long index) {
       return floorEntry.getValue().getTerm(index);
     }
 
-    synchronized void append(List<ConsecutiveIndices> entriesTermIndices) {
-      for(ConsecutiveIndices indices : entriesTermIndices) {
-        // validate startIndex
-        final Map.Entry<Long, ConsecutiveIndices> lastEntry = map.lastEntry();
-        if (lastEntry != null) {
-          Preconditions.assertSame(lastEntry.getValue().getNextIndex(), indices.startIndex, "startIndex");
+    synchronized boolean append(List<ConsecutiveIndices> entriesTermIndices) {
+      for(int i = 0; i < entriesTermIndices.size(); i++) {
+        final ConsecutiveIndices indices = entriesTermIndices.get(i);
+        final ConsecutiveIndices previous = map.put(indices.startIndex, indices);
+        if (previous != null) {
+          // index already exists, revert this append
+          map.put(previous.startIndex, previous);
+          for(int j = 0; j < i; j++) {
+            map.remove(entriesTermIndices.get(j).startIndex);
+          }
+          return false;
         }
-        map.put(indices.startIndex, indices);
       }
+      return true;
     }
 
     synchronized void removeExisting(List<ConsecutiveIndices> entriesTermIndices) {