This is an automated email from the ASF dual-hosted git repository.
jt2594838 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 f8135bb7284 [Test] Fix Ratis mini-cluster shutdown race (#18879)
f8135bb7284 is described below
commit f8135bb728497934771f708ce4dac21572137ae7
Author: Caideyipi <[email protected]>
AuthorDate: Thu Oct 15 12:24:35 2026 +0800
[Test] Fix Ratis mini-cluster shutdown race (#18879)
---
.../iotdb/consensus/ratis/RecoverReadTest.java | 70 ++++++++++++++++++++++
.../apache/iotdb/consensus/ratis/TestUtils.java | 51 +++++++++++++++-
2 files changed, 119 insertions(+), 2 deletions(-)
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RecoverReadTest.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RecoverReadTest.java
index 017118594ba..f35c7cb6642 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RecoverReadTest.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/RecoverReadTest.java
@@ -24,7 +24,9 @@ import org.apache.iotdb.commons.request.IConsensusRequest;
import org.apache.iotdb.consensus.common.Peer;
import org.apache.iotdb.consensus.config.RatisConfig;
import org.apache.iotdb.consensus.exception.RatisReadUnavailableException;
+import org.apache.iotdb.consensus.ratis.utils.Utils;
+import org.apache.ratis.server.RaftServer;
import org.apache.ratis.util.TimeDuration;
import org.junit.After;
import org.junit.Assert;
@@ -35,6 +37,10 @@ import org.slf4j.LoggerFactory;
import java.io.File;
import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
@@ -46,6 +52,7 @@ public class RecoverReadTest {
private static class SlowRecoverStateMachine extends
TestUtils.IntegerCounter {
private final TimeDuration stallDuration;
private final AtomicBoolean stallApply = new AtomicBoolean(false);
+ private final CountDownLatch stopped = new CountDownLatch(1);
private SlowRecoverStateMachine(TimeDuration stallDuration) {
this.stallDuration = stallDuration;
@@ -70,6 +77,11 @@ public class RecoverReadTest {
return super.write(request);
}
+ @Override
+ public void stop() {
+ stopped.countDown();
+ }
+
@Override
public boolean takeSnapshot(File snapshotDir) {
// allow no snapshot to ensure the raft log be replayed
@@ -126,6 +138,64 @@ public class RecoverReadTest {
}
}
+ @Test
+ public void stopWaitsForInFlightLeaderCheck() throws Exception {
+ final ConsensusGroupId gid = miniCluster.getGid();
+ for (RatisConsensus server : miniCluster.getServers()) {
+ server.createLocalPeer(gid, miniCluster.getPeers());
+ }
+ miniCluster.waitUntilActiveLeaderElectedAndReady();
+
+ int leaderIndex = 0;
+ while (!miniCluster.getServer(leaderIndex).isLeader(gid)) {
+ leaderIndex++;
+ }
+ final RaftServer.Division leader =
+ miniCluster
+ .getServer(leaderIndex)
+ .getServer()
+ .getDivision(Utils.fromConsensusGroupIdToRaftGroupId(gid));
+ final SlowRecoverStateMachine stateMachine =
+ (SlowRecoverStateMachine)
miniCluster.getStateMachines().get(leaderIndex);
+ final CountDownLatch checkStarted = new CountDownLatch(1);
+ final CountDownLatch finishCheck = new CountDownLatch(1);
+ final CountDownLatch stopStarted = new CountDownLatch(1);
+ final ExecutorService executor = Executors.newFixedThreadPool(2);
+ try {
+ final Future<?> check =
+ executor.submit(
+ () -> {
+ // LeaderStateImpl performs its leadership checks under this
monitor.
+ synchronized (leader) {
+ checkStarted.countDown();
+ Assert.assertTrue(finishCheck.await(30, TimeUnit.SECONDS));
+ leader.getRaftLog().getLastEntryTermIndex();
+ }
+ return null;
+ });
+ Assert.assertTrue(checkStarted.await(30, TimeUnit.SECONDS));
+ final Future<?> stop =
+ executor.submit(
+ () -> {
+ stopStarted.countDown();
+ miniCluster.stop();
+ return null;
+ });
+ Assert.assertTrue(stopStarted.await(30, TimeUnit.SECONDS));
+ Assert.assertFalse(
+ "The state machine must remain open until the in-flight leader check
finishes",
+ stateMachine.stopped.await(5, TimeUnit.SECONDS));
+ finishCheck.countDown();
+ check.get(30, TimeUnit.SECONDS);
+ stop.get(30, TimeUnit.SECONDS);
+ Assert.assertEquals(0, stateMachine.stopped.getCount());
+ } finally {
+ finishCheck.countDown();
+ executor.shutdown();
+ Assert.assertTrue(executor.awaitTermination(30, TimeUnit.SECONDS));
+ }
+ }
+
/* mimics the situation before this patch */
@Test
public void inconsistentReadAfterRestart() throws Exception {
diff --git
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java
index 3217d1cc58e..093da2064e2 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/ratis/TestUtils.java
@@ -39,9 +39,16 @@ import
org.apache.iotdb.consensus.exception.RatisReadUnavailableException;
import org.apache.iotdb.consensus.ratis.utils.Retriable;
import org.apache.iotdb.consensus.ratis.utils.Utils;
+import org.apache.ratis.protocol.ClientId;
+import org.apache.ratis.protocol.LeaderElectionManagementRequest;
+import org.apache.ratis.protocol.RaftClientReply;
+import org.apache.ratis.protocol.RaftGroupId;
+import org.apache.ratis.protocol.TransferLeadershipRequest;
+import org.apache.ratis.server.RaftServer;
import org.apache.ratis.thirdparty.com.google.common.base.Preconditions;
import org.apache.ratis.util.ExitUtils;
import org.apache.ratis.util.FileUtils;
+import org.apache.ratis.util.LifeCycle;
import org.apache.ratis.util.TimeDuration;
import org.apache.ratis.util.Timestamp;
import org.junit.Assert;
@@ -307,13 +314,53 @@ public class TestUtils {
}
void stop() throws IOException {
- for (RatisConsensus server : servers) {
- server.stop();
+ if (isStopped.get()) {
+ return;
+ }
+ try {
+ stepDownBeforeStopping();
+ } finally {
+ for (RatisConsensus server : servers) {
+ server.stop();
+ }
}
waitUntilPortsReleased(peerPorts);
isStopped.set(true);
}
+ private void stepDownBeforeStopping() throws IOException {
+ final ClientId clientId = ClientId.randomId();
+ final List<RaftServer> runningServers = new ArrayList<>();
+ // Ratis 3.3.1 closes the log without joining the leader event
processor. Step down through
+ // that processor first, so an in-flight leadership check cannot access
an already closed log.
+ // Pause elections on every peer before stepping down to prevent a
replacement leader.
+ for (RatisConsensus server : servers) {
+ final RaftServer raftServer = server.getServer();
+ if (raftServer.getLifeCycleState() != LifeCycle.State.RUNNING) {
+ continue;
+ }
+ runningServers.add(raftServer);
+ for (RaftGroupId groupId : raftServer.getGroupIds()) {
+ final RaftClientReply reply =
+ raftServer.leaderElectionManagement(
+ LeaderElectionManagementRequest.newPause(
+ clientId, raftServer.getId(), groupId, 0));
+ Assert.assertTrue(reply.toString(), reply.isSuccess());
+ }
+ }
+ for (RaftServer raftServer : runningServers) {
+ for (RaftGroupId groupId : raftServer.getGroupIds()) {
+ if (raftServer.getDivision(groupId).getInfo().isLeader()) {
+ final RaftClientReply reply =
+ raftServer.transferLeadership(
+ new TransferLeadershipRequest(
+ clientId, raftServer.getId(), groupId, 1, null,
10_000));
+ Assert.assertTrue(reply.toString(), reply.isSuccess());
+ }
+ }
+ }
+ }
+
void cleanUp() throws IOException {
stop();
for (File storage : peerStorage) {
