From 444bf44e92a0401c8ebd07f05e78d7f18f7db56b Mon Sep 17 00:00:00 2001 From: Caideyipi <87789683+Caideyipi@users.noreply.github.com> Date: Thu, 15 Oct 2026 17:43:49 +0800 Subject: [PATCH] Reuse DataRegion insert lock for IoTConsensus shutdown fencing --- .../dataregion/DataExecutionVisitor.java | 10 +- .../dataregion/DataRegionStateMachine.java | 6 +- .../IoTConsensusDataRegionStateMachine.java | 7 - .../storageengine/dataregion/DataRegion.java | 129 ++++--- .../dataregion/IoTConsensusShutdownTest.java | 343 ++++++++++++++++-- ...nerSeqCompactionWithFastPerformerTest.java | 3 +- ...qCompactionWithReadChunkPerformerTest.java | 7 +- 7 files changed, 422 insertions(+), 83 deletions(-) diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java index 96bbb6ccc16ba..c0034ccc77c55 100644 --- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java +++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataExecutionVisitor.java @@ -78,7 +78,6 @@ public TSStatus visitRelationalInsertRows(RelationalInsertRowsNode node, DataReg public TSStatus visitInsertRow(InsertRowNode node, DataRegion dataRegion) { try { dataRegion.insert(node); - dataRegion.insertSeparatorToWAL(node); return StatusUtils.OK; } catch (OutOfTTLException e) { LOGGER.warn(DataNodeMiscMessages.ERROR_EXECUTING_PLAN_NODE_CAUSED, node, e.getMessage()); @@ -104,7 +103,6 @@ public TSStatus visitRelationalInsertTablet( public TSStatus visitInsertTablet(final InsertTabletNode node, final DataRegion dataRegion) { try { dataRegion.insertTablet(node); - dataRegion.insertSeparatorToWAL(node); return StatusUtils.OK; } catch (final OutOfTTLException e) { LOGGER.debug(DataNodeMiscMessages.ERROR_EXECUTING_PLAN_NODE_CAUSED, node, e.getMessage()); @@ -144,7 +142,6 @@ public TSStatus visitInsertTablet(final InsertTabletNode node, final DataRegion public TSStatus visitInsertRows(InsertRowsNode node, DataRegion dataRegion) { try { dataRegion.insert(node); - dataRegion.insertSeparatorToWAL(node); return StatusUtils.OK; } catch (WriteProcessRejectException e) { LOGGER.warn(DataNodeMiscMessages.REJECT_EXECUTING_PLAN_NODE, node, e.getMessage()); @@ -184,7 +181,6 @@ public TSStatus visitInsertRows(InsertRowsNode node, DataRegion dataRegion) { public TSStatus visitInsertMultiTablets(InsertMultiTabletsNode node, DataRegion dataRegion) { try { dataRegion.insertTablets(node); - dataRegion.insertSeparatorToWAL(node); return StatusUtils.OK; } catch (WriteProcessRejectException e) { LOGGER.warn(DataNodeMiscMessages.REJECT_EXECUTING_PLAN_NODE, node, e.getMessage()); @@ -222,7 +218,6 @@ public TSStatus visitInsertRowsOfOneDevice( InsertRowsOfOneDeviceNode node, DataRegion dataRegion) { try { dataRegion.insert(node); - dataRegion.insertSeparatorToWAL(node); return StatusUtils.OK; } catch (WriteProcessRejectException e) { LOGGER.warn(DataNodeMiscMessages.REJECT_EXECUTING_PLAN_NODE, node, e.getMessage()); @@ -284,6 +279,8 @@ public TSStatus visitDeleteData(DeleteDataNode node, DataRegion dataRegion) { } dataRegion.insertSeparatorToWAL(node); return StatusUtils.OK; + } catch (WriteProcessRejectException e) { + return RpcUtils.getStatus(e.getErrorCode(), e.getMessage()); } catch (IOException | IllegalPathException e) { LOGGER.error(DataNodeMiscMessages.ERROR_EXECUTING_PLAN_NODE, node, e); return new TSStatus(TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode()); @@ -297,8 +294,9 @@ public TSStatus visitDeleteData( final RelationalDeleteDataNode node, final DataRegion dataRegion) { try { dataRegion.deleteByTable(node); - dataRegion.insertSeparatorToWAL(node); return StatusUtils.OK; + } catch (final WriteProcessRejectException e) { + return RpcUtils.getStatus(e.getErrorCode(), e.getMessage()); } catch (final IOException e) { LOGGER.error(DataNodeMiscMessages.ERROR_EXECUTING_PLAN_NODE, node, e); return new TSStatus(TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode()); 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 127169ee2f8fa..3d2fb3d2006b3 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 = executeWrite(planNode); + result = planNode.accept(new DataExecutionVisitor(), region); // Let pipe retry with the original event instead of retrying a possibly mutated node here. if (needRetryForSpecificCases(result.getCode(), planNode)) { retryTime++; @@ -284,10 +284,6 @@ 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 e9275f15587b1..f3fecdf69f38b 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,13 +50,6 @@ 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/storageengine/dataregion/DataRegion.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/dataregion/DataRegion.java index 1ac4d80b94fd3..2dcb699807d9a 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 @@ -234,7 +234,6 @@ 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; @@ -287,11 +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. + // Accessed only under insertLock's write lock; stopping permanently closes write admission. private boolean ioTConsensusWritesStopped; + // Table deletions release insertLock while updating sealed files. Drain that work, including + // its WAL separator, without extending the insert critical section over mod-file I/O. + private int ongoingIoTConsensusDeletions; + private final Condition ioTConsensusDeletionsDrained = insertLock.writeLock().newCondition(); /** Condition to safely delete data region. */ private final Condition deletedCondition = insertLock.writeLock().newCondition(); @@ -1199,6 +1199,7 @@ public void insert(InsertRowNode insertRowNode) throws WriteProcessException { writeLock("InsertRow"); PERFORMANCE_OVERVIEW_METRICS.recordScheduleLockCost(System.nanoTime() - startTime); try { + checkIoTConsensusWriteAllowed(); if (deleted) { return; } @@ -1254,6 +1255,7 @@ public void insert(InsertRowNode insertRowNode) throws WriteProcessException { PERFORMANCE_OVERVIEW_METRICS.recordScheduleUpdateLastCacheCost( System.nanoTime() - startTime); } + logSeparatorToWAL(insertRowNode); } finally { writeUnlock(); } @@ -1432,6 +1434,7 @@ public void insertTablet(InsertTabletNode insertTabletNode) writeLock("insertTablet"); PERFORMANCE_OVERVIEW_METRICS.recordScheduleLockCost(System.nanoTime() - startTime); try { + checkIoTConsensusWriteAllowed(); if (deleted) { logger.info( StorageEngineMessages @@ -1460,6 +1463,7 @@ public void insertTablet(InsertTabletNode insertTabletNode) if (!noFailure) { throw new BatchProcessException(results); } + logSeparatorToWAL(insertTabletNode); } finally { writeUnlock(); } @@ -2540,27 +2544,27 @@ 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(); + /** Called under insertLock's write lock, after any memory-backpressure wait. */ + protected void checkIoTConsensusWriteAllowed() throws WriteProcessRejectException { + if (config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS) + && (ioTConsensusWritesStopped + || CommonDescriptor.getInstance().getConfig().isStopping() + || CommonDescriptor.getInstance().getConfig().isReadOnly())) { + throw new WriteProcessRejectException( + CommonMessages.SYSTEM_READ_ONLY, TSStatusCode.SYSTEM_READ_ONLY.getStatusCode()); } } /** Reject new IoTConsensus applies and drain admitted applies before closing storage or WAL. */ public void stopIoTConsensusWrites() { - ioTConsensusWriteLock.writeLock().lock(); + writeLock("stopIoTConsensusWrites"); try { ioTConsensusWritesStopped = true; + while (ongoingIoTConsensusDeletions != 0) { + ioTConsensusDeletionsDrained.awaitUninterruptibly(); + } } finally { - ioTConsensusWriteLock.writeLock().unlock(); + writeUnlock(); } } @@ -3321,7 +3325,7 @@ private void getTwoKindsOfTsFiles( } public void deleteByDevice(final MeasurementPath pattern, final DeleteDataNode node) - throws IOException { + throws IOException, WriteProcessRejectException { if (SettleService.getINSTANCE().getFilesToBeSettledCount().get() != 0) { throw new IOException( StorageEngineMessages @@ -3338,6 +3342,7 @@ public void deleteByDevice(final MeasurementPath pattern, final DeleteDataNode n boolean hasReleasedLock = false; try { + checkIoTConsensusWriteAllowed(); if (deleted) { return; } @@ -3373,6 +3378,8 @@ public void deleteByDevice(final MeasurementPath pattern, final DeleteDataNode n hasReleasedLock = true; deleteDataInSealedFiles(sealedTsFileResource, deletion); + } catch (WriteProcessRejectException e) { + throw e; } catch (Exception e) { throw new IOException(e); } finally { @@ -3387,7 +3394,8 @@ public TDropTableModProgress getDropTableModProgress(final long procedureId) { return progress != null && progress.left == procedureId ? progress.right.snapshot() : null; } - public void deleteByTable(RelationalDeleteDataNode node) throws IOException { + public void deleteByTable(RelationalDeleteDataNode node) + throws IOException, WriteProcessRejectException { if (node.getDatabaseName() != null && !node.getDatabaseName().equals(databaseName)) { // not targeted on this database, return return; @@ -3410,7 +3418,13 @@ public void deleteByTable(RelationalDeleteDataNode node) throws IOException { writeLock("delete"); boolean hasReleasedLock = false; + boolean ioTConsensusDeletionAdmitted = false; try { + checkIoTConsensusWriteAllowed(); + if (config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS)) { + ongoingIoTConsensusDeletions++; + ioTConsensusDeletionAdmitted = true; + } if (progress != null) { dropTableModProgress = new Pair<>(procedureId, progress); } @@ -3501,9 +3515,23 @@ public void deleteByTable(RelationalDeleteDataNode node) throws IOException { for (int i = 0; i < modEntries.size(); i++) { deleteDataInSealedFiles(sealedTsFileResourceLists.get(i), modEntries.get(i), progress); } + writeLock("deleteSeparatorToWAL"); + hasReleasedLock = false; + // Shutdown waits for this admitted deletion even if it has already closed admission. + logSeparatorToWAL(node); + } catch (WriteProcessRejectException e) { + throw e; } catch (Exception e) { throw new IOException(e); } finally { + if (ioTConsensusDeletionAdmitted) { + if (hasReleasedLock) { + writeLock("finishIoTConsensusDeletion"); + hasReleasedLock = false; + } + ongoingIoTConsensusDeletions--; + ioTConsensusDeletionsDrained.signalAll(); + } if (!hasReleasedLock) { writeUnlock(); } @@ -3511,7 +3539,7 @@ public void deleteByTable(RelationalDeleteDataNode node) throws IOException { } public void deleteDataDirectly(MeasurementPath pathToDelete, DeleteDataNode node) - throws IOException { + throws IOException, WriteProcessRejectException { final long startTime = node.getDeleteStartTime(); final long endTime = node.getDeleteEndTime(); final long searchIndex = node.getSearchIndex(); @@ -3526,6 +3554,7 @@ public void deleteDataDirectly(MeasurementPath pathToDelete, DeleteDataNode node boolean releasedLock = false; try { + checkIoTConsensusWriteAllowed(); if (deleted) { return; } @@ -3556,6 +3585,8 @@ public void deleteDataDirectly(MeasurementPath pathToDelete, DeleteDataNode node writeUnlock(); releasedLock = true; deleteDataDirectlyInFile(sealedTsFileResource, deletion); + } catch (WriteProcessRejectException e) { + throw e; } catch (Exception e) { throw new IOException(e); } finally { @@ -3720,25 +3751,34 @@ public void insertSeparatorToWAL() { public void insertSeparatorToWAL(final SearchNode sourceNode) { writeLock("insertSeparatorToWAL"); try { - if (deleted) { + if (deleted || ioTConsensusWritesStopped) { return; } - final ContinuousSameSearchIndexSeparatorNode separatorNode = - new ContinuousSameSearchIndexSeparatorNode(); - if (Objects.nonNull(sourceNode)) { - separatorNode - .setRoutingEpoch(sourceNode.getRoutingEpoch()) - .setPhysicalTime(sourceNode.getPhysicalTime()) - .setNodeId(sourceNode.getNodeId()) - .setSyncIndex(sourceNode.getSyncIndex()); - } - getWALNode() - .ifPresent(walNode -> walNode.log(TsFileProcessor.MEMTABLE_NOT_EXIST, separatorNode)); + logSeparatorToWAL(sourceNode); } finally { writeUnlock(); } } + /** Called under insertLock's write lock as part of an admitted write. */ + private void logSeparatorToWAL(final SearchNode sourceNode) { + if (deleted + || !config.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS)) { + return; + } + final ContinuousSameSearchIndexSeparatorNode separatorNode = + new ContinuousSameSearchIndexSeparatorNode(); + if (Objects.nonNull(sourceNode)) { + separatorNode + .setRoutingEpoch(sourceNode.getRoutingEpoch()) + .setPhysicalTime(sourceNode.getPhysicalTime()) + .setNodeId(sourceNode.getNodeId()) + .setSyncIndex(sourceNode.getSyncIndex()); + } + getWALNode() + .ifPresent(walNode -> walNode.log(TsFileProcessor.MEMTABLE_NOT_EXIST, separatorNode)); + } + @SuppressWarnings("OptionalGetWithoutIsPresent") private boolean canSkipDelete(TsFileResource tsFileResource, ModEntry deletion) { long fileStartTime = tsFileResource.getTimeIndex().getMinStartTime(); @@ -5024,6 +5064,7 @@ public void insert(InsertRowsOfOneDeviceNode insertRowsOfOneDeviceNode) writeLock("InsertRowsOfOneDevice"); PERFORMANCE_OVERVIEW_METRICS.recordScheduleLockCost(System.nanoTime() - startTime); try { + checkIoTConsensusWriteAllowed(); if (deleted) { return; } @@ -5125,13 +5166,14 @@ public void insert(InsertRowsOfOneDeviceNode insertRowsOfOneDeviceNode) PERFORMANCE_OVERVIEW_METRICS.recordScheduleUpdateLastCacheCost( System.nanoTime() - startTime); } + if (!insertRowsOfOneDeviceNode.getResults().isEmpty()) { + throw new BatchProcessException( + StorageEngineMessages.PARTIAL_FAILED_INSERTING_ROWS_ONE_DEVICE); + } + logSeparatorToWAL(insertRowsOfOneDeviceNode); } finally { writeUnlock(); } - if (!insertRowsOfOneDeviceNode.getResults().isEmpty()) { - throw new BatchProcessException( - StorageEngineMessages.PARTIAL_FAILED_INSERTING_ROWS_ONE_DEVICE); - } } public void insert(InsertRowsNode insertRowsNode) @@ -5141,6 +5183,7 @@ public void insert(InsertRowsNode insertRowsNode) writeLock("InsertRows"); PERFORMANCE_OVERVIEW_METRICS.recordScheduleLockCost(System.nanoTime() - startTime); try { + checkIoTConsensusWriteAllowed(); if (deleted) { return; } @@ -5211,6 +5254,7 @@ public void insert(InsertRowsNode insertRowsNode) if (!insertRowsNode.getResults().isEmpty()) { throw new BatchProcessException(StorageEngineMessages.PARTIAL_FAILED_INSERTING_ROWS); } + logSeparatorToWAL(insertRowsNode); } finally { writeUnlock(); } @@ -5229,6 +5273,7 @@ public void insertTablets(InsertMultiTabletsNode insertMultiTabletsNode) writeLock("insertTablets"); PERFORMANCE_OVERVIEW_METRICS.recordScheduleLockCost(System.nanoTime() - startTime); try { + checkIoTConsensusWriteAllowed(); if (deleted) { logger.info( StorageEngineMessages @@ -5282,14 +5327,14 @@ public void insertTablets(InsertMultiTabletsNode insertMultiTabletsNode) } } updateTsFileProcessorMetric(insertMultiTabletsNode, infoForMetrics); - + if (!insertMultiTabletsNode.getResults().isEmpty()) { + throw new BatchProcessException( + StorageEngineMessages.PARTIAL_FAILED_INSERTING_MULTI_TABLETS); + } + logSeparatorToWAL(insertMultiTabletsNode); } finally { writeUnlock(); } - - if (!insertMultiTabletsNode.getResults().isEmpty()) { - throw new BatchProcessException(StorageEngineMessages.PARTIAL_FAILED_INSERTING_MULTI_TABLETS); - } } private void updateTsFileProcessorMetric(InsertNode insertNode, long[] infoForMetrics) { 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 fed7a6151a197..bf2f5c21e6ade 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 @@ -27,8 +27,10 @@ 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.MeasurementPath; 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.PlanNode; import org.apache.iotdb.commons.queryengine.plan.planner.plan.node.PlanNodeId; import org.apache.iotdb.commons.request.IConsensusRequest; import org.apache.iotdb.consensus.ConsensusFactory; @@ -38,13 +40,28 @@ 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.exception.DataRegionException; +import org.apache.iotdb.db.exception.WriteProcessRejectException; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertMultiTabletsNode; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertRowsOfOneDeviceNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.InsertTabletNode; +import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.RelationalDeleteDataNode; +import org.apache.iotdb.db.queryengine.plan.statement.StatementTestUtils; +import org.apache.iotdb.db.schemaengine.table.DataNodeTableCache; 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.modification.DeletionPredicate; +import org.apache.iotdb.db.storageengine.dataregion.modification.ModificationFile; +import org.apache.iotdb.db.storageengine.dataregion.modification.TableDeletionEntry; +import org.apache.iotdb.db.storageengine.dataregion.modification.TagPredicate.NOP; import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResource; +import org.apache.iotdb.db.storageengine.dataregion.tsfile.TsFileResourceStatus; 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; @@ -53,6 +70,7 @@ 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.storageengine.rescon.memory.SystemInfo; import org.apache.iotdb.db.utils.EnvironmentUtils; import org.apache.iotdb.db.utils.constant.TestConstant; import org.apache.iotdb.rpc.TSStatusCode; @@ -62,6 +80,7 @@ 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.common.TimeRange; import org.apache.tsfile.read.expression.QueryExpression; import org.apache.tsfile.read.query.dataset.QueryDataSet; import org.apache.tsfile.write.schema.MeasurementSchema; @@ -73,8 +92,11 @@ import org.junit.rules.TemporaryFolder; import java.io.File; +import java.io.IOException; import java.lang.reflect.Field; +import java.util.Arrays; import java.util.Collections; +import java.util.List; import java.util.concurrent.CountDownLatch; import java.util.concurrent.ExecutorService; import java.util.concurrent.Executors; @@ -108,7 +130,7 @@ public class IoTConsensusShutdownTest { private NodeStatus previousNodeStatus; private String previousStatusReason; private boolean previousStopping; - private DataRegion dataRegion; + private TestDataRegion dataRegion; private IoTConsensusDataRegionStateMachine stateMachine; @Before @@ -125,12 +147,7 @@ public void setUp() throws Exception { commonConfig.setNodeStatus(NodeStatus.Running); commonConfig.setStopping(false); EnvironmentUtils.envSetUp(); - dataRegion = - new DataRegion( - TestConstant.OUTPUT_DATA_DIR + "shutdown-info", - "0", - new TsFileFlushPolicy.DirectFlushPolicy(), - DATABASE); + dataRegion = new TestDataRegion(); StorageEngine.getInstance().setDataRegion(new DataRegionId(0), dataRegion); stateMachine = new IoTConsensusDataRegionStateMachine(dataRegion); } @@ -273,18 +290,32 @@ public void testSyncLogBecomesReadOnlyAfterPreApplyCheck() throws Exception { @Test public void testShutdownDrainsAdmittedApplyBeforeClosingStorage() throws Exception { + checkShutdownDrainsAdmittedApply(false); + } + + @Test + public void testShutdownDrainsWalSeparatorBeforeClosingStorage() throws Exception { + checkShutdownDrainsAdmittedApply(true); + } + + private void checkShutdownDrainsAdmittedApply(final boolean delaySeparator) 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 InsertRowNode delayedRow = + doAnswer( + invocation -> { + admittedApply.countDown(); + assertTrue(resumeApply.await(30, TimeUnit.SECONDS)); + return invocation.callRealMethod(); + }) + .when(pendingRow); + if (delaySeparator) { + delayedRow.getRoutingEpoch(); + } else { + delayedRow.setLastFragment(true); + } final ExecutorService executor = Executors.newFixedThreadPool(2); final CountDownLatch shutdownStarted = new CountDownLatch(1); final CountDownLatch admissionClosed = new CountDownLatch(1); @@ -292,8 +323,7 @@ public void testShutdownDrainsAdmittedApplyBeforeClosingStorage() throws Excepti 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. + // This request has passed the storage admission check and holds insertLock. assertTrue(admittedApply.await(10, TimeUnit.SECONDS)); commonConfig.setStopping(true); commonConfig.setNodeStatus(NodeStatus.ReadOnly); @@ -383,7 +413,9 @@ protected void waitBeforeNextWriteRetry() throws InterruptedException { resumeRetry.countDown(); assertEquals( TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), write.get(10, TimeUnit.SECONDS).getCode()); - assertEquals(1, applyAttempts.get()); + // The retry enters the visitor, but its storage admission check rejects it before any WAL + // or processor can be created. + assertEquals(2, applyAttempts.get()); assertTrue(dataRegion.getWorkSequenceTsFileProcessors().isEmpty()); assertTrue(dataRegion.getWorkUnsequenceTsFileProcessors().isEmpty()); assertEquals(0, WALManager.getInstance().getWALNodesNum()); @@ -408,6 +440,236 @@ public void testReadOnlyCanBecomeWritableBeforeShutdown() throws Exception { assertRows(dataRegion.getSequenceFileList().get(0).getTsFile(), 1, 2); } + @Test + public void testWriteWaitingForInsertLockIsRejectedAfterShutdown() throws Exception { + writeAndAssertSuccess(1); + final ExecutorService executor = Executors.newSingleThreadExecutor(); + final AtomicReference writerThread = new AtomicReference<>(); + final Future write; + dataRegion.writeLock("testShutdownWhileWaitingForInsertLock"); + try { + write = + executor.submit( + () -> { + writerThread.set(Thread.currentThread()); + return stateMachine.write((IConsensusRequest) insertRow(2)); + }); + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .until( + () -> + writerThread.get() != null + && writerThread.get().getState() == Thread.State.WAITING); + // Close admission while the writer has already entered the visitor and is queued for + // insertLock. The stopping and read-only flags deliberately remain unset. + dataRegion.stopIoTConsensusWrites(); + } finally { + dataRegion.writeUnlock(); + executor.shutdown(); + } + try { + assertEquals( + TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), write.get(10, TimeUnit.SECONDS).getCode()); + dataRegion.syncCloseAllWorkingTsFileProcessors(); + assertRows(dataRegion.getSequenceFileList().get(0).getTsFile(), 1); + } finally { + executor.shutdownNow(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + @Test + public void testMemoryBackpressureDoesNotBlockShutdownOrFlush() throws Exception { + writeAndAssertSuccess(1); + final Field rejectedField = SystemInfo.class.getDeclaredField("rejected"); + rejectedField.setAccessible(true); + final boolean previousRejected = rejectedField.getBoolean(SystemInfo.getInstance()); + final ExecutorService executor = Executors.newFixedThreadPool(2); + final AtomicReference writerThread = new AtomicReference<>(); + rejectedField.setBoolean(SystemInfo.getInstance(), true); + final Future write = + executor.submit( + () -> { + writerThread.set(Thread.currentThread()); + return stateMachine.write((IConsensusRequest) insertRow(2)); + }); + try { + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .until( + () -> + writerThread.get() != null + && writerThread.get().getState() == Thread.State.TIMED_WAITING); + executor.submit(() -> dataRegion.stopIoTConsensusWrites()).get(10, TimeUnit.SECONDS); + assertFalse(write.isDone()); + // The same insertLock remains available to flush the memory that the writer is waiting for. + executor + .submit(() -> dataRegion.syncCloseAllWorkingTsFileProcessors()) + .get(10, TimeUnit.SECONDS); + assertEquals( + TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), write.get(10, TimeUnit.SECONDS).getCode()); + assertTrue(dataRegion.getWorkSequenceTsFileProcessors().isEmpty()); + assertRows(dataRegion.getSequenceFileList().get(0).getTsFile(), 1); + } finally { + rejectedField.setBoolean(SystemInfo.getInstance(), previousRejected); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + } + } + + @Test + public void testAllWritePathsRejectAfterStorageStop() throws Exception { + writeAndAssertSuccess(1); + final PlanNodeId id = new PlanNodeId("shutdown"); + final List writes = + Arrays.asList( + insertRow(2), + insertTablet(2), + new InsertRowsNode( + id, Collections.singletonList(0), Collections.singletonList(insertRow(2))), + new InsertRowsOfOneDeviceNode( + id, Collections.singletonList(0), Collections.singletonList(insertRow(2))), + new InsertMultiTabletsNode( + id, Collections.singletonList(0), Collections.singletonList(insertTablet(2))), + StatementTestUtils.genInsertRowNode(2), + StatementTestUtils.genInsertTabletNode(1, 2), + new DeleteDataNode( + id, Collections.singletonList(new MeasurementPath(DEVICE + ".s1")), 0, 10), + new DeleteDataNode( + id, Collections.singletonList(new MeasurementPath(DATABASE + ".**")), 0, 10), + tableDeletion()); + StorageEngine.getInstance().stop(); + WALManager.getInstance().stop(); + assertFalse(commonConfig.isReadOnly()); + assertFalse(commonConfig.isStopping()); + for (final PlanNode node : writes) { + assertEquals( + node.getClass().getSimpleName(), + TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), + stateMachine.write((IConsensusRequest) node).getCode()); + } + dataRegion.insertSeparatorToWAL(); + assertEquals(0, WALManager.getInstance().getWALNodesNum()); + assertTrue(dataRegion.getWorkSequenceTsFileProcessors().isEmpty()); + assertTrue(dataRegion.getWorkUnsequenceTsFileProcessors().isEmpty()); + assertRows(dataRegion.getSequenceFileList().get(0).getTsFile(), 1); + } + + @Test + public void testOtherConsensusAppliesRemainWritable() throws Exception { + final DataRegionStateMachine otherStateMachine = new DataRegionStateMachine(dataRegion); + commonConfig.setNodeStatus(NodeStatus.ReadOnly); + commonConfig.setStopping(true); + try { + long time = 0; + for (final String protocol : + Arrays.asList( + ConsensusFactory.RATIS_CONSENSUS, + ConsensusFactory.SIMPLE_CONSENSUS, + ConsensusFactory.IOT_CONSENSUS_V2)) { + config.setDataRegionConsensusProtocolClass(protocol); + assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + otherStateMachine.write((IConsensusRequest) insertRow(++time)).getCode()); + } + } finally { + config.setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS); + commonConfig.setNodeStatus(NodeStatus.Running); + commonConfig.setStopping(false); + } + dataRegion.syncCloseAllWorkingTsFileProcessors(); + assertRows(dataRegion.getSequenceFileList().get(0).getTsFile(), 1, 2, 3); + } + + @Test + public void testShutdownDrainsTableDeletionOutsideInsertLock() throws Exception { + checkShutdownDrainsTableDeletion(false); + } + + @Test + public void testFailedTableDeletionDoesNotBlockShutdown() throws Exception { + checkShutdownDrainsTableDeletion(true); + } + + private void checkShutdownDrainsTableDeletion(final boolean failDeletion) throws Exception { + DataNodeTableCache.getInstance() + .preUpdateTable(DATABASE, StatementTestUtils.genTsTable(), null); + DataNodeTableCache.getInstance() + .commitUpdateTable(DATABASE, StatementTestUtils.tableName(), null); + assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + stateMachine + .write((IConsensusRequest) StatementTestUtils.genInsertTabletNode(2, 0)) + .getCode()); + dataRegion.syncCloseAllWorkingTsFileProcessors(); + final TsFileResource resource = dataRegion.getSequenceFileList().get(0); + assertTrue(resource.transformStatus(TsFileResourceStatus.COMPACTION_CANDIDATE)); + assertTrue(resource.transformStatus(TsFileResourceStatus.COMPACTING)); + final ModificationFile modFile = spy(resource.getModFileForWrite()); + resource.setExclusiveModFile(modFile); + final CountDownLatch modFileClosing = new CountDownLatch(1); + final CountDownLatch resumeDeletion = new CountDownLatch(1); + doAnswer( + invocation -> { + modFileClosing.countDown(); + assertTrue(resumeDeletion.await(30, TimeUnit.SECONDS)); + if (failDeletion) { + throw new IOException(); + } + return invocation.callRealMethod(); + }) + .when(modFile) + .close(); + final ExecutorService executor = Executors.newFixedThreadPool(2); + final AtomicReference shutdownThread = new AtomicReference<>(); + final Future deletion = + executor.submit(() -> stateMachine.write((IConsensusRequest) tableDeletion())); + try { + assertTrue(modFileClosing.await(10, TimeUnit.SECONDS)); + // Mod-file I/O still runs without insertLock, so an ordinary flush can proceed. + executor.submit(() -> dataRegion.timedFlushSeqMemTable()).get(10, TimeUnit.SECONDS); + final Future shutdown = + executor.submit( + () -> { + shutdownThread.set(Thread.currentThread()); + commonConfig.setStopping(true); + StorageEngine.getInstance().stop(); + WALManager.getInstance().stop(); + }); + Awaitility.await() + .atMost(10, TimeUnit.SECONDS) + .until( + () -> + shutdownThread.get() != null + && shutdownThread.get().getState() == Thread.State.WAITING); + assertFalse(shutdown.isDone()); + assertEquals( + TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(), + stateMachine.write((IConsensusRequest) insertRow(3)).getCode()); + resumeDeletion.countDown(); + assertEquals( + (failDeletion ? TSStatusCode.WRITE_PROCESS_ERROR : TSStatusCode.SUCCESS_STATUS) + .getStatusCode(), + deletion.get(10, TimeUnit.SECONDS).getCode()); + shutdown.get(10, TimeUnit.SECONDS); + assertEquals(0, WALManager.getInstance().getWALNodesNum()); + } finally { + resumeDeletion.countDown(); + executor.shutdownNow(); + assertTrue(executor.awaitTermination(10, TimeUnit.SECONDS)); + doAnswer(invocation -> invocation.callRealMethod()).when(modFile).close(); + modFile.close(); + } + } + + private static RelationalDeleteDataNode tableDeletion() { + return new RelationalDeleteDataNode( + new PlanNodeId("shutdown"), + new TableDeletionEntry( + new DeletionPredicate(StatementTestUtils.tableName(), new NOP()), new TimeRange(0, 10)), + DATABASE); + } + private IoTConsensusServerImpl newServer( IoTConsensusDataRegionStateMachine targetStateMachine, ScheduledExecutorService backgroundTaskService, @@ -549,9 +811,48 @@ private void writeAndAssertSuccess(long time) throws Exception { } private void writeUncheckedAndAssertSuccess(long time) throws Exception { - assertEquals( - TSStatusCode.SUCCESS_STATUS.getStatusCode(), - insertRow(time).accept(new DataExecutionVisitor(), dataRegion).getCode()); + dataRegion.bypassWriteChecks = true; + try { + assertEquals( + TSStatusCode.SUCCESS_STATUS.getStatusCode(), + insertRow(time).accept(new DataExecutionVisitor(), dataRegion).getCode()); + } finally { + dataRegion.bypassWriteChecks = false; + } + } + + private static class TestDataRegion extends DataRegion { + // Only the two fault-injection tests bypass admission to reproduce the old late-write bug. + private boolean bypassWriteChecks; + + private TestDataRegion() throws DataRegionException { + super( + TestConstant.OUTPUT_DATA_DIR + "shutdown-info", + "0", + new TsFileFlushPolicy.DirectFlushPolicy(), + DATABASE); + } + + @Override + protected void checkIoTConsensusWriteAllowed() throws WriteProcessRejectException { + if (!bypassWriteChecks) { + super.checkIoTConsensusWriteAllowed(); + } + } + } + + private static InsertTabletNode insertTablet(final long time) throws Exception { + return new InsertTabletNode( + new PlanNodeId("shutdown"), + new PartialPath(DEVICE), + false, + new String[] {MEASUREMENT}, + new TSDataType[] {TSDataType.INT32}, + new MeasurementSchema[] {new MeasurementSchema(MEASUREMENT, TSDataType.INT32)}, + new long[] {time}, + null, + new Object[] {new int[] {(int) time * 10}}, + 1); } private static InsertRowNode insertRow(long time) throws Exception { diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithFastPerformerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithFastPerformerTest.java index 4c283fad63679..fc882ad01a119 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithFastPerformerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithFastPerformerTest.java @@ -25,6 +25,7 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.DataRegionException; import org.apache.iotdb.db.exception.StorageEngineException; +import org.apache.iotdb.db.exception.WriteProcessRejectException; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; import org.apache.iotdb.db.storageengine.buffer.BloomFilterCache; import org.apache.iotdb.db.storageengine.buffer.ChunkCache; @@ -1112,7 +1113,7 @@ public void testAppendChunk() throws Exception { @Test public void testCompactionWithDeletionsDuringCompactions() - throws MetadataException, IOException, DataRegionException { + throws MetadataException, IOException, DataRegionException, WriteProcessRejectException { // create source seq files List sourceResources = new ArrayList<>(); List> chunkPagePointsNum = new ArrayList<>(); diff --git a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithReadChunkPerformerTest.java b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithReadChunkPerformerTest.java index 875239ce6b6ea..4454160d99eb9 100644 --- a/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithReadChunkPerformerTest.java +++ b/iotdb-core/datanode/src/test/java/org/apache/iotdb/db/storageengine/dataregion/compaction/inner/InnerSeqCompactionWithReadChunkPerformerTest.java @@ -25,6 +25,7 @@ import org.apache.iotdb.db.conf.IoTDBDescriptor; import org.apache.iotdb.db.exception.DataRegionException; import org.apache.iotdb.db.exception.StorageEngineException; +import org.apache.iotdb.db.exception.WriteProcessRejectException; import org.apache.iotdb.db.queryengine.plan.planner.plan.node.write.DeleteDataNode; import org.apache.iotdb.db.storageengine.buffer.BloomFilterCache; import org.apache.iotdb.db.storageengine.buffer.ChunkCache; @@ -1074,7 +1075,11 @@ public void testAppendChunk() throws Exception { @Test public void testCompactionWithDeletionsDuringCompactions() - throws MetadataException, IOException, DataRegionException, InterruptedException { + throws MetadataException, + IOException, + DataRegionException, + InterruptedException, + WriteProcessRejectException { // create source seq files List sourceResources = new ArrayList<>(); List> chunkPagePointsNum = new ArrayList<>();