Fix bug in statemachine mode (#139)

openmessaging · Apr 28, 2022 · 52c8107 · 52c8107
1 parent 103cf0a
commit 52c8107
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 2 deletions.
diff --git a/src/main/java/io/openmessaging/storage/dledger/DLedgerEntryPusher.java b/src/main/java/io/openmessaging/storage/dledger/DLedgerEntryPusher.java
@@ -82,6 +82,7 @@ public DLedgerEntryPusher(DLedgerConfig dLedgerConfig, MemberState memberState,
         }
         this.entryHandler = new EntryHandler(logger);
         this.quorumAckChecker = new QuorumAckChecker(logger);
+        this.fsmCaller = Optional.empty();
     }
 
     public void startup() {
@@ -372,7 +373,7 @@ public void doWork() {
                     }
 
                     if (ackNum == 0) {
-                        checkResponseFuturesTimeout(quorumIndex);
+                        checkResponseFuturesTimeout(quorumIndex + 1);
                         waitForRunning(1);
                     }
 

diff --git a/src/main/java/io/openmessaging/storage/dledger/DLedgerServer.java b/src/main/java/io/openmessaging/storage/dledger/DLedgerServer.java
@@ -88,6 +88,7 @@ public DLedgerServer(DLedgerConfig dLedgerConfig) {
             t.setName("DLedgerServer-ScheduledExecutor");
             return t;
         });
+        this.fsmCaller = Optional.empty();
     }
 
     public void startup() {

diff --git a/src/main/java/io/openmessaging/storage/dledger/statemachine/StateMachineCaller.java b/src/main/java/io/openmessaging/storage/dledger/statemachine/StateMachineCaller.java
@@ -159,7 +159,7 @@ private void doCommitted(final long committedIndex) {
         // Check response timeout.
         if (iter.getCompleteAckNums() == 0) {
             if (this.entryPusher != null) {
-                this.entryPusher.checkResponseFuturesTimeout(this.lastAppliedIndex.get());
+                this.entryPusher.checkResponseFuturesTimeout(this.lastAppliedIndex.get() + 1);
             }
         }
     }