(fix) Iterator#setErrorAndRollback may break ReadIndex promise, #317 (#361)

* (fix) Iterator#setErrorAndRollback may break ReadIndex promise, #317

* (feat) Adds comment

* fix hang with CI

* (fix) FSMCallerTest#testOnCommittedError

* feat/release 1.3.0 (#363)

* minor change

* release 1.3.0

* release 1.3.0

* fix hang with ut

Author: killme2008

jraft-core/pom.xml

@@ -5,7 +5,7 @@
     <parent>
         <artifactId>jraft-parent</artifactId>
         <groupId>com.alipay.sofa</groupId>
-        <version>1.2.6</version>
+        <version>1.3.0</version>
     </parent>
     <artifactId>jraft-core</artifactId>
     <packaging>jar</packaging>

jraft-core/src/main/java/com/alipay/sofa/jraft/Iterator.java

@@ -67,8 +67,9 @@ public interface Iterator extends java.util.Iterator<ByteBuffer> {
      * |ntail| tasks (starting from the last iterated one) as not applied. After
      * this point, no further changes on the StateMachine as well as the Node
      * would be allowed and you should try to repair this replica or just drop it.
-     * 
-     * If |statInfo| is not NULL, it should describe the detail of the error.
+     *
+     * @param ntail the number of tasks (starting from the last iterated one)  considered as not to be applied.
+     * @param st    Status to describe the detail of the error.
      */
     void setErrorAndRollback(final long ntail, final Status st);
 }

jraft-core/src/main/java/com/alipay/sofa/jraft/ReadOnlyService.java

@@ -17,6 +17,7 @@
 package com.alipay.sofa.jraft;
 
 import com.alipay.sofa.jraft.closure.ReadIndexClosure;
+import com.alipay.sofa.jraft.error.RaftException;
 import com.alipay.sofa.jraft.option.ReadOnlyServiceOptions;
 
 /**
@@ -43,4 +44,10 @@ public interface ReadOnlyService extends Lifecycle<ReadOnlyServiceOptions> {
      */
     void join() throws InterruptedException;
 
+    /**
+     * Called when the node is turned into error state.
+     * @param error error with raft info
+     */
+    void setError(final RaftException error);
+
 }

jraft-core/src/main/java/com/alipay/sofa/jraft/core/FSMCallerImpl.java

@@ -515,10 +515,10 @@ private void doCommitted(final long committedIndex) {
             final long lastIndex = iterImpl.getIndex() - 1;
             final long lastTerm = this.logManager.getTerm(lastIndex);
             final LogId lastAppliedId = new LogId(lastIndex, lastTerm);
-            this.lastAppliedIndex.set(committedIndex);
+            this.lastAppliedIndex.set(lastIndex);
             this.lastAppliedTerm = lastTerm;
             this.logManager.setAppliedId(lastAppliedId);
-            notifyLastAppliedIndexUpdated(committedIndex);
+            notifyLastAppliedIndexUpdated(lastIndex);
         } finally {
             this.nodeMetrics.recordLatency("fsm-commit", Utils.monotonicMs() - startMs);
         }

jraft-core/src/main/java/com/alipay/sofa/jraft/core/NodeImpl.java

@@ -2373,6 +2373,9 @@ public void onError(final RaftException error) {
             // onError of fsmCaller is guaranteed to be executed once.
             this.fsmCaller.onError(error);
         }
