This is an automated email from the ASF dual-hosted git repository. Caideyipi pushed a commit to branch fix/pipe-cluster-replica-restart-barrier in repository https://gitbox.apache.org/repos/asf/iotdb.git
commit 7642e305e57b7e79d04afbf7ac1838eb6108a900 Author: Caideyipi <[email protected]> AuthorDate: Tue Oct 13 11:27:32 2026 +0800 Fix queued IoTConsensus writes during shutdown --- .../manual/enhanced/IoTDBPipeClusterIT.java | 38 +++++---- .../auto/enhanced/IoTDBPipeClusterIT.java | 38 +++++---- .../consensus/iot/IoTConsensusServerImpl.java | 8 +- .../consensus/iot/IoTConsensusServerImplTest.java | 94 ++++++++++++++++++++++ 4 files changed, 145 insertions(+), 33 deletions(-) diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java index a5e505f4fd7..20793825f7d 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java @@ -449,6 +449,8 @@ public class IoTDBPipeClusterIT extends AbstractPipeTableModelDualManualIT { private void waitForTableDataRegionReplicationComplete( final BaseEnv env, final List<String> databases) { + // IoTConsensus accepts writes on any replica, including a previous leader during failover. + // Check every replica's outgoing replication before flushing or restarting the cluster. await() .pollInterval(500, TimeUnit.MILLISECONDS) .atMost(2, TimeUnit.MINUTES) @@ -456,24 +458,23 @@ public class IoTDBPipeClusterIT extends AbstractPipeTableModelDualManualIT { () -> { try (final SyncConfigNodeIServiceClient client = (SyncConfigNodeIServiceClient) env.getLeaderConfigNodeConnection()) { - final List<TRegionInfo> leaderRegionInfoList = - showTableDataRegionLeaders(databases, client); + final List<TRegionInfo> regionInfoList = showTableDataRegions(databases, client); Assert.assertFalse( - "No table DataRegion leader found for databases " + databases, - leaderRegionInfoList.isEmpty()); + "No table DataRegion found for databases " + databases, + regionInfoList.isEmpty()); - for (final TRegionInfo regionInfo : leaderRegionInfoList) { - final DataNodeWrapper leaderNode = + for (final TRegionInfo regionInfo : regionInfoList) { + final DataNodeWrapper replicaNode = findDataNodeWrapperByPort(env, regionInfo.getClientRpcPort()); final String metricsUrl = "http://"; - + leaderNode.getIp() + + replicaNode.getIp() + ":" - + leaderNode.getMetricPort() + + replicaNode.getMetricPort() + "/metrics"; final String metricsContent = env.getUrlContent(metricsUrl, null); Assert.assertNotNull( - "Failed to fetch metrics from leader DataNode at " + metricsUrl, + "Failed to fetch metrics from replica DataNode at " + metricsUrl, metricsContent); assertSyncLagIsZero(metricsContent, buildDataRegionTag(regionInfo), metricsUrl); } @@ -483,6 +484,17 @@ public class IoTDBPipeClusterIT extends AbstractPipeTableModelDualManualIT { private List<TRegionInfo> showTableDataRegionLeaders( final List<String> databases, final SyncConfigNodeIServiceClient client) throws TException { + final List<TRegionInfo> result = new ArrayList<>(); + for (final TRegionInfo regionInfo : showTableDataRegions(databases, client)) { + if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) { + result.add(regionInfo); + } + } + return result; + } + + private List<TRegionInfo> showTableDataRegions( + final List<String> databases, final SyncConfigNodeIServiceClient client) throws TException { final TShowRegionResp showRegionResp = client.showRegion( new TShowRegionReq() @@ -491,13 +503,7 @@ public class IoTDBPipeClusterIT extends AbstractPipeTableModelDualManualIT { .setIsTableModel(true)); Assert.assertEquals( TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode()); - final List<TRegionInfo> result = new ArrayList<>(); - for (final TRegionInfo regionInfo : showRegionResp.getRegionInfoList()) { - if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) { - result.add(regionInfo); - } - } - return result; + return showRegionResp.getRegionInfoList(); } private DataNodeWrapper findDataNodeWrapperByPort(final BaseEnv env, final int port) { diff --git a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java index 529147a77b0..b64a43fc04d 100644 --- a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java @@ -458,6 +458,8 @@ public class IoTDBPipeClusterIT extends AbstractPipeDualTreeModelAutoIT { private void waitForTreeDataRegionReplicationComplete( final BaseEnv env, final List<String> databases) { + // IoTConsensus accepts writes on any replica, including a previous leader during failover. + // Check every replica's outgoing replication before flushing or restarting the cluster. await() .pollInterval(500, TimeUnit.MILLISECONDS) .atMost(2, TimeUnit.MINUTES) @@ -465,24 +467,23 @@ public class IoTDBPipeClusterIT extends AbstractPipeDualTreeModelAutoIT { () -> { try (final SyncConfigNodeIServiceClient client = (SyncConfigNodeIServiceClient) env.getLeaderConfigNodeConnection()) { - final List<TRegionInfo> leaderRegionInfoList = - showTreeDataRegionLeaders(databases, client); + final List<TRegionInfo> regionInfoList = showTreeDataRegions(databases, client); Assert.assertFalse( - "No tree DataRegion leader found for databases " + databases, - leaderRegionInfoList.isEmpty()); + "No tree DataRegion found for databases " + databases, + regionInfoList.isEmpty()); - for (final TRegionInfo regionInfo : leaderRegionInfoList) { - final DataNodeWrapper leaderNode = + for (final TRegionInfo regionInfo : regionInfoList) { + final DataNodeWrapper replicaNode = findDataNodeWrapperByPort(env, regionInfo.getClientRpcPort()); final String metricsUrl = "http://"; - + leaderNode.getIp() + + replicaNode.getIp() + ":" - + leaderNode.getMetricPort() + + replicaNode.getMetricPort() + "/metrics"; final String metricsContent = env.getUrlContent(metricsUrl, null); Assert.assertNotNull( - "Failed to fetch metrics from leader DataNode at " + metricsUrl, + "Failed to fetch metrics from replica DataNode at " + metricsUrl, metricsContent); assertSyncLagIsZero(metricsContent, buildDataRegionTag(regionInfo), metricsUrl); } @@ -492,6 +493,17 @@ public class IoTDBPipeClusterIT extends AbstractPipeDualTreeModelAutoIT { private List<TRegionInfo> showTreeDataRegionLeaders( final List<String> databases, final SyncConfigNodeIServiceClient client) throws TException { + final List<TRegionInfo> result = new ArrayList<>(); + for (final TRegionInfo regionInfo : showTreeDataRegions(databases, client)) { + if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) { + result.add(regionInfo); + } + } + return result; + } + + private List<TRegionInfo> showTreeDataRegions( + final List<String> databases, final SyncConfigNodeIServiceClient client) throws TException { final TShowRegionResp showRegionResp = client.showRegion( new TShowRegionReq() @@ -499,13 +511,7 @@ public class IoTDBPipeClusterIT extends AbstractPipeDualTreeModelAutoIT { .setDatabases(databases)); Assert.assertEquals( TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode()); - final List<TRegionInfo> result = new ArrayList<>(); - for (final TRegionInfo regionInfo : showRegionResp.getRegionInfoList()) { - if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) { - result.add(regionInfo); - } - } - return result; + return showRegionResp.getRegionInfoList(); } private DataNodeWrapper findDataNodeWrapperByPort(final BaseEnv env, final int port) { diff --git a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java index ba4ad87b1fd..b61a4979618 100644 --- a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java +++ b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java @@ -1583,7 +1583,13 @@ public class IoTConsensusServerImpl { for (int i = 0; i < insertNodes.size(); i++) { IConsensusRequest insertNode = insertNodes.get(i); insertNode.markAsGeneratedByRemoteConsensusLeader(); - TSStatus status = stateMachine.write(insertNode); + // The replica may become read-only while this batch waits for its turn. Recheck before + // applying each request so shutdown cannot acknowledge writes after the final flush. + TSStatus status = + isReadOnly() + ? RpcUtils.getStatus( + TSStatusCode.SYSTEM_READ_ONLY, ConsensusMessages.SYNC_LOG_SYSTEM_READ_ONLY) + : stateMachine.write(insertNode); subStatus.add(status); if (RetryUtils.needRetryForWrite(status.getCode())) { for (int j = i + 1; j < insertNodes.size(); j++) { diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java index cc33cf1573a..291a8d99d1b 100644 --- a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java +++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java @@ -20,15 +20,21 @@ package org.apache.iotdb.consensus.iot; import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.common.rpc.thrift.TSStatus; import org.apache.iotdb.commons.audit.UserDataTransferAuditHandler; import org.apache.iotdb.commons.consensus.ConsensusGroupId; import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType; import org.apache.iotdb.commons.request.IConsensusRequest; import org.apache.iotdb.consensus.common.Peer; +import org.apache.iotdb.consensus.common.request.DeserializedBatchIndexedConsensusRequest; +import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest; import org.apache.iotdb.consensus.config.IoTConsensusConfig; import org.apache.iotdb.consensus.config.UserDataTransferAuditClassifier; +import org.apache.iotdb.consensus.i18n.ConsensusMessages; +import org.apache.iotdb.consensus.iot.util.TestEntry; import org.apache.iotdb.consensus.iot.util.TestStateMachine; +import org.apache.iotdb.rpc.TSStatusCode; import org.junit.Rule; import org.junit.Test; @@ -46,6 +52,8 @@ import java.util.concurrent.Executors; import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; @@ -59,6 +67,92 @@ public class IoTConsensusServerImplTest { @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + @Test + public void testSyncLogBecomesReadOnlyWhileWaiting() throws Exception { + final ScheduledExecutorService backgroundTaskService = + Executors.newSingleThreadScheduledExecutor(); + final AtomicBoolean readOnly = new AtomicBoolean(); + final TestStateMachine stateMachine = + new TestStateMachine() { + @Override + public boolean isReadOnly() { + return readOnly.get(); + } + }; + final Peer thisNode = new Peer(new DataRegionId(1), 0, new TEndPoint("127.0.0.1", 6667)); + final IoTConsensusServerImpl server = + new IoTConsensusServerImpl( + temporaryFolder.getRoot().getAbsolutePath(), + null, + DirectoryStrategyType.SEQUENCE_STRATEGY, + thisNode, + Collections.singletonList(thisNode), + stateMachine, + backgroundTaskService, + null, + null, + IoTConsensusConfig.newBuilder() + .setReplication( + IoTConsensusConfig.Replication.newBuilder() + .setMaxWaitingTimeForWaitBatchInMs(100) + .build()) + .build()); + final DeserializedBatchIndexedConsensusRequest request = + new DeserializedBatchIndexedConsensusRequest(1, 2, 2, 1, 2); + for (int index = 1; index <= 2; index++) { + request.add( + new IndexedConsensusRequest( + index, index, Collections.singletonList(new TestEntry(index, thisNode)))); + } + final AtomicReference<TSStatus> status = new AtomicReference<>(); + final Thread syncLogThread = new Thread(() -> status.set(server.syncLog(1, request))); + try { + assertFalse(server.isReadOnly()); + syncLogThread.start(); + final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + while (syncLogThread.getState() != Thread.State.TIMED_WAITING + && System.nanoTime() < deadline) { + Thread.sleep(10); + } + assertEquals(Thread.State.TIMED_WAITING, syncLogThread.getState()); + readOnly.set(true); + syncLogThread.join(TimeUnit.SECONDS.toMillis(15)); + assertFalse(syncLogThread.isAlive()); + assertEquals(2, status.get().getSubStatusSize()); + assertEquals( + TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), + status.get().getSubStatus().get(0).getCode()); + assertEquals( + ConsensusMessages.SYNC_LOG_SYSTEM_READ_ONLY, + status.get().getSubStatus().get(0).getMessage()); + assertEquals( + TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode(), + status.get().getSubStatus().get(1).getCode()); + assertTrue(stateMachine.getRequestSet().isEmpty()); + assertEquals(0, server.getEffectiveSafePhysicalTime(1)); + + // The sender must be able to retry the complete batch after the replica becomes writable. + readOnly.set(false); + server.reloadConsensusConfig( + IoTConsensusConfig.newBuilder() + .setReplication( + IoTConsensusConfig.Replication.newBuilder() + .setMaxPendingBatchesNum(1) + .setMaxWaitingTimeForWaitBatchInMs(100) + .build()) + .build()); + final TSStatus retryStatus = server.syncLog(1, request); + assertTrue( + retryStatus.getSubStatus().stream() + .allMatch(s -> s.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode())); + assertEquals(2, stateMachine.getRequestSet().size()); + } finally { + syncLogThread.interrupt(); + syncLogThread.join(TimeUnit.SECONDS.toMillis(5)); + backgroundTaskService.shutdownNow(); + } + } + @Test public void testSnapshotTransferAppliesGroupClassifier() { final DataRegionId groupId = new DataRegionId(1);
