fix: base read-index failure threshold (#1249)

Author: fengjiachun

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

@@ -1514,19 +1514,19 @@ private class ReadIndexHeartbeatResponseClosure extends RpcResponseClosureAdapte
         final ReadIndexResponse.Builder             respBuilder;
         final RpcResponseClosure<ReadIndexResponse> closure;
         final int                                   quorum;
-        final int                                   failPeersThreshold;
+        final int                                   expectedFollowerResponses;
         int                                         ackSuccess;
         int                                         ackFailures;
         boolean                                     isDone;
 
         public ReadIndexHeartbeatResponseClosure(final RpcResponseClosure<ReadIndexResponse> closure,
                                                  final ReadIndexResponse.Builder rb, final int quorum,
-                                                 final int peersCount) {
+                                                 final int expectedFollowerResponses) {
             super();
             this.closure = closure;
             this.respBuilder = rb;
             this.quorum = quorum;
-            this.failPeersThreshold = peersCount % 2 == 0 ? (quorum - 1) : quorum;
+            this.expectedFollowerResponses = expectedFollowerResponses;
             this.ackSuccess = 0;
             this.ackFailures = 0;
             this.isDone = false;
@@ -1548,7 +1548,8 @@ public synchronized void run(final Status status) {
                 this.closure.setResponse(this.respBuilder.build());
                 this.closure.run(Status.OK());
                 this.isDone = true;
-            } else if (this.ackFailures >= this.failPeersThreshold) {
+                // The extra 1 is the leader's own vote, so only follower failures are counted here.
+            } else if (this.ackFailures > this.expectedFollowerResponses + 1 - this.quorum) {
                 this.respBuilder.setSuccess(false);
                 this.closure.setResponse(this.respBuilder.build());
                 this.closure.run(Status.OK());
@@ -1654,8 +1655,14 @@ private void readLeader(final ReadIndexRequest request, final ReadIndexResponse.
                 final List<PeerId> peers = this.conf.getConf().getPeers();
                 Requires.requireTrue(peers != null && !peers.isEmpty(), "Empty peers");
                 final List<PeerId> targetPeers = filterPeersForReadIndex(peers);
+                int expectedResponses = 0;
+                for (final PeerId peer : targetPeers) {
+                    if (!peer.equals(this.serverId)) {
+                        expectedResponses++;
+                    }
+                }
                 final ReadIndexHeartbeatResponseClosure heartbeatDone = new ReadIndexHeartbeatResponseClosure(closure,
-                    respBuilder, quorum, peers.size());
+                    respBuilder, quorum, expectedResponses);
                 // Send heartbeat requests to followers
                 for (final PeerId peer : targetPeers) {
                     if (peer.equals(this.serverId)) {
@@ -1691,7 +1698,6 @@ private List<PeerId> filterPeersForReadIndex(final List<PeerId> allPeers) {
             if (peer.equals(this.serverId)) {
                 continue;
             }
-
             final ThreadId rid = this.replicatorGroup.getReplicator(peer);
             if (rid == null) {
                 continue;

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

@@ -17,6 +17,7 @@
 package com.alipay.sofa.jraft.core;
 
 import java.io.File;
+import java.lang.reflect.Constructor;
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -72,6 +73,10 @@
 import com.alipay.sofa.jraft.option.BootstrapOptions;
 import com.alipay.sofa.jraft.option.NodeOptions;
 import com.alipay.sofa.jraft.option.RaftOptions;
+import com.alipay.sofa.jraft.rpc.RpcRequests.AppendEntriesResponse;
+import com.alipay.sofa.jraft.rpc.RpcRequests.ReadIndexResponse;
+import com.alipay.sofa.jraft.rpc.RpcResponseClosure;
+import com.alipay.sofa.jraft.rpc.RpcResponseClosureAdapter;
 import com.alipay.sofa.jraft.rpc.RaftRpcServerFactory;
 import com.alipay.sofa.jraft.rpc.RpcServer;
 import com.alipay.sofa.jraft.storage.SnapshotThrottle;
@@ -1454,6 +1459,61 @@ public void run(final Status status, final long index, final byte[] reqCtx) {
         cluster.stopAll();
     }
 
+    @Test
+    public void testReadIndexHeartbeatResponseClosureFailsWhenFilteredRecipientsCannotReachQuorum() throws Exception {
+        final NodeImpl node = new NodeImpl();
+        try {
+            final Status[] callbackStatus = new Status[1];
+            final ReadIndexResponse[] callbackResponse = new ReadIndexResponse[1];
+            final RpcResponseClosure<ReadIndexResponse> done = new RpcResponseClosureAdapter<ReadIndexResponse>() {
+
+                @Override
+                public void run(final Status status) {
+                    callbackStatus[0] = status;
+                    callbackResponse[0] = getResponse();
+                }
+            };
+
+            final RpcResponseClosureAdapter<AppendEntriesResponse> heartbeatDone = newReadIndexHeartbeatResponseClosure(
+                node, done, ReadIndexResponse.newBuilder().setIndex(11), 3, 2);
+
+            heartbeatDone.setResponse(AppendEntriesResponse.newBuilder().setTerm(1).setSuccess(true).build());
+            heartbeatDone.run(Status.OK());
+            assertNull(callbackStatus[0]);
+            assertNull(callbackResponse[0]);
+
+            heartbeatDone.setResponse(AppendEntriesResponse.newBuilder().setTerm(1).setSuccess(false).build());
+            heartbeatDone.run(Status.OK());
+            assertNotNull(callbackStatus[0]);
+            assertTrue(callbackStatus[0].isOk());
+            assertNotNull(callbackResponse[0]);
+            assertFalse(callbackResponse[0].getSuccess());
+            assertEquals(11, callbackResponse[0].getIndex());
+        } finally {
+            assertEquals(0, NodeImpl.GLOBAL_NUM_NODES.decrementAndGet());
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    private RpcResponseClosureAdapter<AppendEntriesResponse> newReadIndexHeartbeatResponseClosure(final NodeImpl node,
+                                                                                                  final RpcResponseClosure<ReadIndexResponse> done,
+                                                                                                  final ReadIndexResponse.Builder respBuilder,
+                                                                                                  final int quorum,
+                                                                                                  final int expectedFollowerResponses)
+                                                                                                                                      throws Exception {
+        for (final Class<?> innerClass : NodeImpl.class.getDeclaredClasses()) {
+            if ("ReadIndexHeartbeatResponseClosure".equals(innerClass.getSimpleName())) {
+                final Constructor<?> constructor = innerClass.getDeclaredConstructor(NodeImpl.class,
+                    RpcResponseClosure.class, ReadIndexResponse.Builder.class, int.class, int.class);
+                constructor.setAccessible(true);
+                return (RpcResponseClosureAdapter<AppendEntriesResponse>) constructor.newInstance(node, done,
+                    respBuilder, quorum, expectedFollowerResponses);
+            }
+        }
+        fail("ReadIndexHeartbeatResponseClosure not found");
+        return null;
+    }
+
     @Test
     public void testReadIndexTimeout() throws Exception {
         final List<PeerId> peers = TestUtils.generatePeers(3);