1. solve the issue that commit index may go back in some extreme situation during config change

2. code cleanup

Author: popduke

base-kv/base-kv-raft/src/main/java/com/baidu/bifromq/basekv/raft/BasicStateStoreTest.java

@@ -34,6 +34,9 @@
 import org.testng.Assert;
 import org.testng.annotations.Test;
 
+/**
+ * Base class for state store test cases.
+ */
 public abstract class BasicStateStoreTest {
 
     protected abstract IRaftStateStore createStorage(String id, Snapshot initSnapshot);
@@ -75,12 +78,6 @@ public void testSaveAndLoadCurrentTerm() {
         assertEquals(stateStorage.currentTerm(), 0);
         stateStorage.saveTerm(1);
         assertEquals(stateStorage.currentTerm(), 1);
-        try {
-            stateStorage.saveTerm(0);
-            fail();
-        } catch (Throwable t) {
-
-        }
     }
 
     @Test
@@ -127,7 +124,7 @@ public void testAppendFirstEntryFailedWithLowerBoundCheck() {
             .build();
         IRaftStateStore stateStorage = createStorage(localId(), initSnapshot);
         try {
-            stateStorage.append(asList(LogEntry.newBuilder()
+            stateStorage.append(List.of(LogEntry.newBuilder()
                 .setTerm(1)
                 .setIndex(5)
                 .setData(ByteString.EMPTY)
@@ -137,7 +134,7 @@ public void testAppendFirstEntryFailedWithLowerBoundCheck() {
             assertTrue(e instanceof IndexOutOfBoundsException);
         }
         try {
-            stateStorage.append(asList(LogEntry.newBuilder()
+            stateStorage.append(List.of(LogEntry.newBuilder()
                 .setTerm(1)
                 .setIndex(7)
                 .setData(ByteString.EMPTY)
@@ -159,7 +156,7 @@ public void testAppendFirstEntryFailedWithLowerBoundCheck() {
                 .build()), true);
 
         try {
-            stateStorage.append(asList(LogEntry.newBuilder()
+            stateStorage.append(List.of(LogEntry.newBuilder()
                 .setTerm(1)
                 .setIndex(5)
                 .setData(ByteString.EMPTY)
@@ -169,7 +166,7 @@ public void testAppendFirstEntryFailedWithLowerBoundCheck() {
             assertTrue(e instanceof IndexOutOfBoundsException);
         }
         try {
-            stateStorage.append(asList(LogEntry.newBuilder()
+            stateStorage.append(List.of(LogEntry.newBuilder()
                 .setTerm(1)
                 .setIndex(9)
                 .setData(ByteString.EMPTY)
@@ -200,7 +197,7 @@ public void testAsyncAppend() throws InterruptedException {
             assertEquals(stateStorage.entryAt(stateStorage.lastIndex()).get().getIndex(), stateStorage.lastIndex());
         }
         countDownLatch.await();
-        assertTrue(1 <= stabledIndexes.size());
+        assertFalse(stabledIndexes.isEmpty());
         assertTrue(stabledIndexes.get(0) >= 1);
     }
 
@@ -270,7 +267,7 @@ public void testTruncateWholeLog() {
         assertEquals(stateStorage.firstIndex(), 6);
         assertEquals(stateStorage.lastIndex(), 5);
 
-        List<LogEntry> newEntries = asList(LogEntry.newBuilder()
+        List<LogEntry> newEntries = List.of(LogEntry.newBuilder()
             .setTerm(1)
             .setIndex(6)
             .setData(ByteString.EMPTY)
@@ -293,7 +290,7 @@ public void testTruncateWholeLog() {
         assertEquals(stateStorage.firstIndex(), 6);
         assertEquals(stateStorage.lastIndex(), 7);
 
-        newEntries = asList(LogEntry.newBuilder()
+        newEntries = List.of(LogEntry.newBuilder()
             .setTerm(1)
             .setIndex(6)
             .setData(ByteString.EMPTY)

base-kv/base-kv-raft/src/main/java/com/baidu/bifromq/basekv/raft/ClusterConfigHelper.java

@@ -13,62 +13,61 @@
 
 package com.baidu.bifromq.basekv.raft;
 
-
 import com.baidu.bifromq.basekv.raft.proto.RaftNodeSyncState;
 
 interface IPeerLogReplicator {
 
     /**
-     * Current matching index
+     * Current matching index.
      *
-     * @return
+     * @return the index of the last log entry that has been replicated to the given peer
      */
     long matchIndex();
 
     /**
-     * Next index to send
+     * Next index to send.
      *
-     * @return
+     * @return the index of the next log entry to send to the given peer
      */
     long nextIndex();
 
     /**
-     * Current status
+     * Current status of the replicator.
      *
-     * @return
+     * @return the current status of the replicator
      */
     RaftNodeSyncState status();
 
     /**
-     * an external clock signal to drive the state machine forward in case no other stimuli available
+     * an external clock signal to drive the state machine forward in case no other stimuli happens.
      *
      * @return true if the replicator has changed its state after tick
      */
     boolean tick();
 
     /**
-     * the amount of matchIndex advanced per tick always non-negative
+     * the amount of matchIndex advanced per tick always non-negative.
      *
-     * @return
+     * @return the amount of matchIndex advanced per tick
      */
     long catchupRate();
 
     /**
-     * a flag indicating whether the append entries for given peer should be paused
+     * a flag indicating whether the append entries for given peer should be paused.
      *
-     * @return
+     * @return true if the replicator has changed its state after calling this method
      */
     boolean pauseReplicating();
 
     /**
-     * a flag indicating whether the given peer need a heartbeat due to heartbeatTimeoutTick exceed
+     * a flag indicating whether the given peer need a heartbeat due to heartbeatTimeoutTick exceed.
      *
      * @return true if peer need a heartbeat
      */
     boolean needHeartbeat();
 
     /**
-     * backoff the next index when peer follower rejected the append entries request
+     * backoff the next index when peer follower rejected the append entries request.
      *
      * @param peerRejectedIndex the index of mismatched log which is literally the prevLogIndex in appendEntries rpc
      * @param peerLastIndex     the index of last log entry in peer's raft log
@@ -77,17 +76,17 @@ interface IPeerLogReplicator {
     boolean backoff(long peerRejectedIndex, long peerLastIndex);
 
     /**
-     * update the match index when peer follower accepted the append entries request
+     * update the match index when peer follower accepted the append entries request.
      *
      * @param peerLastIndex the index of last log entry in peer's raft log
      * @return true if the replicator has changed its state after calling this method
      */
     boolean confirmMatch(long peerLastIndex);
 
     /**
-     * advance the next index after sending log entries up to endIndex(inclusively) to follower
+     * advance the next index after sending log entries up to endIndex(inclusively) to follower.
      *
-     * @param endIndex
+     * @param endIndex the index of the last log entry to send to the given peer
      * @return true if the replicator has changed its state after calling this method
      */
     boolean replicateBy(long endIndex);

base-kv/base-kv-raft/src/main/java/com/baidu/bifromq/basekv/raft/IPeerLogReplicator.java

@@ -16,7 +16,7 @@
 import com.baidu.bifromq.basekv.raft.proto.ClusterConfig;
 import com.baidu.bifromq.basekv.raft.proto.RaftNodeStatus;
 
-public interface IRaftNodeState {
+interface IRaftNodeState {
     String id();
 
     RaftNodeStatus getState();