You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ta...@apache.org on 2022/05/26 15:29:24 UTC
[iotdb] branch master updated: [IOTDB-3297] leader interface change (#6024)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 8cc2bf6221 [IOTDB-3297] leader interface change (#6024)
8cc2bf6221 is described below
commit 8cc2bf6221daa8555de702a514d0d24b277972c3
Author: SzyWilliam <48...@users.noreply.github.com>
AuthorDate: Thu May 26 23:29:18 2022 +0800
[IOTDB-3297] leader interface change (#6024)
* leader interface change
* spotless
* add comments
---
.../iotdb/consensus/ratis/RatisConsensus.java | 33 ++++++++++++----------
.../iotdb/consensus/ratis/RatisConsensusTest.java | 10 ++++---
2 files changed, 24 insertions(+), 19 deletions(-)
diff --git a/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java b/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
index 03bca95863..b5a027d0ab 100644
--- a/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
+++ b/consensus/src/main/java/org/apache/iotdb/consensus/ratis/RatisConsensus.java
@@ -44,7 +44,6 @@ import org.apache.iotdb.consensus.exception.RatisRequestFailedException;
import org.apache.commons.pool2.KeyedObjectPool;
import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
-import org.apache.ratis.client.RaftClient;
import org.apache.ratis.client.RaftClientConfigKeys;
import org.apache.ratis.client.RaftClientRpc;
import org.apache.ratis.conf.Parameters;
@@ -58,6 +57,7 @@ import org.apache.ratis.protocol.RaftClientRequest;
import org.apache.ratis.protocol.RaftGroup;
import org.apache.ratis.protocol.RaftGroupId;
import org.apache.ratis.protocol.RaftPeer;
+import org.apache.ratis.protocol.RaftPeerId;
import org.apache.ratis.protocol.exceptions.NotLeaderException;
import org.apache.ratis.server.DivisionInfo;
import org.apache.ratis.server.RaftServer;
@@ -426,14 +426,18 @@ class RatisConsensus implements IConsensus {
}
/**
- * transferLeader in Ratis implementation is not guaranteed to transfer leadership to the
- * designated peer Thus, it may produce undetermined results. Caller should not count on this API.
+ * transferLeader is implemented by 1. modify peer priority 2. ask current leader to step down
+ *
+ * <p>1. call setConfiguration to upgrade newLeader's priority to 1 and degrade all follower peers
+ * to 0. By default, Ratis gives every Raft Peer same priority 0. Ratis does not allow a peer with
+ * priority <= currentLeader.priority to becomes the leader, so we have to upgrade leader's
+ * priority to 1
+ *
+ * <p>2. call transferLeadership to force current leader to step down and raise a new round of
+ * election. In this election, the newLeader peer with priority 1 is guaranteed to be elected.
*/
@Override
public ConsensusGenericResponse transferLeader(ConsensusGroupId groupId, Peer newLeader) {
- // By default, Ratis gives every Raft Peer same priority 0
- // Ratis does not allow a peer.priority <= currentLeader.priority to becomes the leader
- // So we have to enhance to leader's priority
// first fetch the newest information
@@ -468,7 +472,9 @@ class RatisConsensus implements IConsensus {
return failed(new RatisRequestFailedException(configChangeReply.getException()));
}
// TODO tuning for timeoutMs
- reply = client.getRaftClient().admin().transferLeadership(newRaftLeader.getId(), 5000);
+ // when newLeaderPeerId == null, ratis forces current leader to step down and raise new
+ // election
+ reply = client.getRaftClient().admin().transferLeadership(null, 5000);
if (!reply.isSuccess()) {
return failed(new RatisRequestFailedException(reply.getException()));
}
@@ -527,19 +533,16 @@ class RatisConsensus implements IConsensus {
@Override
public Peer getLeader(ConsensusGroupId groupId) {
- if (isLeader(groupId)) {
- return new Peer(groupId, Utils.formRaftPeerIdToTEndPoint(myself.getId()));
- }
-
RaftGroupId raftGroupId = Utils.fromConsensusGroupIdToRaftGroupId(groupId);
- RaftClient client;
+ RaftPeerId leaderId;
+
try {
- client = server.getDivision(raftGroupId).getRaftClient();
+ leaderId = server.getDivision(raftGroupId).getInfo().getLeaderId();
} catch (IOException e) {
- logger.warn("cannot find raft client for group " + groupId);
+ logger.warn("fetch division info for group " + groupId + " failed due to: ", e);
return null;
}
- TEndPoint leaderEndpoint = Utils.formRaftPeerIdToTEndPoint(client.getLeaderId());
+ TEndPoint leaderEndpoint = Utils.formRaftPeerIdToTEndPoint(leaderId);
return new Peer(groupId, leaderEndpoint);
}
diff --git a/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java b/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
index bf5b9277c8..839de7fe96 100644
--- a/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
+++ b/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RatisConsensusTest.java
@@ -121,13 +121,14 @@ public class RatisConsensusTest {
// 2. Do Consensus 10
doConsensus(servers.get(0), group.getGroupId(), 10, 10);
- int leader = getLeaderOrdinal();
+ // pick a random leader
+ int leader = 1;
int follower1 = (leader + 1) % 3;
int follower2 = (leader + 2) % 3;
// 3. Remove two Peers from Group (peer 0 and peer 2)
// transfer the leader to peer1
- // servers.get(0).transferLeader(gid, peers.get(1));
+ servers.get(follower1).transferLeader(gid, peers.get(leader));
// Assert.assertTrue(servers.get(1).isLeader(gid));
// first use removePeer to inform the group leader of configuration change
servers.get(leader).removePeer(gid, peers.get(follower1));
@@ -157,11 +158,12 @@ public class RatisConsensusTest {
// 6. try consensus with all 3 peers
doConsensus(servers.get(2), gid, 10, 30);
- leader = getLeaderOrdinal();
+ // pick a random leader
+ leader = 0;
follower1 = (leader + 1) % 3;
follower2 = (leader + 2) % 3;
// 7. again, group contains only peer0
- // servers.get(0).transferLeader(group.getGroupId(), peers.get(0));
+ servers.get(0).transferLeader(group.getGroupId(), peers.get(leader));
servers
.get(leader)
.changePeer(group.getGroupId(), Collections.singletonList(peers.get(leader)));