From 7642e305e57b7e79d04afbf7ac1838eb6108a900 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Tue, 13 Oct 2026 11:27:32 +0800 Subject: [PATCH 1/4] Fix queued IoTConsensus writes during shutdown --- .../manual/enhanced/IoTDBPipeClusterIT.java | 38 ++++---- .../auto/enhanced/IoTDBPipeClusterIT.java | 38 ++++---- .../consensus/iot/IoTConsensusServerImpl.java | 8 +- .../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 a5e505f4fd75e..20793825f7d24 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 @@ private void flushTableDataRegionReplicasAfterReplicationComplete( private void waitForTableDataRegionReplicationComplete( final BaseEnv env, final List 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 @@ private void waitForTableDataRegionReplicationComplete( () -> { try (final SyncConfigNodeIServiceClient client = (SyncConfigNodeIServiceClient) env.getLeaderConfigNodeConnection()) { - final List leaderRegionInfoList = - showTableDataRegionLeaders(databases, client); + final List 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 @@ private void waitForTableDataRegionReplicationComplete( private List showTableDataRegionLeaders( final List databases, final SyncConfigNodeIServiceClient client) throws TException { + final List result = new ArrayList<>(); + for (final TRegionInfo regionInfo : showTableDataRegions(databases, client)) { + if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) { + result.add(regionInfo); + } + } + return result; + } + + private List showTableDataRegions( + final List databases, final SyncConfigNodeIServiceClient client) throws TException { final TShowRegionResp showRegionResp = client.showRegion( new TShowRegionReq() @@ -491,13 +503,7 @@ private List showTableDataRegionLeaders( .setIsTableModel(true)); Assert.assertEquals( TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode()); - final List 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 529147a77b084..b64a43fc04d6b 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 @@ private void flushTreeDataRegionReplicasAfterReplicationComplete( private void waitForTreeDataRegionReplicationComplete( final BaseEnv env, final List 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 @@ private void waitForTreeDataRegionReplicationComplete( () -> { try (final SyncConfigNodeIServiceClient client = (SyncConfigNodeIServiceClient) env.getLeaderConfigNodeConnection()) { - final List leaderRegionInfoList = - showTreeDataRegionLeaders(databases, client); + final List 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 @@ private void waitForTreeDataRegionReplicationComplete( private List showTreeDataRegionLeaders( final List databases, final SyncConfigNodeIServiceClient client) throws TException { + final List result = new ArrayList<>(); + for (final TRegionInfo regionInfo : showTreeDataRegions(databases, client)) { + if (RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) { + result.add(regionInfo); + } + } + return result; + } + + private List showTreeDataRegions( + final List databases, final SyncConfigNodeIServiceClient client) throws TException { final TShowRegionResp showRegionResp = client.showRegion( new TShowRegionReq() @@ -499,13 +511,7 @@ private List showTreeDataRegionLeaders( .setDatabases(databases)); Assert.assertEquals( TSStatusCode.SUCCESS_STATUS.getStatusCode(), showRegionResp.getStatus().getCode()); - final List 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 ba4ad87b1fdf9..b61a49796187a 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 @@ private TSStatus cacheAndInsertLatestNode(DeserializedBatchIndexedConsensusReque 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 cc33cf1573a4c..291a8d99d1b2e 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.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 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); From a3b7dbd84321fb1f0805fa06029c84e01faf8039 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 14 Oct 2026 14:17:11 +0800 Subject: [PATCH 2/4] Test queued IoTConsensus writes across storage shutdown --- .../dataregion/IoTConsensusShutdownTest.java | 317 ++++++++++++++++++ 1 file changed, 317 insertions(+) create mode 100644 iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java new file mode 100644 index 0000000000000..5cdd3f78b46ae --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java @@ -0,0 +1,317 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ + +package org.apache.iotdb.db.consensus.statemachine.dataregion; + +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.commons.concurrent.ExceptionalCountDownLatch; +import org.apache.iotdb.commons.conf.CommonConfig; +import org.apache.iotdb.commons.conf.CommonDescriptor; +import org.apache.iotdb.commons.consensus.DataRegionId; +import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType; +import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; +import org.apache.iotdb.commons.request.IConsensusRequest; +import org.apache.iotdb.consensus.ConsensusFactory; +import org.apache.iotdb.consensus.common.Peer; +import org.apache.iotdb.consensus.common.request.DeserializedBatchIndexedConsensusRequest; +import org.apache.iotdb.consensus.config.IoTConsensusConfig; +import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl; +import org.apache.iotdb.db.conf.IoTDBConfig; +import org.apache.iotdb.db.conf.IoTDBDescriptor; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; +import org.apache.iotdb.db.storageengine.StorageEngine; +import org.apache.iotdb.db.storageengine.dataregion.DataRegion; +import org.apache.iotdb.db.storageengine.dataregion.compaction.schedule.CompactionTaskManager; +import org.apache.iotdb.db.storageengine.dataregion.flush.TsFileFlushPolicy; +import org.apache.iotdb.db.storageengine.dataregion.memtable.TsFileProcessor; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager; +import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALBuffer; +import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode; +import org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRecoverManager; +import org.apache.iotdb.db.storageengine.dataregion.wal.recover.file.UnsealedTsFileRecoverPerformer; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode; +import org.apache.iotdb.db.storageengine.dataregion.wal.utils.listener.WALRecoverListener; +import org.apache.iotdb.db.utils.EnvironmentUtils; +import org.apache.iotdb.db.utils.constant.TestConstant; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.apache.tsfile.enums.TSDataType; +import org.apache.tsfile.read.TsFileReader; +import org.apache.tsfile.read.TsFileSequenceReader; +import org.apache.tsfile.read.common.Path; +import org.apache.tsfile.read.common.RowRecord; +import org.apache.tsfile.read.expression.QueryExpression; +import org.apache.tsfile.read.query.dataset.QueryDataSet; +import org.apache.tsfile.write.schema.MeasurementSchema; +import org.awaitility.Awaitility; +import org.junit.After; +import org.junit.Before; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.TemporaryFolder; + +import java.io.File; +import java.lang.reflect.Field; +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutorService; +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.AtomicReference; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class IoTConsensusShutdownTest { + + private static final String DATABASE = "root.shutdown"; + private static final String DEVICE = DATABASE + ".d1"; + private static final String MEASUREMENT = "s1"; + + @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder(); + + private final IoTDBConfig config = IoTDBDescriptor.getInstance().getConfig(); + private final CommonConfig commonConfig = CommonDescriptor.getInstance().getConfig(); + private WALMode previousWalMode; + private int previousDataNodeId; + private String previousConsensusProtocol; + private NodeStatus previousNodeStatus; + private String previousStatusReason; + private DataRegion dataRegion; + private IoTConsensusDataRegionStateMachine stateMachine; + + @Before + public void setUp() throws Exception { + previousWalMode = config.getWalMode(); + previousDataNodeId = config.getDataNodeId(); + previousConsensusProtocol = config.getDataRegionConsensusProtocolClass(); + previousNodeStatus = commonConfig.getNodeStatus(); + previousStatusReason = commonConfig.getStatusReason(); + config.setWalMode(WALMode.ASYNC); + config.setDataNodeId(0); + config.setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS); + commonConfig.setNodeStatus(NodeStatus.Running); + EnvironmentUtils.envSetUp(); + dataRegion = + new DataRegion( + TestConstant.OUTPUT_DATA_DIR + "shutdown-info", + "0", + new TsFileFlushPolicy.DirectFlushPolicy(), + DATABASE); + StorageEngine.getInstance().setDataRegion(new DataRegionId(0), dataRegion); + stateMachine = new IoTConsensusDataRegionStateMachine(dataRegion); + } + + @After + public void tearDown() throws Exception { + try { + commonConfig.setNodeStatus(NodeStatus.Running); + if (dataRegion != null) { + dataRegion.syncCloseAllWorkingTsFileProcessors(); + } + CompactionTaskManager.getInstance().stop(); + EnvironmentUtils.cleanEnv(); + } finally { + config.setWalMode(previousWalMode); + config.setDataNodeId(previousDataNodeId); + config.setDataRegionConsensusProtocolClass(previousConsensusProtocol); + commonConfig.setNodeStatus(NodeStatus.Running); + commonConfig.setNodeStatusWithReason(previousNodeStatus, previousStatusReason); + } + } + + @Test + public void testQueuedSyncLogDoesNotReopenStorageAfterShutdown() throws Exception { + writeAndAssertSuccess(1); + final ScheduledExecutorService backgroundTaskService = + Executors.newSingleThreadScheduledExecutor(); + final Peer thisNode = new Peer(new DataRegionId(0), 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() + .setMaxPendingBatchesNum(2) + .setMaxWaitingTimeForWaitBatchInMs(TimeUnit.MINUTES.toMillis(5)) + .build()) + .build()); + final DeserializedBatchIndexedConsensusRequest request = + new DeserializedBatchIndexedConsensusRequest(1, 2, 2, 1, 2); + request.add(insertRow(2)); + request.add(insertRow(3)); + final AtomicReference status = new AtomicReference<>(); + final Thread syncLogThread = new Thread(() -> status.set(server.syncLog(1, request))); + try { + assertFalse(server.isReadOnly()); + syncLogThread.start(); + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .until(() -> syncLogThread.getState() == Thread.State.TIMED_WAITING); + + // Follow the production order: read-only, final flush, storage stop, then WAL stop. + commonConfig.setNodeStatus(NodeStatus.ReadOnly); + WALManager.getInstance().waitAllWALFlushed(); + StorageEngine.getInstance().syncCloseAllProcessor(); + final File flushedFile = dataRegion.getSequenceFileList().get(0).getTsFile(); + StorageEngine.getInstance().stop(); + WALManager.getInstance().stop(); + assertEquals(0, WALManager.getInstance().getWALNodesNum()); + assertTrue(syncLogThread.isAlive()); + + // An empty preceding batch wakes the ordered request without another storage write. + server.syncLog(1, new DeserializedBatchIndexedConsensusRequest(0, 0, 0, 1, 0)); + syncLogThread.join(TimeUnit.SECONDS.toMillis(10)); + assertFalse(syncLogThread.isAlive()); + assertEquals(2, status.get().getSubStatusSize()); + assertEquals( + status.get() + + ", WAL nodes: " + + WALManager.getInstance().getWALNodesNum() + + ", working TsFile processors: " + + dataRegion.getWorkSequenceTsFileProcessors().size(), + TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), + status.get().getSubStatus().get(0).getCode()); + assertEquals( + TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode(), + status.get().getSubStatus().get(1).getCode()); + assertTrue(dataRegion.getWorkSequenceTsFileProcessors().isEmpty()); + assertTrue(dataRegion.getWorkUnsequenceTsFileProcessors().isEmpty()); + assertEquals(0, WALManager.getInstance().getWALNodesNum()); + assertEquals(0, server.getEffectiveSafePhysicalTime(1)); + assertSingleRow(flushedFile, 1); + } finally { + syncLogThread.interrupt(); + syncLogThread.join(TimeUnit.SECONDS.toMillis(10)); + backgroundTaskService.shutdownNow(); + } + } + + @Test + public void testAsyncWriteAfterTsFileFlushIsRecoveredWhenWalCloses() throws Exception { + writeAndAssertSuccess(1); + commonConfig.setNodeStatus(NodeStatus.ReadOnly); + WALManager.getInstance().waitAllWALFlushed(); + StorageEngine.getInstance().syncCloseAllProcessor(); + assertTrue(dataRegion.getWorkSequenceTsFileProcessors().isEmpty()); + final File flushedFile = dataRegion.getSequenceFileList().get(0).getTsFile(); + final WALBuffer buffer = ((WALNode) dataRegion.getWALNode().get()).getWALBuffer(); + + // Hold the real WAL sync executor so the async ACK cannot imply that the WAL is durable. + final Field syncBufferThreadField = WALBuffer.class.getDeclaredField("syncBufferThread"); + syncBufferThreadField.setAccessible(true); + final ExecutorService syncExecutor = (ExecutorService) syncBufferThreadField.get(buffer); + final CountDownLatch syncBlocked = new CountDownLatch(1); + final CountDownLatch resumeSync = new CountDownLatch(1); + final Future blocker = + syncExecutor.submit( + () -> { + syncBlocked.countDown(); + assertTrue(resumeSync.await(30, TimeUnit.SECONDS)); + return null; + }); + final File lateFile; + try { + assertTrue(syncBlocked.await(10, TimeUnit.SECONDS)); + // Call the real state machine directly to model the old unchecked SyncLog apply. + writeAndAssertSuccess(2); + assertFalse(buffer.isAllWALEntriesConsumed()); + assertEquals(1, dataRegion.getWorkSequenceTsFileProcessors().size()); + final TsFileProcessor processor = + dataRegion.getWorkSequenceTsFileProcessors().iterator().next(); + assertEquals(1, processor.getWorkMemTable().getTotalPointsNum()); + lateFile = processor.getTsFileResource().getTsFile(); + } finally { + resumeSync.countDown(); + blocker.get(10, TimeUnit.SECONDS); + } + + // Normal WAL close drains the late write, even though there is no further TsFile flush. + WALManager.getInstance().stop(); + dataRegion.forceCloseAllWorkingTsFileProcessors(); + dataRegion.getWorkSequenceTsFileProcessors().clear(); + dataRegion.getWorkUnsequenceTsFileProcessors().clear(); + assertFalse(new File(lateFile.getPath() + TsFileResource.RESOURCE_SUFFIX).exists()); + + // Discard the memtable and recover through the actual checkpoint/WAL recovery path. + final WALRecoverManager recoverManager = WALRecoverManager.getInstance(); + recoverManager.clear(); + final UnsealedTsFileRecoverPerformer performer = + new UnsealedTsFileRecoverPerformer( + new TsFileResource(lateFile), true, recovered -> assertFalse(recovered.canWrite())); + recoverManager.addRecoverPerformer(performer); + recoverManager.setAllDataRegionScannedLatch(new ExceptionalCountDownLatch(0)); + recoverManager.recover(); + assertEquals(WALRecoverListener.Status.SUCCESS, performer.getRecoverListener().waitForResult()); + assertSingleRow(flushedFile, 1); + assertSingleRow(lateFile, 2); + } + + private void writeAndAssertSuccess(long time) throws Exception { + assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + stateMachine.write((IConsensusRequest) insertRow(time)).getCode()); + } + + private static InsertRowNode insertRow(long time) throws Exception { + final InsertRowNode node = + new InsertRowNode( + new PlanNodeId("shutdown"), + new PartialPath(DEVICE), + false, + new String[] {MEASUREMENT}, + new TSDataType[] {TSDataType.INT32}, + time, + new Object[] {(int) time * 10}, + false); + node.setMeasurementSchemas( + new MeasurementSchema[] {new MeasurementSchema(MEASUREMENT, TSDataType.INT32)}); + return node; + } + + private static void assertSingleRow(File file, long time) throws Exception { + try (TsFileSequenceReader sequenceReader = new TsFileSequenceReader(file.getPath()); + TsFileReader reader = new TsFileReader(sequenceReader)) { + final QueryDataSet dataSet = + reader.query( + QueryExpression.create( + Collections.singletonList(new Path(DEVICE, MEASUREMENT, false)), null)); + assertTrue(dataSet.hasNext()); + final RowRecord row = dataSet.next(); + assertEquals(time, row.getTimestamp()); + assertEquals((int) time * 10, row.getFields().get(0).getIntV()); + assertFalse(dataSet.hasNext()); + } + } +} From 9b29f1cf0096897b84a1360723a580bf8c32b95a Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 14 Oct 2026 16:45:41 +0800 Subject: [PATCH 3/4] Reproduce WAL checkpoint loss from writes after shutdown --- .../manual/enhanced/IoTDBPipeClusterIT.java | 3 ++ .../consensus/iot/IoTConsensusServerImpl.java | 3 +- .../dataregion/IoTConsensusShutdownTest.java | 51 +++++++++++++++++++ 3 files changed, 56 insertions(+), 1 deletion(-) 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 20793825f7d24..cde83cc4da6db 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 @@ -348,6 +348,9 @@ public void testPipeAfterDataRegionLeaderStop() { try (final SyncConfigNodeIServiceClient client = (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) { + // Distinguish a source-side write loss during the restart from a p2 transfer failure. + TableModelUtils.assertData("test1", "test1", 0, 300, senderEnv); + // Create a new pipe and write new data final Map sourceAttributes = new HashMap<>(); final Map processorAttributes = new HashMap<>(); 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 b61a49796187a..afd0c799a2c3e 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 @@ -1584,7 +1584,8 @@ private TSStatus cacheAndInsertLatestNode(DeserializedBatchIndexedConsensusReque IConsensusRequest insertNode = insertNodes.get(i); insertNode.markAsGeneratedByRemoteConsensusLeader(); // 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. + // applying each request: a late write can reopen a stopped WAL node without recovery, + // append a new checkpoint header to the old file, and acknowledge unrecoverable data. TSStatus status = isReadOnly() ? RpcUtils.getStatus( diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java index 5cdd3f78b46ae..1099b8b11310e 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java @@ -47,6 +47,7 @@ import org.apache.iotdb.db.storageengine.dataregion.wal.WALManager; import org.apache.iotdb.db.storageengine.dataregion.wal.buffer.WALBuffer; import org.apache.iotdb.db.storageengine.dataregion.wal.node.WALNode; +import org.apache.iotdb.db.storageengine.dataregion.wal.recover.CheckpointRecoverUtils; import org.apache.iotdb.db.storageengine.dataregion.wal.recover.WALRecoverManager; import org.apache.iotdb.db.storageengine.dataregion.wal.recover.file.UnsealedTsFileRecoverPerformer; import org.apache.iotdb.db.storageengine.dataregion.wal.utils.WALMode; @@ -83,6 +84,7 @@ import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertTrue; public class IoTConsensusShutdownTest { @@ -218,6 +220,55 @@ public void testQueuedSyncLogDoesNotReopenStorageAfterShutdown() throws Exceptio } } + @Test + public void testUncheckedWriteAfterWalManagerStopLosesCheckpoint() throws Exception { + writeAndAssertSuccess(1); + commonConfig.setNodeStatus(NodeStatus.ReadOnly); + WALManager.getInstance().waitAllWALFlushed(); + StorageEngine.getInstance().syncCloseAllProcessor(); + final File flushedFile = dataRegion.getSequenceFileList().get(0).getTsFile(); + final WALNode originalWalNode = (WALNode) dataRegion.getWALNode().get(); + StorageEngine.getInstance().stop(); + WALManager.getInstance().stop(); + + // Bypass the guarded SyncLog entry point to reproduce the old late-apply behavior. Unlike a + // write before WAL stop, this recreates the node in a directory containing closed WAL files. + writeAndAssertSuccess(2); + final WALNode lateWalNode = (WALNode) dataRegion.getWALNode().get(); + assertNotSame(originalWalNode, lateWalNode); + assertEquals(originalWalNode.getLogDirectory(), lateWalNode.getLogDirectory()); + final File lateFile = + dataRegion + .getWorkSequenceTsFileProcessors() + .iterator() + .next() + .getTsFileResource() + .getTsFile(); + WALManager.getInstance().waitAllWALFlushed(); + WALManager.getInstance().stop(); + // The new CheckpointManager appends another header to the existing _0.checkpoint file. Even + // with a drained and normally closed WAL, recovery cannot find the late memtable in that file. + assertTrue( + CheckpointRecoverUtils.recoverMemTableInfo(lateWalNode.getLogDirectory()) + .getMemTableId2Info() + .isEmpty()); + dataRegion.forceCloseAllWorkingTsFileProcessors(); + dataRegion.getWorkSequenceTsFileProcessors().clear(); + dataRegion.getWorkUnsequenceTsFileProcessors().clear(); + + final WALRecoverManager recoverManager = WALRecoverManager.getInstance(); + recoverManager.clear(); + final UnsealedTsFileRecoverPerformer performer = + new UnsealedTsFileRecoverPerformer( + new TsFileResource(lateFile), true, recovered -> assertFalse(recovered.canWrite())); + recoverManager.addRecoverPerformer(performer); + recoverManager.setAllDataRegionScannedLatch(new ExceptionalCountDownLatch(0)); + recoverManager.recover(); + assertEquals(WALRecoverListener.Status.SUCCESS, performer.getRecoverListener().waitForResult()); + assertSingleRow(flushedFile, 1); + assertTrue(performer.getTsFileResource().isEmpty()); + } + @Test public void testAsyncWriteAfterTsFileFlushIsRecoveredWhenWalCloses() throws Exception { writeAndAssertSuccess(1); From d878e7be265da51d91e774f62b9003b7a9859c87 Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Wed, 14 Oct 2026 17:49:25 +0800 Subject: [PATCH 4/4] Fence IoTConsensus applies before storage shutdown --- .../consensus/iot/IoTConsensusServerImpl.java | 6 +- .../dataregion/DataRegionStateMachine.java | 6 +- .../IoTConsensusDataRegionStateMachine.java | 7 + .../db/service/DataNodeShutdownHook.java | 3 + .../iotdb/db/storageengine/StorageEngine.java | 13 + .../storageengine/dataregion/DataRegion.java | 32 ++ .../dataregion/IoTConsensusShutdownTest.java | 281 ++++++++++++++++-- 7 files changed, 316 insertions(+), 32 deletions(-) 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 afd0c799a2c3e..edb725f05c1aa 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,9 +1583,9 @@ private TSStatus cacheAndInsertLatestNode(DeserializedBatchIndexedConsensusReque for (int i = 0; i < insertNodes.size(); i++) { IConsensusRequest insertNode = insertNodes.get(i); insertNode.markAsGeneratedByRemoteConsensusLeader(); - // The replica may become read-only while this batch waits for its turn. Recheck before - // applying each request: a late write can reopen a stopped WAL node without recovery, - // append a new checkpoint header to the old file, and acknowledge unrecoverable data. + // Reject requests that became read-only while queued. This is an early rejection only: + // the state machine must also fence the full apply against storage shutdown, since the + // replica can become read-only after this check. TSStatus status = isReadOnly() ? RpcUtils.getStatus( diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java index 3d2fb3d2006b3..127169ee2f8fa 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java @@ -251,7 +251,7 @@ protected TSStatus write(PlanNode planNode) { TSStatus result = null; int retryTime = 0; while (retryTime < MAX_WRITE_RETRY_TIMES) { - result = planNode.accept(new DataExecutionVisitor(), region); + result = executeWrite(planNode); // Let pipe retry with the original event instead of retrying a possibly mutated node here. if (needRetryForSpecificCases(result.getCode(), planNode)) { retryTime++; @@ -284,6 +284,10 @@ protected TSStatus write(PlanNode planNode) { return result; } + protected TSStatus executeWrite(PlanNode planNode) { + return planNode.accept(new DataExecutionVisitor(), region); + } + protected void waitBeforeNextWriteRetry() throws InterruptedException { Thread.sleep(WRITE_RETRY_WAIT_TIME_IN_MS); } diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusDataRegionStateMachine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusDataRegionStateMachine.java index f3fecdf69f38b..e9275f15587b1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusDataRegionStateMachine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusDataRegionStateMachine.java @@ -50,6 +50,13 @@ public IoTConsensusDataRegionStateMachine(DataRegion region) { super(region); } + @Override + protected TSStatus executeWrite(PlanNode planNode) { + final DataRegion targetRegion = region; + return targetRegion.executeIoTConsensusWrite( + () -> planNode.accept(new DataExecutionVisitor(), targetRegion)); + } + @Override public TSStatus write(IConsensusRequest request) { try { diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java index 828c4b811ac49..cb1095da10434 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java @@ -120,6 +120,9 @@ public void run() { CommonDescriptor.getInstance() .getConfig() .setNodeStatusWithReason(NodeStatus.ReadOnly, NodeStatus.STOPPING); + // A pre-apply read-only check can race with this transition. Drain admitted IoTConsensus + // applies, including their WAL separators, and permanently reject new ones before flushing. + StorageEngine.getInstance().stopIoTConsensusWrites(); // Wait all wal are flushed WALManager.getInstance().waitAllWALFlushed(); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java index 93ec1f560a6bc..f3a86a0c511a1 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java @@ -433,6 +433,7 @@ private void asyncRecoverTsFileResource() { @Override public void stop() { + stopIoTConsensusWrites(); loadTsFileManager.stop(); for (DataRegion dataRegion : dataRegionMap.values()) { if (dataRegion != null) { @@ -453,6 +454,7 @@ public void stop() { @Override public void shutdown(long milliseconds) throws ShutdownException { + stopIoTConsensusWrites(); loadTsFileManager.stop(); try { for (DataRegion dataRegion : dataRegionMap.values()) { @@ -524,6 +526,17 @@ public synchronized void reset() { dataRegionMap.clear(); } + public void stopIoTConsensusWrites() { + if (!CONFIG.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS)) { + return; + } + for (DataRegion dataRegion : dataRegionMap.values()) { + if (dataRegion != null) { + dataRegion.stopIoTConsensusWrites(); + } + } + } + /** flush command Sync asyncCloseOneProcessor all file node processors. */ public void syncCloseAllProcessor() { LOGGER.info(StorageEngineMessages.START_CLOSING_ALL_DB_PROCESSOR); diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index ead75db6bac6e..31629000e8829 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java @@ -33,6 +33,7 @@ import org.apache.iotdb.commons.exception.MetadataException; import org.apache.iotdb.commons.exception.MetadataLeaseFencedException.LeaseFencedRetryPolicy; import org.apache.iotdb.commons.file.SystemFileFactory; +import org.apache.iotdb.commons.i18n.CommonMessages; import org.apache.iotdb.commons.path.IFullPath; import org.apache.iotdb.commons.path.MeasurementPath; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; @@ -231,6 +232,7 @@ import java.util.concurrent.locks.Condition; import java.util.concurrent.locks.ReadWriteLock; import java.util.concurrent.locks.ReentrantReadWriteLock; +import java.util.function.Supplier; import java.util.stream.Collectors; import java.util.stream.Stream; @@ -284,6 +286,12 @@ public class DataRegion implements IDataRegionForQuery { */ private final ReadWriteLock insertLock = new ReentrantReadWriteLock(); + // Covers a complete IoTConsensus apply, including its WAL separator. Keep this separate from + // insertLock so memory backpressure and retry waits do not prevent flushing memtables. + private final ReadWriteLock ioTConsensusWriteLock = new ReentrantReadWriteLock(); + // Accessed only under ioTConsensusWriteLock; stopping a region permanently closes admission. + private boolean ioTConsensusWritesStopped; + /** Condition to safely delete data region. */ private final Condition deletedCondition = insertLock.writeLock().newCondition(); @@ -2527,6 +2535,30 @@ public void timedFlushUnseqMemTable() { WritingMetrics.getInstance().recordTimedFlushMemTableCount(count); } + public TSStatus executeIoTConsensusWrite(Supplier write) { + ioTConsensusWriteLock.readLock().lock(); + try { + if (ioTConsensusWritesStopped + || CommonDescriptor.getInstance().getConfig().isStopping() + || CommonDescriptor.getInstance().getConfig().isReadOnly()) { + return RpcUtils.getStatus(TSStatusCode.SYSTEM_READ_ONLY, CommonMessages.SYSTEM_READ_ONLY); + } + return write.get(); + } finally { + ioTConsensusWriteLock.readLock().unlock(); + } + } + + /** Reject new IoTConsensus applies and drain admitted applies before closing storage or WAL. */ + public void stopIoTConsensusWrites() { + ioTConsensusWriteLock.writeLock().lock(); + try { + ioTConsensusWritesStopped = true; + } finally { + ioTConsensusWriteLock.writeLock().unlock(); + } + } + /** This method will be blocked until all tsfile processors are closed. */ public void syncCloseAllWorkingTsFileProcessors() { try { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java index 1099b8b11310e..fed7a6151a197 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java @@ -28,6 +28,7 @@ import org.apache.iotdb.commons.consensus.DataRegionId; import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType; import org.apache.iotdb.commons.path.PartialPath; +import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.IPlanVisitor; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.commons.request.IConsensusRequest; import org.apache.iotdb.consensus.ConsensusFactory; @@ -80,12 +81,16 @@ import java.util.concurrent.Future; import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicInteger; import java.util.concurrent.atomic.AtomicReference; import static org.junit.Assert.assertEquals; import static org.junit.Assert.assertFalse; import static org.junit.Assert.assertNotSame; import static org.junit.Assert.assertTrue; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doAnswer; +import static org.mockito.Mockito.spy; public class IoTConsensusShutdownTest { @@ -102,6 +107,7 @@ public class IoTConsensusShutdownTest { private String previousConsensusProtocol; private NodeStatus previousNodeStatus; private String previousStatusReason; + private boolean previousStopping; private DataRegion dataRegion; private IoTConsensusDataRegionStateMachine stateMachine; @@ -112,10 +118,12 @@ public void setUp() throws Exception { previousConsensusProtocol = config.getDataRegionConsensusProtocolClass(); previousNodeStatus = commonConfig.getNodeStatus(); previousStatusReason = commonConfig.getStatusReason(); + previousStopping = commonConfig.isStopping(); config.setWalMode(WALMode.ASYNC); config.setDataNodeId(0); config.setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS); commonConfig.setNodeStatus(NodeStatus.Running); + commonConfig.setStopping(false); EnvironmentUtils.envSetUp(); dataRegion = new DataRegion( @@ -130,6 +138,7 @@ public void setUp() throws Exception { @After public void tearDown() throws Exception { try { + commonConfig.setStopping(false); commonConfig.setNodeStatus(NodeStatus.Running); if (dataRegion != null) { dataRegion.syncCloseAllWorkingTsFileProcessors(); @@ -142,6 +151,7 @@ public void tearDown() throws Exception { config.setDataRegionConsensusProtocolClass(previousConsensusProtocol); commonConfig.setNodeStatus(NodeStatus.Running); commonConfig.setNodeStatusWithReason(previousNodeStatus, previousStatusReason); + commonConfig.setStopping(previousStopping); } } @@ -150,25 +160,7 @@ public void testQueuedSyncLogDoesNotReopenStorageAfterShutdown() throws Exceptio writeAndAssertSuccess(1); final ScheduledExecutorService backgroundTaskService = Executors.newSingleThreadScheduledExecutor(); - final Peer thisNode = new Peer(new DataRegionId(0), 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() - .setMaxPendingBatchesNum(2) - .setMaxWaitingTimeForWaitBatchInMs(TimeUnit.MINUTES.toMillis(5)) - .build()) - .build()); + final IoTConsensusServerImpl server = newServer(stateMachine, backgroundTaskService, 2); final DeserializedBatchIndexedConsensusRequest request = new DeserializedBatchIndexedConsensusRequest(1, 2, 2, 1, 2); request.add(insertRow(2)); @@ -183,7 +175,9 @@ public void testQueuedSyncLogDoesNotReopenStorageAfterShutdown() throws Exceptio .until(() -> syncLogThread.getState() == Thread.State.TIMED_WAITING); // Follow the production order: read-only, final flush, storage stop, then WAL stop. + commonConfig.setStopping(true); commonConfig.setNodeStatus(NodeStatus.ReadOnly); + StorageEngine.getInstance().stopIoTConsensusWrites(); WALManager.getInstance().waitAllWALFlushed(); StorageEngine.getInstance().syncCloseAllProcessor(); final File flushedFile = dataRegion.getSequenceFileList().get(0).getTsFile(); @@ -220,6 +214,225 @@ public void testQueuedSyncLogDoesNotReopenStorageAfterShutdown() throws Exceptio } } + @Test + public void testSyncLogBecomesReadOnlyAfterPreApplyCheck() throws Exception { + writeAndAssertSuccess(1); + final CountDownLatch preApplyCheckPassed = new CountDownLatch(1); + final CountDownLatch resumeApply = new CountDownLatch(1); + final IoTConsensusDataRegionStateMachine delayedStateMachine = spy(stateMachine); + doAnswer( + invocation -> { + preApplyCheckPassed.countDown(); + assertTrue(resumeApply.await(30, TimeUnit.SECONDS)); + return invocation.callRealMethod(); + }) + .when(delayedStateMachine) + .write(any(IConsensusRequest.class)); + final ScheduledExecutorService backgroundTaskService = + Executors.newSingleThreadScheduledExecutor(); + final ExecutorService syncLogExecutor = Executors.newSingleThreadExecutor(); + final IoTConsensusServerImpl server = newServer(delayedStateMachine, backgroundTaskService, 1); + final DeserializedBatchIndexedConsensusRequest request = + new DeserializedBatchIndexedConsensusRequest(1, 2, 2, 1, 2); + request.add(insertRow(2)); + request.add(insertRow(3)); + final Future syncLog = syncLogExecutor.submit(() -> server.syncLog(1, request)); + try { + // The consensus-layer read-only check has returned false, but apply has not started. + assertTrue(preApplyCheckPassed.await(10, TimeUnit.SECONDS)); + commonConfig.setStopping(true); + commonConfig.setNodeStatus(NodeStatus.ReadOnly); + StorageEngine.getInstance().stopIoTConsensusWrites(); + WALManager.getInstance().waitAllWALFlushed(); + StorageEngine.getInstance().syncCloseAllProcessor(); + final File flushedFile = dataRegion.getSequenceFileList().get(0).getTsFile(); + StorageEngine.getInstance().stop(); + WALManager.getInstance().stop(); + assertEquals(0, WALManager.getInstance().getWALNodesNum()); + + resumeApply.countDown(); + final TSStatus status = syncLog.get(10, TimeUnit.SECONDS); + assertEquals(2, status.getSubStatusSize()); + assertEquals( + TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), status.getSubStatus().get(0).getCode()); + assertEquals( + TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode(), + status.getSubStatus().get(1).getCode()); + assertEquals(0, server.getEffectiveSafePhysicalTime(1)); + assertEquals(0, WALManager.getInstance().getWALNodesNum()); + assertTrue(dataRegion.getWorkSequenceTsFileProcessors().isEmpty()); + assertTrue(dataRegion.getWorkUnsequenceTsFileProcessors().isEmpty()); + assertSingleRow(flushedFile, 1); + } finally { + resumeApply.countDown(); + syncLogExecutor.shutdownNow(); + assertTrue(syncLogExecutor.awaitTermination(10, TimeUnit.SECONDS)); + backgroundTaskService.shutdownNow(); + } + } + + @Test + public void testShutdownDrainsAdmittedApplyBeforeClosingStorage() throws Exception { + writeAndAssertSuccess(1); + final CountDownLatch admittedApply = new CountDownLatch(1); + final CountDownLatch resumeApply = new CountDownLatch(1); + final InsertRowNode pendingRow = spy(insertRow(2)); + doAnswer( + invocation -> { + admittedApply.countDown(); + assertTrue(resumeApply.await(30, TimeUnit.SECONDS)); + return invocation.callRealMethod(); + }) + .when(pendingRow) + .accept(any(IPlanVisitor.class), any(DataRegion.class)); + final ExecutorService executor = Executors.newFixedThreadPool(2); + final CountDownLatch shutdownStarted = new CountDownLatch(1); + final CountDownLatch admissionClosed = new CountDownLatch(1); + final AtomicReference shutdownThread = new AtomicReference<>(); + final Future write = + executor.submit(() -> stateMachine.write((IConsensusRequest) pendingRow)); + try { + // Unlike the pre-apply-check test, this request has passed the check under the lifecycle + // lock. + assertTrue(admittedApply.await(10, TimeUnit.SECONDS)); + commonConfig.setStopping(true); + commonConfig.setNodeStatus(NodeStatus.ReadOnly); + final Future shutdown = + executor.submit( + () -> { + shutdownThread.set(Thread.currentThread()); + shutdownStarted.countDown(); + StorageEngine.getInstance().stopIoTConsensusWrites(); + admissionClosed.countDown(); + WALManager.getInstance().waitAllWALFlushed(); + StorageEngine.getInstance().syncCloseAllProcessor(); + StorageEngine.getInstance().stop(); + WALManager.getInstance().stop(); + }); + assertTrue(shutdownStarted.await(10, TimeUnit.SECONDS)); + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .until( + () -> + admissionClosed.getCount() == 0 + || shutdownThread.get().getState() == Thread.State.WAITING); + assertEquals(1, admissionClosed.getCount()); + assertFalse(shutdown.isDone()); + assertFalse(dataRegion.getSequenceFileList().get(0).isClosed()); + + resumeApply.countDown(); + assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), write.get(10, TimeUnit.SECONDS).getCode()); + shutdown.get(10, TimeUnit.SECONDS); + assertRows(dataRegion.getSequenceFileList().get(0).getTsFile(), 1, 2); + assertTrue(dataRegion.getWorkSequenceTsFileProcessors().isEmpty()); + assertTrue(dataRegion.getWorkUnsequenceTsFileProcessors().isEmpty()); + assertEquals(0, WALManager.getInstance().getWALNodesNum()); + + // Stopping storage closes admission permanently, even if the node status changes again. + commonConfig.setStopping(false); + commonConfig.setNodeStatus(NodeStatus.Running); + assertEquals( + TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), + stateMachine.write((IConsensusRequest) insertRow(3)).getCode()); + assertEquals(0, WALManager.getInstance().getWALNodesNum()); + } finally { + resumeApply.countDown(); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + @Test + public void testShutdownDuringWriteRetryDoesNotHoldAdmissionLock() throws Exception { + writeAndAssertSuccess(1); + final CountDownLatch retryWaiting = new CountDownLatch(1); + final CountDownLatch resumeRetry = new CountDownLatch(1); + stateMachine = + new IoTConsensusDataRegionStateMachine(dataRegion) { + @Override + protected void waitBeforeNextWriteRetry() throws InterruptedException { + retryWaiting.countDown(); + assertTrue(resumeRetry.await(30, TimeUnit.SECONDS)); + } + }; + final AtomicInteger applyAttempts = new AtomicInteger(); + final InsertRowNode pendingRow = spy(insertRow(2)); + doAnswer( + invocation -> + applyAttempts.incrementAndGet() == 1 + ? new TSStatus(TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode()) + : invocation.callRealMethod()) + .when(pendingRow) + .accept(any(IPlanVisitor.class), any(DataRegion.class)); + final ExecutorService executor = Executors.newSingleThreadExecutor(); + final Future write = + executor.submit(() -> stateMachine.write((IConsensusRequest) pendingRow)); + try { + assertTrue(retryWaiting.await(10, TimeUnit.SECONDS)); + commonConfig.setStopping(true); + commonConfig.setNodeStatus(NodeStatus.ReadOnly); + StorageEngine.getInstance().stopIoTConsensusWrites(); + WALManager.getInstance().waitAllWALFlushed(); + StorageEngine.getInstance().syncCloseAllProcessor(); + final File flushedFile = dataRegion.getSequenceFileList().get(0).getTsFile(); + StorageEngine.getInstance().stop(); + WALManager.getInstance().stop(); + assertFalse(write.isDone()); + + resumeRetry.countDown(); + assertEquals( + TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), write.get(10, TimeUnit.SECONDS).getCode()); + assertEquals(1, applyAttempts.get()); + assertTrue(dataRegion.getWorkSequenceTsFileProcessors().isEmpty()); + assertTrue(dataRegion.getWorkUnsequenceTsFileProcessors().isEmpty()); + assertEquals(0, WALManager.getInstance().getWALNodesNum()); + assertSingleRow(flushedFile, 1); + } finally { + resumeRetry.countDown(); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + @Test + public void testReadOnlyCanBecomeWritableBeforeShutdown() throws Exception { + writeAndAssertSuccess(1); + commonConfig.setNodeStatus(NodeStatus.ReadOnly); + assertEquals( + TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), + stateMachine.write((IConsensusRequest) insertRow(2)).getCode()); + commonConfig.setNodeStatus(NodeStatus.Running); + writeAndAssertSuccess(2); + dataRegion.syncCloseAllWorkingTsFileProcessors(); + assertRows(dataRegion.getSequenceFileList().get(0).getTsFile(), 1, 2); + } + + private IoTConsensusServerImpl newServer( + IoTConsensusDataRegionStateMachine targetStateMachine, + ScheduledExecutorService backgroundTaskService, + int maxPendingBatchesNum) + throws Exception { + final Peer thisNode = new Peer(new DataRegionId(0), 0, new TEndPoint("127.0.0.1", 6667)); + return new IoTConsensusServerImpl( + temporaryFolder.getRoot().getAbsolutePath(), + null, + DirectoryStrategyType.SEQUENCE_STRATEGY, + thisNode, + Collections.singletonList(thisNode), + targetStateMachine, + backgroundTaskService, + null, + null, + IoTConsensusConfig.newBuilder() + .setReplication( + IoTConsensusConfig.Replication.newBuilder() + .setMaxPendingBatchesNum(maxPendingBatchesNum) + .setMaxWaitingTimeForWaitBatchInMs(TimeUnit.MINUTES.toMillis(5)) + .build()) + .build()); + } + @Test public void testUncheckedWriteAfterWalManagerStopLosesCheckpoint() throws Exception { writeAndAssertSuccess(1); @@ -231,9 +444,9 @@ public void testUncheckedWriteAfterWalManagerStopLosesCheckpoint() throws Except StorageEngine.getInstance().stop(); WALManager.getInstance().stop(); - // Bypass the guarded SyncLog entry point to reproduce the old late-apply behavior. Unlike a - // write before WAL stop, this recreates the node in a directory containing closed WAL files. - writeAndAssertSuccess(2); + // Bypass lifecycle admission with the real visitor to reproduce the old late-apply behavior. + // Unlike a write before WAL stop, this recreates the node in a directory containing closed WAL. + writeUncheckedAndAssertSuccess(2); final WALNode lateWalNode = (WALNode) dataRegion.getWALNode().get(); assertNotSame(originalWalNode, lateWalNode); assertEquals(originalWalNode.getLogDirectory(), lateWalNode.getLogDirectory()); @@ -295,8 +508,8 @@ public void testAsyncWriteAfterTsFileFlushIsRecoveredWhenWalCloses() throws Exce final File lateFile; try { assertTrue(syncBlocked.await(10, TimeUnit.SECONDS)); - // Call the real state machine directly to model the old unchecked SyncLog apply. - writeAndAssertSuccess(2); + // Call the real visitor directly to model the old unchecked SyncLog apply. + writeUncheckedAndAssertSuccess(2); assertFalse(buffer.isAllWALEntriesConsumed()); assertEquals(1, dataRegion.getWorkSequenceTsFileProcessors().size()); final TsFileProcessor processor = @@ -335,6 +548,12 @@ private void writeAndAssertSuccess(long time) throws Exception { stateMachine.write((IConsensusRequest) insertRow(time)).getCode()); } + private void writeUncheckedAndAssertSuccess(long time) throws Exception { + assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + insertRow(time).accept(new DataExecutionVisitor(), dataRegion).getCode()); + } + private static InsertRowNode insertRow(long time) throws Exception { final InsertRowNode node = new InsertRowNode( @@ -352,16 +571,22 @@ private static InsertRowNode insertRow(long time) throws Exception { } private static void assertSingleRow(File file, long time) throws Exception { + assertRows(file, time); + } + + private static void assertRows(File file, long... times) throws Exception { try (TsFileSequenceReader sequenceReader = new TsFileSequenceReader(file.getPath()); TsFileReader reader = new TsFileReader(sequenceReader)) { final QueryDataSet dataSet = reader.query( QueryExpression.create( Collections.singletonList(new Path(DEVICE, MEASUREMENT, false)), null)); - assertTrue(dataSet.hasNext()); - final RowRecord row = dataSet.next(); - assertEquals(time, row.getTimestamp()); - assertEquals((int) time * 10, row.getFields().get(0).getIntV()); + for (long time : times) { + assertTrue(dataSet.hasNext()); + final RowRecord row = dataSet.next(); + assertEquals(time, row.getTimestamp()); + assertEquals((int) time * 10, row.getFields().get(0).getIntV()); + } assertFalse(dataSet.hasNext()); } }