Fix bug in statemachine mode (#139)

Author: hzh0425

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);
                     }
 

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() {

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);
             }
         }
     }