You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by as...@apache.org on 2021/12/21 15:14:53 UTC
[ignite-3] branch main updated: IGNITE-15271 Don't stop following on updated vote request - Fixes #268.
This is an automated email from the ASF dual-hosted git repository.
ascherbakov pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/ignite-3.git
The following commit(s) were added to refs/heads/main by this push:
new 74e2aaf IGNITE-15271 Don't stop following on updated vote request - Fixes #268.
74e2aaf is described below
commit 74e2aaf3cf47b4cbb32f66c9c4893cf0fe043377
Author: Alexey Scherbakov <al...@gmail.com>
AuthorDate: Tue Dec 21 18:14:29 2021 +0300
IGNITE-15271 Don't stop following on updated vote request - Fixes #268.
Signed-off-by: Alexey Scherbakov <al...@gmail.com>
---
.../apache/ignite/raft/jraft/core/ItNodeTest.java | 62 ++++++++++++++++++++++
.../apache/ignite/raft/jraft/core/NodeImpl.java | 9 +++-
.../apache/ignite/raft/jraft/rpc/RpcClientEx.java | 2 +
.../raft/jraft/rpc/impl/IgniteRpcClient.java | 24 +++++++++
.../apache/ignite/raft/jraft/util/Endpoint.java | 5 ++
.../apache/ignite/raft/jraft/test/TestUtils.java | 18 +++++++
6 files changed, 119 insertions(+), 1 deletion(-)
diff --git a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItNodeTest.java b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItNodeTest.java
index c335594..6c726d2 100644
--- a/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItNodeTest.java
+++ b/modules/raft/src/integrationTest/java/org/apache/ignite/raft/jraft/core/ItNodeTest.java
@@ -37,6 +37,7 @@ import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BooleanSupplier;
import java.util.stream.Stream;
import org.apache.ignite.internal.testframework.WorkDirectory;
@@ -2730,6 +2731,67 @@ public class ItNodeTest {
}
@Test
+ public void testLeaderPropagatedBeforeVote() throws Exception {
+ // start five nodes
+ List<PeerId> peers = TestUtils.generatePeers(3);
+
+ cluster = new TestCluster("unitest", dataPath, peers, 3_000, testInfo);
+
+ for (PeerId peer : peers) {
+ RaftOptions opts = new RaftOptions();
+ opts.setElectionHeartbeatFactor(4); // Election timeout divisor.
+ assertTrue(cluster.start(peer.getEndpoint(), false, 300, false, null, opts));
+ }
+
+ List<NodeImpl> nodes = cluster.getNodes();
+
+ AtomicReference<String> guard = new AtomicReference();
+
+ // Block only one vote message.
+ for (NodeImpl node : nodes) {
+ RpcClientEx rpcClientEx = TestUtils.sender(node);
+ rpcClientEx.recordMessages((msg, nodeId) -> true);
+ rpcClientEx.blockMessages((msg, nodeId) -> {
+ if (msg instanceof RpcRequests.RequestVoteRequest) {
+ RpcRequests.RequestVoteRequest msg0 = (RpcRequests.RequestVoteRequest)msg;
+
+ if (msg0.preVote())
+ return false;
+
+ if (guard.compareAndSet(null, nodeId))
+ return true;
+ }
+
+ if (msg instanceof RpcRequests.AppendEntriesRequest && nodeId.equals(guard.get())) {
+ RpcRequests.AppendEntriesRequest tmp = (RpcRequests.AppendEntriesRequest) msg;
+
+ if (tmp.entriesList() != null && !tmp.entriesList().isEmpty()) {
+ return true;
+ }
+ }
+
+ return false;
+ });
+ }
+
+ cluster.waitLeader();
+ Node leader = cluster.getLeader();
+ cluster.ensureLeader(leader);
+
+ RpcClientEx client = TestUtils.sender(leader);
+
+ client.stopBlock(1); // Unblock vote message.
+
+ // The follower shouldn't stop following on receiving stale vote request.
+ Node follower = cluster.getNode(new Endpoint(NetworkAddress.from(guard.get())));
+
+ boolean res =
+ waitForCondition(() -> ((MockStateMachine) follower.getOptions().getFsm()).getOnStopFollowingTimes() != 0, 1_000);
+
+ assertFalse(res, "The follower shouldn't stop following");
+ }
+
+ @Test
@Disabled("https://issues.apache.org/jira/browse/IGNITE-15202")
public void readCommittedUserLog() throws Exception {
// setup cluster
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/NodeImpl.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/NodeImpl.java
index 0eff1cf..82021be 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/NodeImpl.java
+++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/core/NodeImpl.java
@@ -1833,11 +1833,18 @@ public class NodeImpl implements Node, RaftServerService {
stepDown(request.term(), false, new Status(RaftError.EHIGHERTERMRESPONSE,
"Raft node receives higher term RequestVoteRequest."));
}
+ else if (candidateId.equals(leaderId)) { // Already follows a leader in this term.
+ LOG.info("Node {} ignores RequestVoteRequest from {}, term={}, currTerm={}.", getNodeId(),
+ request.serverId(), request.term(), this.currTerm);
+
+ break;
+ }
}
else {
// ignore older term
- LOG.info("Node {} ignore RequestVoteRequest from {}, term={}, currTerm={}.", getNodeId(),
+ LOG.info("Node {} ignores RequestVoteRequest from {}, term={}, currTerm={}.", getNodeId(),
request.serverId(), request.term(), this.currTerm);
+
break;
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/RpcClientEx.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/RpcClientEx.java
index f4ee4e0..60a58a5 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/RpcClientEx.java
+++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/RpcClientEx.java
@@ -25,6 +25,8 @@ public interface RpcClientEx extends RpcClient {
void stopBlock();
+ void stopBlock(int cnt);
+
void recordMessages(BiPredicate<Object, String> predicate);
void stopRecord();
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/IgniteRpcClient.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/IgniteRpcClient.java
index 1562188..8ef2fbb 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/IgniteRpcClient.java
+++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/rpc/impl/IgniteRpcClient.java
@@ -179,6 +179,30 @@ public class IgniteRpcClient implements RpcClientEx {
}
/** {@inheritDoc} */
+ @Override public void stopBlock(int cnt) {
+ ArrayList<Object[]> msgs = new ArrayList<>();
+
+ synchronized (this) {
+ while(cnt-- > 0) {
+ Object[] tmp = blockedMsgs.poll();
+
+ if (tmp == null)
+ break;
+
+ msgs.add(tmp);
+ }
+
+ blockPred = null;
+ }
+
+ for (Object[] msg : msgs) {
+ Runnable r = (Runnable) msg[4];
+
+ r.run();
+ }
+ }
+
+ /** {@inheritDoc} */
@Override public void recordMessages(BiPredicate<Object, String> predicate) {
this.recordPred = predicate;
}
diff --git a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/util/Endpoint.java b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/util/Endpoint.java
index 71d886d..05e12db 100644
--- a/modules/raft/src/main/java/org/apache/ignite/raft/jraft/util/Endpoint.java
+++ b/modules/raft/src/main/java/org/apache/ignite/raft/jraft/util/Endpoint.java
@@ -17,6 +17,7 @@
package org.apache.ignite.raft.jraft.util;
import java.io.Serializable;
+import org.apache.ignite.network.NetworkAddress;
/**
* A IP address with port.
@@ -33,6 +34,10 @@ public class Endpoint implements Copiable<Endpoint>, Serializable {
super();
}
+ public Endpoint(NetworkAddress addr) {
+ this(addr.host(), addr.port());
+ }
+
public Endpoint(String address, int port) {
super();
this.ip = address;
diff --git a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/test/TestUtils.java b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/test/TestUtils.java
index c4dbcf2..006304a 100644
--- a/modules/raft/src/test/java/org/apache/ignite/raft/jraft/test/TestUtils.java
+++ b/modules/raft/src/test/java/org/apache/ignite/raft/jraft/test/TestUtils.java
@@ -29,13 +29,17 @@ import java.util.function.BooleanSupplier;
import org.apache.ignite.lang.IgniteInternalException;
import org.apache.ignite.network.ClusterService;
import org.apache.ignite.raft.jraft.JRaftUtils;
+import org.apache.ignite.raft.jraft.Node;
import org.apache.ignite.raft.jraft.RaftMessagesFactory;
import org.apache.ignite.raft.jraft.conf.ConfigurationEntry;
+import org.apache.ignite.raft.jraft.core.NodeImpl;
import org.apache.ignite.raft.jraft.entity.EnumOutter;
import org.apache.ignite.raft.jraft.entity.LogEntry;
import org.apache.ignite.raft.jraft.entity.LogId;
import org.apache.ignite.raft.jraft.entity.PeerId;
+import org.apache.ignite.raft.jraft.rpc.RpcClientEx;
import org.apache.ignite.raft.jraft.rpc.RpcRequests;
+import org.apache.ignite.raft.jraft.rpc.impl.core.DefaultRaftClientService;
import org.apache.ignite.raft.jraft.util.Endpoint;
import org.mockito.ArgumentCaptor;
@@ -196,4 +200,18 @@ public class TestUtils {
Thread.getAllStackTraces().keySet().stream().filter(t -> t.getName().contains("JRaft")).
sorted(comparing(Thread::getName)).collect(toList()).toString());
}
+
+ /**
+ * Returns a message sender for this node.
+ *
+ * @param node The node.
+ * @return The message sender.
+ */
+ public static RpcClientEx sender(Node node) {
+ NodeImpl node0 = (NodeImpl) node;
+
+ DefaultRaftClientService rpcService = (DefaultRaftClientService) node0.getRpcClientService();
+
+ return (RpcClientEx) rpcService.getRpcClient();
+ }
}