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..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<>(); @@ -449,6 +452,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 +461,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 +487,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 +506,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..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,7 +1583,14 @@ private TSStatus cacheAndInsertLatestNode(DeserializedBatchIndexedConsensusReque for (int i = 0; i < insertNodes.size(); i++) { IConsensusRequest insertNode = insertNodes.get(i); insertNode.markAsGeneratedByRemoteConsensusLeader(); - TSStatus status = stateMachine.write(insertNode); + // 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( + 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); 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 new file mode 100644 index 0000000000000..fed7a6151a197 --- /dev/null +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusShutdownTest.java @@ -0,0 +1,593 @@ +/* + * 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.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; +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.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; +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.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 { + + 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 boolean previousStopping; + 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(); + 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( + 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.setStopping(false); + 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); + commonConfig.setStopping(previousStopping); + } + } + + @Test + public void testQueuedSyncLogDoesNotReopenStorageAfterShutdown() throws Exception { + writeAndAssertSuccess(1); + final ScheduledExecutorService backgroundTaskService = + Executors.newSingleThreadScheduledExecutor(); + final IoTConsensusServerImpl server = newServer(stateMachine, backgroundTaskService, 2); + 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.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()); + 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 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); + 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 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()); + 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); + 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 visitor directly to model the old unchecked SyncLog apply. + writeUncheckedAndAssertSuccess(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 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( + 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 { + 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)); + 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()); + } + } +}