handle the lost quorum recovering from special config changing during zombie cleanup

Author: popduke

base-kv/base-kv-store-server/src/main/java/com/baidu/bifromq/basekv/store/option/KVRangeOptions.java

@@ -46,11 +46,12 @@ public class KVRangeOptions {
     @Builder.Default
     private int statsCollectIntervalSec = 5;
     @Builder.Default
-    private int zombieTimeoutSec = 300; // 5min
+    private int zombieTimeoutSec = 60; // 1min
     @Builder.Default
     private RaftConfig walRaftConfig = new RaftConfig()
         .setPreVote(true)
+        .setHeartbeatTimeoutTick(5) // 500ms
         .setInstallSnapshotTimeoutTick(6000) // 10min
-        .setElectionTimeoutTick(30) // 3 sec
+        .setElectionTimeoutTick(30) // 3s
         .setMaxSizePerAppend(100 * 1024 * 1024); // 100MB;
 }

base-kv/base-kv-store-server/src/main/java/com/baidu/bifromq/basekv/store/range/KVRangeFSM.java

@@ -40,6 +40,7 @@
 import static com.google.common.collect.Sets.union;
 import static java.util.Collections.emptySet;
 import static java.util.Collections.singleton;
+import static java.util.Collections.singletonList;
 
 import com.baidu.bifromq.base.util.AsyncRunner;
 import com.baidu.bifromq.baseenv.EnvProvider;
@@ -162,6 +163,7 @@ public class KVRangeFSM implements IKVRangeFSM {
     private final Subject<List<SplitHint>> splitHintsSubject = BehaviorSubject.<List<SplitHint>>create().toSerialized();
     private final Subject<Any> factSubject = BehaviorSubject.create();
     private final KVRangeOptions opts;
+    private final AtomicBoolean recovering = new AtomicBoolean();
     private final AtomicReference<Lifecycle> lifecycle = new AtomicReference<>(Lifecycle.Init);
     private final CompositeDisposable disposables = new CompositeDisposable();
     private final CompletableFuture<Void> closeSignal = new CompletableFuture<>();
@@ -1587,8 +1589,19 @@ private void detectZombieState(KVRangeDescriptor descriptor) {
     private void checkZombieState() {
         if (zombieAt > 0
             && Duration.ofMillis(HLC.INST.getPhysical() - zombieAt).toSeconds() > opts.getZombieTimeoutSec()) {
-            log.info("Zombie state detected, send quit signal.");
-            quitSignal.complete(null);
+            ClusterConfig clusterConfig = wal.latestClusterConfig();
+            if (clusterConfig.getNextVotersCount() > 0
+                && clusterConfig.getVotersList().equals(singletonList(hostStoreId))) {
+                // recover from single voter change lost quorum
+                if (recovering.compareAndSet(false, true)) {
+                    log.info("Recovering from lost quorum during changing config from single voter: \n{}",
+                        clusterConfig);
+                    wal.recover().whenComplete((v, e) -> recovering.set(false));
+                }
+            } else {
+                log.info("Zombie state detected, send quit signal.");
+                quitSignal.complete(null);
+            }
         }
     }
 

base-kv/base-kv-store-server/src/main/java/com/baidu/bifromq/basekv/store/wal/IKVRangeWAL.java

@@ -18,6 +18,7 @@
 import com.baidu.bifromq.basekv.proto.KVRangeSnapshot;
 import com.baidu.bifromq.basekv.raft.IRaftNode;
 import com.baidu.bifromq.basekv.raft.event.ElectionEvent;
+import com.baidu.bifromq.basekv.raft.proto.ClusterConfig;
 import com.baidu.bifromq.basekv.raft.proto.LogEntry;
 import com.baidu.bifromq.basekv.raft.proto.RaftMessage;
 import com.baidu.bifromq.basekv.raft.proto.RaftNodeStatus;
@@ -51,6 +52,8 @@ public interface IKVRangeWAL {
 
     KVRangeSnapshot latestSnapshot();
 
+    ClusterConfig latestClusterConfig();
+
     CompletableFuture<Long> propose(KVRangeCommand command);
 
     Observable<Map<String, RaftNodeSyncState>> replicationStatus();

base-kv/base-kv-store-server/src/main/java/com/baidu/bifromq/basekv/store/wal/KVRangeWAL.java

@@ -29,6 +29,7 @@
 import com.baidu.bifromq.basekv.raft.event.SnapshotRestoredEvent;
 import com.baidu.bifromq.basekv.raft.event.StatusChangedEvent;
 import com.baidu.bifromq.basekv.raft.event.SyncStateChangedEvent;
+import com.baidu.bifromq.basekv.raft.proto.ClusterConfig;
 import com.baidu.bifromq.basekv.raft.proto.LogEntry;
 import com.baidu.bifromq.basekv.raft.proto.RaftMessage;
 import com.baidu.bifromq.basekv.raft.proto.RaftNodeStatus;
@@ -209,6 +210,11 @@ public KVRangeSnapshot latestSnapshot() {
         return ZeroCopyParser.parse(raftNode.latestSnapshot(), KVRangeSnapshot.parser());
     }
 
+    @Override
+    public ClusterConfig latestClusterConfig() {
+        return raftNode.latestClusterConfig();
+    }
+
     @Override
     public CompletableFuture<Void> compact(KVRangeSnapshot snapshot) {
         return raftNode.compact(snapshot.toByteString(), snapshot.getLastAppliedIndex());