slfan1989 commented on code in PR #1571:
URL: https://github.com/apache/ratis/pull/1571#discussion_r3888611301
##########
ratis-server/src/test/java/org/apache/ratis/server/impl/RaftReconfigurationBaseTest.java:
##########
@@ -507,86 +513,119 @@ void runTestBootstrapReconf(int numNewPeer, boolean
startNewPeer, CLUSTER cluste
* retrying.
*/
@Test
- @Flaky("RATIS-2251")
+ @Timeout(120)
public void testKillLeaderDuringReconf() throws Exception {
// originally 3 peers
runWithNewCluster(3, this::runTestKillLeaderDuringReconf);
}
+ /**
+ * Checks whether the server is bootstrapping the given peer.
+ *
+ * @param server the server division to check
+ * @param peerId the ID of the peer to check
+ * @return true if the peer is being bootstrapped in the leader staging
state; otherwise, false
+ */
+ private static boolean isBootstrappingPeer(RaftServer.Division server,
RaftPeerId peerId) {
+ return ((RaftServerImpl) server).getRole().getLeaderState()
+ .filter(LeaderStateImpl::inStagingState)
+ .map(state -> state.isBootStrappingPeer(peerId))
+ .orElse(false);
+ }
+
void runTestKillLeaderDuringReconf(CLUSTER cluster) throws Exception {
- final AtomicBoolean clientRunning = new AtomicBoolean(true);
- Thread clientThread = null;
+ final ExecutorService executor = ConcurrentUtils.newSingleThreadExecutor(
+ JavaUtils.getClassSimpleName(getClass()) +
"-testKillLeaderDuringReconf-client");
try {
final RaftPeerId leaderId = RaftTestUtil.waitForLeader(cluster).getId();
- PeerChanges c1 = cluster.addNewPeers(1, false);
- PeerChanges c2 = cluster.removePeers(1, false, c1.getAddedPeers());
+ final PeerChanges c1 = cluster.addNewPeers(1, false);
+ final PeerChanges c2 = cluster.removePeers(1, false, c1.getAddedPeers());
+ final RaftPeerId newPeerId = c1.getAddedPeers().get(0).getId();
LOG.info("Start setConf: {}", c2.getPeersInNewConf());
LOG.info(cluster.printServers());
- final CompletableFuture<Void> setConf = new CompletableFuture<>();
- clientThread = new Thread(() -> {
- try(final RaftClient client = cluster.createClient(leaderId)) {
- for(int i = 0; clientRunning.get() && !setConf.isDone(); i++) {
- final RaftClientReply reply =
client.admin().setConfiguration(c2.getPeersInNewConf());
- if (reply.isSuccess()) {
- setConf.complete(null);
- return;
- }
- LOG.info("setConf attempt #{} failed, {}", i,
cluster.printServers());
- }
- } catch(Exception e) {
- LOG.error("Failed to setConf", e);
- setConf.completeExceptionally(e);
+ final Future<RaftClientReply> setConfTask = executor.submit(() -> {
+ try (RaftClient client = cluster.createClient(leaderId)) {
+ return client.admin().setConfiguration(c2.getPeersInNewConf());
}
});
- clientThread.start();
-
- // the leader cannot generate the (old, new) conf, and it will keep
- // bootstrapping the 1 new peer since it has not started yet.
-
Assertions.assertFalse(((RaftConfigurationImpl)cluster.getLeader().getRaftConf()).isTransitional());
- // (0) the first conf entry, (1) the 1st setConf entry, (2) a metadata
entry
- // (3) new current conf entry (4) a metadata entry
- {
- final RaftLog leaderLog = cluster.getLeader().getRaftLog();
- for(LogEntryProto e : RaftTestUtil.getLogEntryProtos(leaderLog)) {
- LOG.info("{}", LogProtoUtils.toLogEntryString(e));
+ try {
+ JavaUtils.attempt(() -> {
+ assertFalse(setConfTask.isDone(),
+ () -> "setConfiguration completed before the leader was killed;
" + cluster.printServers());
+ assertTrue(isBootstrappingPeer(cluster.getDivision(leaderId),
newPeerId),
+ () -> "Leader " + leaderId + " is not bootstrapping peer " +
newPeerId
+ + "; " + cluster.printServers());
+ }, 10, cluster.getTimeoutMax(), "wait for the original leader to
bootstrap " + newPeerId, LOG);
+
+ // The leader cannot generate the (old, new) conf, and it will keep
+ // bootstrapping the new peer since it has not started yet.
+
assertFalse(((RaftConfigurationImpl)cluster.getLeader().getRaftConf()).isTransitional());
+
+ // (0) the first conf entry, (1) the 1st setConf entry, (2) a metadata
entry
+ // (3) new current conf entry (4) a metadata entry
+ {
+ final RaftLog leaderLog = cluster.getLeader().getRaftLog();
+ for(LogEntryProto e : RaftTestUtil.getLogEntryProtos(leaderLog)) {
+ LOG.info("{}", LogProtoUtils.toLogEntryString(e));
+ }
+ final long commitIndex = leaderLog.getLastCommittedIndex();
+ assertTrue(commitIndex <= 2,
+ () -> "commitIndex = " + commitIndex + " > 2; " +
cluster.printServers());
}
- final long commitIndex = leaderLog.getLastCommittedIndex();
- Assertions.assertTrue(commitIndex <= 2, "commitIndex = " + commitIndex
+ " > 2");
- }
-
- final RaftPeerId killed = RaftTestUtil.waitAndKillLeader(cluster);
- Assertions.assertEquals(leaderId, killed);
- final RaftPeerId newLeaderId =
RaftTestUtil.waitForLeader(cluster).getId();
- LOG.info("newLeaderId: {}", newLeaderId);
- TimeDuration.valueOf(1500, TimeUnit.MILLISECONDS).sleep();
Review Comment:
A fixed `1500 ms` sleep does not guarantee that the new leader has taken
over the reconfiguration and started bootstrapping the new peer.
Replace it with a state-based wait that confirms the new leader recognizes
`newPeerId` as a bootstrapping peer before the peer is started.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