+        if (this.readOnlyService != null) {
+            this.readOnlyService.setError(error);
+        }
         this.writeLock.lock();
         try {
             // If it is leader, need to wake up a new one;

jraft-core/src/main/java/com/alipay/sofa/jraft/core/ReadOnlyServiceImpl.java

@@ -39,6 +39,7 @@
 import com.alipay.sofa.jraft.entity.ReadIndexState;
 import com.alipay.sofa.jraft.entity.ReadIndexStatus;
 import com.alipay.sofa.jraft.error.RaftError;
+import com.alipay.sofa.jraft.error.RaftException;
 import com.alipay.sofa.jraft.option.RaftOptions;
 import com.alipay.sofa.jraft.option.ReadOnlyServiceOptions;
 import com.alipay.sofa.jraft.rpc.RpcRequests.ReadIndexRequest;
@@ -82,6 +83,8 @@ public class ReadOnlyServiceImpl implements ReadOnlyService, LastAppliedLogIndex
 
     private NodeMetrics                                nodeMetrics;
 
+    private volatile RaftException                     error;
+
     // <logIndex, statusList>
     private final TreeMap<Long, List<ReadIndexStatus>> pendingNotifyStatus         = new TreeMap<>();
 
@@ -217,6 +220,20 @@ private void executeReadIndexEvents(final List<ReadIndexEvent> events) {
         this.node.handleReadIndexRequest(request, new ReadIndexResponseClosure(states, request));
     }
 
+    private void resetPendingStatusError(final Status st) {
+        this.lock.lock();
+        try {
+            for (final List<ReadIndexStatus> statuses : this.pendingNotifyStatus.values()) {
+                for (final ReadIndexStatus status : statuses) {
+                    reportError(status, st);
+                }
+            }
+            this.pendingNotifyStatus.clear();
+        } finally {
+            this.lock.unlock();
+        }
+    }
+
     @Override
     public boolean init(final ReadOnlyServiceOptions opts) {
         this.node = opts.getNode();
@@ -250,6 +267,13 @@ public boolean init(final ReadOnlyServiceOptions opts) {
         return true;
     }
 
+    @Override
+    public synchronized void setError(final RaftException error) {
+        if (this.error == null) {
+            this.error = error;
+        }
+    }
+
     @Override
     public synchronized void shutdown() {
         if (this.shutdownLatch != null) {
@@ -267,6 +291,7 @@ public void join() throws InterruptedException {
             this.shutdownLatch.await();
         }
         this.readIndexDisruptor.shutdown();
+        resetPendingStatusError(new Status(RaftError.ESTOP, "Node is quit."));
         this.scheduledExecutorService.awaitTermination(5, TimeUnit.SECONDS);
     }
 
@@ -331,6 +356,16 @@ public void onApplied(final long appliedIndex) {
                 }
 
             }
+
+            /*
+             * Remaining pending statuses are notified by error if it is presented.
+             * When the node is in error state, consider following situations:
+             * 1. If commitIndex > appliedIndex, then all pending statuses should be notified by error status.
+             * 2. When commitIndex == appliedIndex, there will be no more pending statuses.
+             */
+            if (this.error != null) {
+                resetPendingStatusError(this.error.getStatus());
+            }
         } finally {
             this.lock.unlock();
             if (pendingStatuses != null && !pendingStatuses.isEmpty()) {
@@ -356,6 +391,20 @@ TreeMap<Long, List<ReadIndexStatus>> getPendingNotifyStatus() {
         return this.pendingNotifyStatus;
     }
 
+    private void reportError(final ReadIndexStatus status, final Status st) {
+        final long nowMs = Utils.monotonicMs();
+        final List<ReadIndexState> states = status.getStates();
+        final int taskCount = states.size();
+        for (int i = 0; i < taskCount; i++) {
+            final ReadIndexState task = states.get(i);
+            final ReadIndexClosure done = task.getDone();
+            if (done != null) {
+                this.nodeMetrics.recordLatency("read-index", nowMs - task.getStartTimeMs());
+                done.run(st);
+            }
+        }
+    }
+
     private void notifySuccess(final ReadIndexStatus status) {
         final long nowMs = Utils.monotonicMs();
         final List<ReadIndexState> states = status.getStates();

jraft-core/src/main/java/com/alipay/sofa/jraft/core/Replicator.java

@@ -753,7 +753,7 @@ boolean prepareEntry(final long nextSendingIndex, final int offset, final RaftOu
         }
         emb.setTerm(entry.getId().getTerm());
         if (entry.hasChecksum()) {
-            emb.setChecksum(entry.getChecksum()); //since 1.2.6
+            emb.setChecksum(entry.getChecksum()); // since 1.2.6
         }
         emb.setType(entry.getType());
         if (entry.getPeers() != null) {

jraft-core/src/main/java/com/alipay/sofa/jraft/entity/ReadIndexStatus.java

@@ -27,9 +27,9 @@
  */
 public class ReadIndexStatus {
 
-    private final ReadIndexRequest     request; //raw request
-    private final List<ReadIndexState> states; //read index requests in batch.
-    private final long                 index;  //committed log index.
+    private final ReadIndexRequest     request; // raw request
+    private final List<ReadIndexState> states; // read index requests in batch.
+    private final long                 index;  // committed log index.
 
     public ReadIndexStatus(List<ReadIndexState> states, ReadIndexRequest request, long index) {
         super();

jraft-core/src/test/java/com/alipay/sofa/jraft/core/FSMCallerTest.java

@@ -16,6 +16,10 @@
  */
 package com.alipay.sofa.jraft.core;
 
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
 import java.util.concurrent.CountDownLatch;
 
 import org.junit.After;
@@ -47,10 +51,6 @@
 import com.alipay.sofa.jraft.storage.snapshot.SnapshotWriter;
 import com.alipay.sofa.jraft.test.TestUtils;
 
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertFalse;
-import static org.junit.Assert.assertTrue;
-
 @RunWith(value = MockitoJUnitRunner.class)
 public class FSMCallerTest {
     private FSMCallerImpl    fsmCaller;
@@ -67,9 +67,9 @@ public void setup() {
         this.fsmCaller = new FSMCallerImpl();
         this.closureQueue = new ClosureQueueImpl();
         final FSMCallerOptions opts = new FSMCallerOptions();
-        Mockito.when(node.getNodeMetrics()).thenReturn(new NodeMetrics(false));
-        opts.setNode(node);
-        opts.setFsm(fsm);
+        Mockito.when(this.node.getNodeMetrics()).thenReturn(new NodeMetrics(false));
+        opts.setNode(this.node);
+        opts.setFsm(this.fsm);
         opts.setLogManager(this.logManager);
         opts.setBootstrapId(new LogId(10, 1));
         opts.setClosureQueue(this.closureQueue);
@@ -93,13 +93,13 @@ public void testShutdownJoin() throws Exception {
 
     @Test
     public void testOnCommittedError() throws Exception {
-        Mockito.when(logManager.getTerm(10)).thenReturn(1L);
-        Mockito.when(logManager.getEntry(11)).thenReturn(null);
+        Mockito.when(this.logManager.getTerm(10)).thenReturn(1L);
+        Mockito.when(this.logManager.getEntry(11)).thenReturn(null);
 
         assertTrue(this.fsmCaller.onCommitted(11));
 
         this.fsmCaller.flush();
-        assertEquals(this.fsmCaller.getLastAppliedIndex(), 11);
+        assertEquals(this.fsmCaller.getLastAppliedIndex(), 10);
         Mockito.verify(this.logManager).setAppliedId(new LogId(10, 1));
         assertFalse(this.fsmCaller.getError().getStatus().isOk());
         assertEquals("Fail to get entry at index=11 while committed_index=11", this.fsmCaller.getError().getStatus()
@@ -111,8 +111,8 @@ public void testOnCommitted() throws Exception {
         final LogEntry log = new LogEntry(EntryType.ENTRY_TYPE_DATA);
         log.getId().setIndex(11);
         log.getId().setTerm(1);
-        Mockito.when(logManager.getTerm(11)).thenReturn(1L);
-        Mockito.when(logManager.getEntry(11)).thenReturn(log);
+        Mockito.when(this.logManager.getTerm(11)).thenReturn(1L);
+        Mockito.when(this.logManager.getEntry(11)).thenReturn(log);
         final ArgumentCaptor<Iterator> itArg = ArgumentCaptor.forClass(Iterator.class);
 
         assertTrue(this.fsmCaller.onCommitted(11));
@@ -138,7 +138,7 @@ public void testOnSnapshotLoad() throws Exception {
         this.fsmCaller.onSnapshotLoad(new LoadSnapshotClosure() {
 
             @Override
-            public void run(Status status) {
+            public void run(final Status status) {
                 assertTrue(status.isOk());
                 latch.countDown();
             }
@@ -164,7 +164,7 @@ public void testOnSnapshotLoadFSMError() throws Exception {
         this.fsmCaller.onSnapshotLoad(new LoadSnapshotClosure() {
 
             @Override
-            public void run(Status status) {
+            public void run(final Status status) {
                 assertFalse(status.isOk());
                 assertEquals(-1, status.getCode());
                 assertEquals("StateMachine onSnapshotLoad failed", status.getErrorMsg());
@@ -186,14 +186,14 @@ public void testOnSnapshotSaveEmptyConf() throws Exception {
         this.fsmCaller.onSnapshotSave(new SaveSnapshotClosure() {
 
             @Override
-            public void run(Status status) {
+            public void run(final Status status) {
                 assertFalse(status.isOk());
                 assertEquals("Empty conf entry for lastAppliedIndex=10", status.getErrorMsg());
                 latch.countDown();
             }
 
             @Override
-            public SnapshotWriter start(SnapshotMeta meta) {
+            public SnapshotWriter start(final SnapshotMeta meta) {
                 // TODO Auto-generated method stub
                 return null;
             }
@@ -209,12 +209,12 @@ public void testOnSnapshotSave() throws Exception {
         final SaveSnapshotClosure done = new SaveSnapshotClosure() {
 
             @Override
-            public void run(Status status) {
+            public void run(final Status status) {
 
             }
 
             @Override
-            public SnapshotWriter start(SnapshotMeta meta) {
+            public SnapshotWriter start(final SnapshotMeta meta) {
                 assertEquals(10, meta.getLastIncludedIndex());
                 return writer;
             }
@@ -269,7 +269,7 @@ public void testOnSnapshotLoadStale() throws Exception {
         this.fsmCaller.onSnapshotLoad(new LoadSnapshotClosure() {
 
             @Override
-            public void run(Status status) {
+            public void run(final Status status) {
                 assertFalse(status.isOk());
                 assertEquals(RaftError.ESTALE, status.getRaftError());
                 latch.countDown();