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 532d6c13add Fence IoTConsensus applies before storage shutdown (#18841)
532d6c13add is described below
commit 532d6c13addc04946535729db388fb4c18feb6b9
Author: Caideyipi <[email protected]>
AuthorDate: Thu Oct 15 16:13:07 2026 +0800
Fence IoTConsensus applies before storage shutdown (#18841)
* Fix queued IoTConsensus writes during shutdown
* Test queued IoTConsensus writes across storage shutdown
* Reproduce WAL checkpoint loss from writes after shutdown
* Fence IoTConsensus applies before storage shutdown
---
.../manual/enhanced/IoTDBPipeClusterIT.java | 41 +-
.../auto/enhanced/IoTDBPipeClusterIT.java | 38 +-
.../consensus/iot/IoTConsensusServerImpl.java | 9 +-
.../consensus/iot/IoTConsensusServerImplTest.java | 94 ++++
.../dataregion/DataRegionStateMachine.java | 6 +-
.../IoTConsensusDataRegionStateMachine.java | 7 +
.../iotdb/db/service/DataNodeShutdownHook.java | 3 +
.../iotdb/db/storageengine/StorageEngine.java | 13 +
.../db/storageengine/dataregion/DataRegion.java | 32 ++
.../dataregion/IoTConsensusShutdownTest.java | 593 +++++++++++++++++++++
10 files changed, 802 insertions(+), 34 deletions(-)
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java
index a5e505f4fd7..cde83cc4da6 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 class IoTDBPipeClusterIT extends
AbstractPipeTableModelDualManualIT {
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<String, String> sourceAttributes = new HashMap<>();
final Map<String, String> processorAttributes = new HashMap<>();
@@ -449,6 +452,8 @@ public class IoTDBPipeClusterIT extends
AbstractPipeTableModelDualManualIT {
private void waitForTableDataRegionReplicationComplete(
final BaseEnv env, final List<String> databases) {
+ // IoTConsensus accepts writes on any replica, including a previous leader
during failover.
+ // Check every replica's outgoing replication before flushing or
restarting the cluster.
await()
.pollInterval(500, TimeUnit.MILLISECONDS)
.atMost(2, TimeUnit.MINUTES)
@@ -456,24 +461,23 @@ public class IoTDBPipeClusterIT extends
AbstractPipeTableModelDualManualIT {
() -> {
try (final SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient)
env.getLeaderConfigNodeConnection()) {
- final List<TRegionInfo> leaderRegionInfoList =
- showTableDataRegionLeaders(databases, client);
+ final List<TRegionInfo> regionInfoList =
showTableDataRegions(databases, client);
Assert.assertFalse(
- "No table DataRegion leader found for databases " +
databases,
- leaderRegionInfoList.isEmpty());
+ "No table DataRegion found for databases " + databases,
+ regionInfoList.isEmpty());
- for (final TRegionInfo regionInfo : leaderRegionInfoList) {
- final DataNodeWrapper leaderNode =
+ for (final TRegionInfo regionInfo : regionInfoList) {
+ final DataNodeWrapper replicaNode =
findDataNodeWrapperByPort(env,
regionInfo.getClientRpcPort());
final String metricsUrl =
"http://";
- + leaderNode.getIp()
+ + replicaNode.getIp()
+ ":"
- + leaderNode.getMetricPort()
+ + replicaNode.getMetricPort()
+ "/metrics";
final String metricsContent = env.getUrlContent(metricsUrl,
null);
Assert.assertNotNull(
- "Failed to fetch metrics from leader DataNode at " +
metricsUrl,
+ "Failed to fetch metrics from replica DataNode at " +
metricsUrl,
metricsContent);
assertSyncLagIsZero(metricsContent,
buildDataRegionTag(regionInfo), metricsUrl);
}
@@ -483,6 +487,17 @@ public class IoTDBPipeClusterIT extends
AbstractPipeTableModelDualManualIT {
private List<TRegionInfo> showTableDataRegionLeaders(
final List<String> databases, final SyncConfigNodeIServiceClient client)
throws TException {
+ final List<TRegionInfo> result = new ArrayList<>();
+ for (final TRegionInfo regionInfo : showTableDataRegions(databases,
client)) {
+ if
(RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) {
+ result.add(regionInfo);
+ }
+ }
+ return result;
+ }
+
+ private List<TRegionInfo> showTableDataRegions(
+ final List<String> databases, final SyncConfigNodeIServiceClient client)
throws TException {
final TShowRegionResp showRegionResp =
client.showRegion(
new TShowRegionReq()
@@ -491,13 +506,7 @@ public class IoTDBPipeClusterIT extends
AbstractPipeTableModelDualManualIT {
.setIsTableModel(true));
Assert.assertEquals(
TSStatusCode.SUCCESS_STATUS.getStatusCode(),
showRegionResp.getStatus().getCode());
- final List<TRegionInfo> result = new ArrayList<>();
- for (final TRegionInfo regionInfo : showRegionResp.getRegionInfoList()) {
- if
(RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) {
- result.add(regionInfo);
- }
- }
- return result;
+ return showRegionResp.getRegionInfoList();
}
private DataNodeWrapper findDataNodeWrapperByPort(final BaseEnv env, final
int port) {
diff --git
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java
index 529147a77b0..b64a43fc04d 100644
---
a/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java
+++
b/integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java
@@ -458,6 +458,8 @@ public class IoTDBPipeClusterIT extends
AbstractPipeDualTreeModelAutoIT {
private void waitForTreeDataRegionReplicationComplete(
final BaseEnv env, final List<String> databases) {
+ // IoTConsensus accepts writes on any replica, including a previous leader
during failover.
+ // Check every replica's outgoing replication before flushing or
restarting the cluster.
await()
.pollInterval(500, TimeUnit.MILLISECONDS)
.atMost(2, TimeUnit.MINUTES)
@@ -465,24 +467,23 @@ public class IoTDBPipeClusterIT extends
AbstractPipeDualTreeModelAutoIT {
() -> {
try (final SyncConfigNodeIServiceClient client =
(SyncConfigNodeIServiceClient)
env.getLeaderConfigNodeConnection()) {
- final List<TRegionInfo> leaderRegionInfoList =
- showTreeDataRegionLeaders(databases, client);
+ final List<TRegionInfo> regionInfoList =
showTreeDataRegions(databases, client);
Assert.assertFalse(
- "No tree DataRegion leader found for databases " +
databases,
- leaderRegionInfoList.isEmpty());
+ "No tree DataRegion found for databases " + databases,
+ regionInfoList.isEmpty());
- for (final TRegionInfo regionInfo : leaderRegionInfoList) {
- final DataNodeWrapper leaderNode =
+ for (final TRegionInfo regionInfo : regionInfoList) {
+ final DataNodeWrapper replicaNode =
findDataNodeWrapperByPort(env,
regionInfo.getClientRpcPort());
final String metricsUrl =
"http://";
- + leaderNode.getIp()
+ + replicaNode.getIp()
+ ":"
- + leaderNode.getMetricPort()
+ + replicaNode.getMetricPort()
+ "/metrics";
final String metricsContent = env.getUrlContent(metricsUrl,
null);
Assert.assertNotNull(
- "Failed to fetch metrics from leader DataNode at " +
metricsUrl,
+ "Failed to fetch metrics from replica DataNode at " +
metricsUrl,
metricsContent);
assertSyncLagIsZero(metricsContent,
buildDataRegionTag(regionInfo), metricsUrl);
}
@@ -492,6 +493,17 @@ public class IoTDBPipeClusterIT extends
AbstractPipeDualTreeModelAutoIT {
private List<TRegionInfo> showTreeDataRegionLeaders(
final List<String> databases, final SyncConfigNodeIServiceClient client)
throws TException {
+ final List<TRegionInfo> result = new ArrayList<>();
+ for (final TRegionInfo regionInfo : showTreeDataRegions(databases,
client)) {
+ if
(RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) {
+ result.add(regionInfo);
+ }
+ }
+ return result;
+ }
+
+ private List<TRegionInfo> showTreeDataRegions(
+ final List<String> databases, final SyncConfigNodeIServiceClient client)
throws TException {
final TShowRegionResp showRegionResp =
client.showRegion(
new TShowRegionReq()
@@ -499,13 +511,7 @@ public class IoTDBPipeClusterIT extends
AbstractPipeDualTreeModelAutoIT {
.setDatabases(databases));
Assert.assertEquals(
TSStatusCode.SUCCESS_STATUS.getStatusCode(),
showRegionResp.getStatus().getCode());
- final List<TRegionInfo> result = new ArrayList<>();
- for (final TRegionInfo regionInfo : showRegionResp.getRegionInfoList()) {
- if
(RegionRoleType.Leader.getRoleType().equals(regionInfo.getRoleType())) {
- result.add(regionInfo);
- }
- }
- return result;
+ return showRegionResp.getRegionInfoList();
}
private DataNodeWrapper findDataNodeWrapperByPort(final BaseEnv env, final
int port) {
diff --git
a/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
b/iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java
index 8e70866e379..ca29931f1c9 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
@@ -1617,7 +1617,14 @@ public class IoTConsensusServerImpl {
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 cc33cf1573a..291a8d99d1b 100644
---
a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java
+++
b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java
@@ -20,15 +20,21 @@
package org.apache.iotdb.consensus.iot;
import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.audit.UserDataTransferAuditHandler;
import org.apache.iotdb.commons.consensus.ConsensusGroupId;
import org.apache.iotdb.commons.consensus.DataRegionId;
import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType;
import org.apache.iotdb.commons.request.IConsensusRequest;
import org.apache.iotdb.consensus.common.Peer;
+import
org.apache.iotdb.consensus.common.request.DeserializedBatchIndexedConsensusRequest;
+import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
import org.apache.iotdb.consensus.config.IoTConsensusConfig;
import org.apache.iotdb.consensus.config.UserDataTransferAuditClassifier;
+import org.apache.iotdb.consensus.i18n.ConsensusMessages;
+import org.apache.iotdb.consensus.iot.util.TestEntry;
import org.apache.iotdb.consensus.iot.util.TestStateMachine;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.junit.Rule;
import org.junit.Test;
@@ -46,6 +52,8 @@ import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertFalse;
@@ -59,6 +67,92 @@ public class IoTConsensusServerImplTest {
@Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+ @Test
+ public void testSyncLogBecomesReadOnlyWhileWaiting() throws Exception {
+ final ScheduledExecutorService backgroundTaskService =
+ Executors.newSingleThreadScheduledExecutor();
+ final AtomicBoolean readOnly = new AtomicBoolean();
+ final TestStateMachine stateMachine =
+ new TestStateMachine() {
+ @Override
+ public boolean isReadOnly() {
+ return readOnly.get();
+ }
+ };
+ final Peer thisNode = new Peer(new DataRegionId(1), 0, new
TEndPoint("127.0.0.1", 6667));
+ final IoTConsensusServerImpl server =
+ new IoTConsensusServerImpl(
+ temporaryFolder.getRoot().getAbsolutePath(),
+ null,
+ DirectoryStrategyType.SEQUENCE_STRATEGY,
+ thisNode,
+ Collections.singletonList(thisNode),
+ stateMachine,
+ backgroundTaskService,
+ null,
+ null,
+ IoTConsensusConfig.newBuilder()
+ .setReplication(
+ IoTConsensusConfig.Replication.newBuilder()
+ .setMaxWaitingTimeForWaitBatchInMs(100)
+ .build())
+ .build());
+ final DeserializedBatchIndexedConsensusRequest request =
+ new DeserializedBatchIndexedConsensusRequest(1, 2, 2, 1, 2);
+ for (int index = 1; index <= 2; index++) {
+ request.add(
+ new IndexedConsensusRequest(
+ index, index, Collections.singletonList(new TestEntry(index,
thisNode))));
+ }
+ final AtomicReference<TSStatus> status = new AtomicReference<>();
+ final Thread syncLogThread = new Thread(() -> status.set(server.syncLog(1,
request)));
+ try {
+ assertFalse(server.isReadOnly());
+ syncLogThread.start();
+ final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+ while (syncLogThread.getState() != Thread.State.TIMED_WAITING
+ && System.nanoTime() < deadline) {
+ Thread.sleep(10);
+ }
+ assertEquals(Thread.State.TIMED_WAITING, syncLogThread.getState());
+ readOnly.set(true);
+ syncLogThread.join(TimeUnit.SECONDS.toMillis(15));
+ assertFalse(syncLogThread.isAlive());
+ assertEquals(2, status.get().getSubStatusSize());
+ assertEquals(
+ TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(),
+ status.get().getSubStatus().get(0).getCode());
+ assertEquals(
+ ConsensusMessages.SYNC_LOG_SYSTEM_READ_ONLY,
+ status.get().getSubStatus().get(0).getMessage());
+ assertEquals(
+ TSStatusCode.WRITE_PROCESS_REJECT.getStatusCode(),
+ status.get().getSubStatus().get(1).getCode());
+ assertTrue(stateMachine.getRequestSet().isEmpty());
+ assertEquals(0, server.getEffectiveSafePhysicalTime(1));
+
+ // The sender must be able to retry the complete batch after the replica
becomes writable.
+ readOnly.set(false);
+ server.reloadConsensusConfig(
+ IoTConsensusConfig.newBuilder()
+ .setReplication(
+ IoTConsensusConfig.Replication.newBuilder()
+ .setMaxPendingBatchesNum(1)
+ .setMaxWaitingTimeForWaitBatchInMs(100)
+ .build())
+ .build());
+ final TSStatus retryStatus = server.syncLog(1, request);
+ assertTrue(
+ retryStatus.getSubStatus().stream()
+ .allMatch(s -> s.getCode() ==
TSStatusCode.SUCCESS_STATUS.getStatusCode()));
+ assertEquals(2, stateMachine.getRequestSet().size());
+ } finally {
+ syncLogThread.interrupt();
+ syncLogThread.join(TimeUnit.SECONDS.toMillis(5));
+ backgroundTaskService.shutdownNow();
+ }
+ }
+
@Test
public void testSnapshotTransferAppliesGroupClassifier() {
final DataRegionId groupId = new DataRegionId(1);
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 3d2fb3d2006..127169ee2f8 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 = 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 @@ 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 f3fecdf69f3..e9275f15587 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 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/service/DataNodeShutdownHook.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java
index 828c4b811ac..cb1095da104 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 class DataNodeShutdownHook extends Thread {
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 72bdf195998..2f9c6efa6f4 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
@@ -435,6 +435,7 @@ public class StorageEngine implements IService {
@Override
public void stop() {
+ stopIoTConsensusWrites();
loadTsFileManager.stop();
for (DataRegion dataRegion : dataRegionMap.values()) {
if (dataRegion != null) {
@@ -455,6 +456,7 @@ public class StorageEngine implements IService {
@Override
public void shutdown(long milliseconds) throws ShutdownException {
+ stopIoTConsensusWrites();
loadTsFileManager.stop();
try {
for (DataRegion dataRegion : dataRegionMap.values()) {
@@ -526,6 +528,17 @@ public class StorageEngine implements IService {
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 46952ee87fa..1ac4d80b94f 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.DiskSpaceInsufficientException;
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;
@@ -233,6 +234,7 @@ 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;
@@ -285,6 +287,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();
@@ -2532,6 +2540,30 @@ 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();
+ }
+ }
+
+ /** 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 00000000000..fed7a6151a1
--- /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<TSStatus> 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<TSStatus> 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<Thread> shutdownThread = new AtomicReference<>();
+ 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.
+ 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<TSStatus> 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());
+ }
+ }
+}
