RATIS-2282. LogAppender Restart Due to Premature Log Entry Access During Concurrent Write Processing (#1249)

Author: RkGrit

ratis-server/src/main/java/org/apache/ratis/server/raftlog/segmented/LogSegment.java

@@ -454,10 +454,13 @@ private void append(Op op, ReferenceCountedObject<LogEntryProto> entryRef,
       boolean keepEntryInCache, Consumer<LogEntryProto> logConsumer) {
     final LogEntryProto entry = entryRef.retain();
     try {
-      final LogRecord record = appendLogRecord(op, entry);
+      final LogRecord record = new LogRecord(totalFileSize, entry);
       if (keepEntryInCache) {
         putEntryCache(record.getTermIndex(), entryRef, op);
       }
+      appendLogRecord(op, record);
+      totalFileSize += getEntrySize(entry, op);
+
       if (logConsumer != null) {
         logConsumer.accept(entry);
       }
@@ -466,24 +469,22 @@ private void append(Op op, ReferenceCountedObject<LogEntryProto> entryRef,
     }
   }
 
-
-  private LogRecord appendLogRecord(Op op, LogEntryProto entry) {
-    Objects.requireNonNull(entry, "entry == null");
+  private void appendLogRecord(Op op, LogRecord record) {
+    Objects.requireNonNull(record, "record == null");
     final LogRecord currentLast = records.getLast();
+
+    final long index = record.getTermIndex().getIndex();
     if (currentLast == null) {
-      Preconditions.assertTrue(entry.getIndex() == startIndex,
-          "gap between start index %s and first entry to append %s",
-          startIndex, entry.getIndex());
+      Preconditions.assertTrue(index == startIndex,
+              "%s: gap between start index %s and the entry to append %s", op, startIndex, index);
     } else {
-      Preconditions.assertTrue(entry.getIndex() == currentLast.getTermIndex().getIndex() + 1,
-          "gap between entries %s and %s", entry.getIndex(), currentLast.getTermIndex().getIndex());
+      final long currentLastIndex = currentLast.getTermIndex().getIndex();
+      Preconditions.assertTrue(index == currentLastIndex + 1,
+              "%s: gap between last entry %s and the entry to append %s", op, currentLastIndex, index);
     }
 
-    final LogRecord record = new LogRecord(totalFileSize, entry);
     records.append(record);
-    totalFileSize += getEntrySize(entry, op);
-    endIndex = entry.getIndex();
-    return record;
+    endIndex = index;
   }
 
   ReferenceCountedObject<LogEntryProto> getEntryFromCache(TermIndex ti) {
@@ -514,10 +515,7 @@ synchronized ReferenceCountedObject<LogEntryProto> loadCache(TermIndex ti) throw
   }
 
   LogRecord getLogRecord(long index) {
-    if (index >= startIndex && index <= endIndex) {
-      return records.get(index);
-    }
-    return null;
+    return records.get(index);
   }
 
   TermIndex getLastTermIndex() {