This is an automated email from the ASF dual-hosted git repository.
jt2594838 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new d2b97bc8455 Reuse DataRegion insert lock for IoTConsensus shutdown
fencing (#18893)
d2b97bc8455 is described below
commit d2b97bc8455710a4540df03a1f15192d276629f2
Author: Caideyipi <[email protected]>
AuthorDate: Thu Oct 15 18:41:06 2026 +0800
Reuse DataRegion insert lock for IoTConsensus shutdown fencing (#18893)
---
.../dataregion/DataExecutionVisitor.java | 10 +-
.../dataregion/DataRegionStateMachine.java | 6 +-
.../IoTConsensusDataRegionStateMachine.java | 7 -
.../db/storageengine/dataregion/DataRegion.java | 129 +++++---
.../dataregion/IoTConsensusShutdownTest.java | 343 +++++++++++++++++++--
.../InnerSeqCompactionWithFastPerformerTest.java | 3 +-
...nerSeqCompactionWithReadChunkPerformerTest.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 96bbb6ccc16..c0034ccc77c 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 class DataExecutionVisitor implements
PlanVisitor<TSStatus, DataRegion> {
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 class DataExecutionVisitor implements
PlanVisitor<TSStatus, DataRegion> {
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 class DataExecutionVisitor implements
PlanVisitor<TSStatus, 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 class DataExecutionVisitor implements
PlanVisitor<TSStatus, 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 class DataExecutionVisitor implements
PlanVisitor<TSStatus, DataRegion> {
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 class DataExecutionVisitor implements
PlanVisitor<TSStatus, 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 class DataExecutionVisitor implements
PlanVisitor<TSStatus, DataRegion> {
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 127169ee2f8..3d2fb3d2006 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 @@ public class DataRegionStateMachine extends
BaseStateMachine {
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 @@ public class DataRegionStateMachine extends
BaseStateMachine {
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 e9275f15587..f3fecdf69f3 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 class IoTConsensusDataRegionStateMachine extends
DataRegionStateMachine {
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 1ac4d80b94f..2dcb699807d 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.atomic.AtomicLong;
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 class DataRegion implements IDataRegionForQuery {
writeLock("InsertRow");
PERFORMANCE_OVERVIEW_METRICS.recordScheduleLockCost(System.nanoTime() -
startTime);
try {
+ checkIoTConsensusWriteAllowed();
if (deleted) {
return;
}
@@ -1254,6 +1255,7 @@ public class DataRegion implements IDataRegionForQuery {
PERFORMANCE_OVERVIEW_METRICS.recordScheduleUpdateLastCacheCost(
System.nanoTime() - startTime);
}
+ logSeparatorToWAL(insertRowNode);
} finally {
writeUnlock();
}
@@ -1432,6 +1434,7 @@ public class DataRegion implements IDataRegionForQuery {
writeLock("insertTablet");
PERFORMANCE_OVERVIEW_METRICS.recordScheduleLockCost(System.nanoTime() -
startTime);
try {
+ checkIoTConsensusWriteAllowed();
if (deleted) {
logger.info(
StorageEngineMessages
@@ -1460,6 +1463,7 @@ public class DataRegion implements IDataRegionForQuery {
if (!noFailure) {
throw new BatchProcessException(results);
}
+ logSeparatorToWAL(insertTabletNode);
} finally {
writeUnlock();
}
@@ -2540,27 +2544,27 @@ public class DataRegion implements IDataRegionForQuery {
WritingMetrics.getInstance().recordTimedFlushMemTableCount(count);
}
- public TSStatus executeIoTConsensusWrite(Supplier<TSStatus> 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 @@ public class DataRegion implements IDataRegionForQuery {
}
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 class DataRegion implements IDataRegionForQuery {
boolean hasReleasedLock = false;
try {
+ checkIoTConsensusWriteAllowed();
if (deleted) {
return;
}
@@ -3373,6 +3378,8 @@ public class DataRegion implements IDataRegionForQuery {
hasReleasedLock = true;
deleteDataInSealedFiles(sealedTsFileResource, deletion);
+ } catch (WriteProcessRejectException e) {
+ throw e;
} catch (Exception e) {
throw new IOException(e);
} finally {
@@ -3387,7 +3394,8 @@ public class DataRegion implements IDataRegionForQuery {
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 class DataRegion implements IDataRegionForQuery {
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 class DataRegion implements IDataRegionForQuery {
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 class DataRegion implements IDataRegionForQuery {
}
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 class DataRegion implements IDataRegionForQuery {
boolean releasedLock = false;
try {
+ checkIoTConsensusWriteAllowed();
if (deleted) {
return;
}
@@ -3556,6 +3585,8 @@ public class DataRegion implements IDataRegionForQuery {
writeUnlock();
releasedLock = true;
deleteDataDirectlyInFile(sealedTsFileResource, deletion);
+ } catch (WriteProcessRejectException e) {
+ throw e;
} catch (Exception e) {
throw new IOException(e);
} finally {
@@ -3720,25 +3751,34 @@ public class DataRegion implements IDataRegionForQuery {
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 class DataRegion implements IDataRegionForQuery {
writeLock("InsertRowsOfOneDevice");
PERFORMANCE_OVERVIEW_METRICS.recordScheduleLockCost(System.nanoTime() -
startTime);
try {
+ checkIoTConsensusWriteAllowed();
if (deleted) {
return;
}
@@ -5125,13 +5166,14 @@ public class DataRegion implements IDataRegionForQuery {
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 class DataRegion implements IDataRegionForQuery {
writeLock("InsertRows");
PERFORMANCE_OVERVIEW_METRICS.recordScheduleLockCost(System.nanoTime() -
startTime);
try {
+ checkIoTConsensusWriteAllowed();
if (deleted) {
return;
}
@@ -5211,6 +5254,7 @@ public class DataRegion implements IDataRegionForQuery {
if (!insertRowsNode.getResults().isEmpty()) {
throw new
BatchProcessException(StorageEngineMessages.PARTIAL_FAILED_INSERTING_ROWS);
}
+ logSeparatorToWAL(insertRowsNode);
} finally {
writeUnlock();
}
@@ -5229,6 +5273,7 @@ public class DataRegion implements IDataRegionForQuery {
writeLock("insertTablets");
PERFORMANCE_OVERVIEW_METRICS.recordScheduleLockCost(System.nanoTime() -
startTime);
try {
+ checkIoTConsensusWriteAllowed();
if (deleted) {
logger.info(
StorageEngineMessages
@@ -5282,14 +5327,14 @@ public class DataRegion implements IDataRegionForQuery {
}
}
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 fed7a6151a1..bf2f5c21e6a 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.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.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.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.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.WALRecoverManage
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.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.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.Test;
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 class IoTConsensusShutdownTest {
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 class IoTConsensusShutdownTest {
@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 class IoTConsensusShutdownTest {
final Future<TSStatus> 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 @@ public class IoTConsensusShutdownTest {
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 class IoTConsensusShutdownTest {
assertRows(dataRegion.getSequenceFileList().get(0).getTsFile(), 1, 2);
}
+ @Test
+ public void testWriteWaitingForInsertLockIsRejectedAfterShutdown() throws
Exception {
+ writeAndAssertSuccess(1);
+ final ExecutorService executor = Executors.newSingleThreadExecutor();
+ final AtomicReference<Thread> writerThread = new AtomicReference<>();
+ final Future<TSStatus> 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<Thread> writerThread = new AtomicReference<>();
+ rejectedField.setBoolean(SystemInfo.getInstance(), true);
+ final Future<TSStatus> 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<PlanNode> 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<Thread> shutdownThread = new AtomicReference<>();
+ final Future<TSStatus> 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 @@ public class IoTConsensusShutdownTest {
}
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 4c283fad636..fc882ad01a1 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.commons.queryengine.plan.planner.plan.node.PlanNodeId;
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 class InnerSeqCompactionWithFastPerformerTest {
@Test
public void testCompactionWithDeletionsDuringCompactions()
- throws MetadataException, IOException, DataRegionException {
+ throws MetadataException, IOException, DataRegionException,
WriteProcessRejectException {
// create source seq files
List<TsFileResource> sourceResources = new ArrayList<>();
List<List<Long>> 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 875239ce6b6..4454160d99e 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.commons.queryengine.plan.planner.plan.node.PlanNodeId;
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 class
InnerSeqCompactionWithReadChunkPerformerTest {
@Test
public void testCompactionWithDeletionsDuringCompactions()
- throws MetadataException, IOException, DataRegionException,
InterruptedException {
+ throws MetadataException,
+ IOException,
+ DataRegionException,
+ InterruptedException,
+ WriteProcessRejectException {
// create source seq files
List<TsFileResource> sourceResources = new ArrayList<>();
List<List<Long>> chunkPagePointsNum = new ArrayList<>();
