Fence IoTConsensus applies before storage shutdown

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/tablemodel/manual/enhanced/IoTDBPipeClusterIT.java

@@ -348,6 +348,9 @@ public void testPipeAfterDataRegionLeaderStop() {
 
         try (final SyncConfigNodeIServiceClient client =
             (SyncConfigNodeIServiceClient) senderEnv.getLeaderConfigNodeConnection()) {
+          // Distinguish a source-side write loss during the restart from a p2 transfer failure.
+          TableModelUtils.assertData("test1", "test1", 0, 300, senderEnv);
+
           // Create a new pipe and write new data
           final Map<String, String> sourceAttributes = new HashMap<>();
           final Map<String, String> processorAttributes = new HashMap<>();
@@ -449,31 +452,32 @@ private void flushTableDataRegionReplicasAfterReplicationComplete(
 
   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)
         .untilAsserted(
             () -> {
               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 @@ private void waitForTableDataRegionReplicationComplete(
 
   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 @@ private List<TRegionInfo> showTableDataRegionLeaders(
                 .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) {

integration-test/src/test/java/org/apache/iotdb/pipe/it/dual/treemodel/auto/enhanced/IoTDBPipeClusterIT.java

@@ -458,31 +458,32 @@ private void flushTreeDataRegionReplicasAfterReplicationComplete(
 
   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)
         .untilAsserted(
             () -> {
               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,20 +493,25 @@ private void waitForTreeDataRegionReplicationComplete(
 
   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()
                 .setConsensusGroupType(TConsensusGroupType.DataRegion)
                 .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) {

iotdb-core/consensus/src/main/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImpl.java

@@ -1583,7 +1583,14 @@ private TSStatus cacheAndInsertLatestNode(DeserializedBatchIndexedConsensusReque
         for (int i = 0; i < insertNodes.size(); i++) {
           IConsensusRequest insertNode = insertNodes.get(i);
           insertNode.markAsGeneratedByRemoteConsensusLeader();
-          TSStatus status = stateMachine.write(insertNode);
+          // Reject requests that became read-only while queued. This is an early rejection only:
+          // the state machine must also fence the full apply against storage shutdown, since the
+          // replica can become read-only after this check.
+          TSStatus status =
+              isReadOnly()
+                  ? RpcUtils.getStatus(
+                      TSStatusCode.SYSTEM_READ_ONLY, ConsensusMessages.SYNC_LOG_SYSTEM_READ_ONLY)
+                  : stateMachine.write(insertNode);
           subStatus.add(status);
           if (RetryUtils.needRetryForWrite(status.getCode())) {
             for (int j = i + 1; j < insertNodes.size(); j++) {

iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/IoTConsensusServerImplTest.java

@@ -20,15 +20,21 @@
 package org.apache.iotdb.consensus.iot;
 
 import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
 import org.apache.iotdb.commons.audit.UserDataTransferAuditHandler;
 import org.apache.iotdb.commons.consensus.ConsensusGroupId;
 import org.apache.iotdb.commons.consensus.DataRegionId;
 import org.apache.iotdb.commons.disk.strategy.DirectoryStrategyType;
 import org.apache.iotdb.commons.request.IConsensusRequest;
 import org.apache.iotdb.consensus.common.Peer;
+import org.apache.iotdb.consensus.common.request.DeserializedBatchIndexedConsensusRequest;
+import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
 import org.apache.iotdb.consensus.config.IoTConsensusConfig;
 import org.apache.iotdb.consensus.config.UserDataTransferAuditClassifier;
+import org.apache.iotdb.consensus.i18n.ConsensusMessages;
+import org.apache.iotdb.consensus.iot.util.TestEntry;
 import org.apache.iotdb.consensus.iot.util.TestStateMachine;
+import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.junit.Rule;
 import org.junit.Test;
@@ -46,6 +52,8 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -59,6 +67,92 @@ public class IoTConsensusServerImplTest {
 
   @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
 
+  @Test
+  public void testSyncLogBecomesReadOnlyWhileWaiting() throws Exception {
+    final ScheduledExecutorService backgroundTaskService =
+        Executors.newSingleThreadScheduledExecutor();
+    final AtomicBoolean readOnly = new AtomicBoolean();
+    final TestStateMachine stateMachine =
+        new TestStateMachine() {
+          @Override
+          public boolean isReadOnly() {
+            return readOnly.get();
+          }
+        };
+    final Peer thisNode = new Peer(new DataRegionId(1), 0, new TEndPoint("127.0.0.1", 6667));
+    final IoTConsensusServerImpl server =
+        new IoTConsensusServerImpl(
+            temporaryFolder.getRoot().getAbsolutePath(),
+            null,
+            DirectoryStrategyType.SEQUENCE_STRATEGY,
+            thisNode,
+            Collections.singletonList(thisNode),
+            stateMachine,
+            backgroundTaskService,
+            null,
+            null,
+            IoTConsensusConfig.newBuilder()
+                .setReplication(
+                    IoTConsensusConfig.Replication.newBuilder()
+                        .setMaxWaitingTimeForWaitBatchInMs(100)
+                        .build())
+                .build());
+    final DeserializedBatchIndexedConsensusRequest request =
+        new DeserializedBatchIndexedConsensusRequest(1, 2, 2, 1, 2);
+    for (int index = 1; index <= 2; index++) {
+      request.add(
+          new IndexedConsensusRequest(
+              index, index, Collections.singletonList(new TestEntry(index, thisNode))));
+    }
+    final AtomicReference<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);

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/DataRegionStateMachine.java

@@ -251,7 +251,7 @@ protected TSStatus write(PlanNode planNode) {
     TSStatus result = null;
     int retryTime = 0;
     while (retryTime < MAX_WRITE_RETRY_TIMES) {
-      result = planNode.accept(new DataExecutionVisitor(), region);
+      result = executeWrite(planNode);
       // Let pipe retry with the original event instead of retrying a possibly mutated node here.
       if (needRetryForSpecificCases(result.getCode(), planNode)) {
         retryTime++;
@@ -284,6 +284,10 @@ protected TSStatus write(PlanNode planNode) {
     return result;
   }
 
+  protected TSStatus executeWrite(PlanNode planNode) {
+    return planNode.accept(new DataExecutionVisitor(), region);
+  }
+
   protected void waitBeforeNextWriteRetry() throws InterruptedException {
     Thread.sleep(WRITE_RETRY_WAIT_TIME_IN_MS);
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/consensus/statemachine/dataregion/IoTConsensusDataRegionStateMachine.java

@@ -50,6 +50,13 @@ public IoTConsensusDataRegionStateMachine(DataRegion region) {
     super(region);
   }
 
+  @Override
+  protected TSStatus executeWrite(PlanNode planNode) {
+    final DataRegion targetRegion = region;
+    return targetRegion.executeIoTConsensusWrite(
+        () -> planNode.accept(new DataExecutionVisitor(), targetRegion));
+  }
+
   @Override
   public TSStatus write(IConsensusRequest request) {
     try {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNodeShutdownHook.java

@@ -120,6 +120,9 @@ public void run() {
     CommonDescriptor.getInstance()
         .getConfig()
         .setNodeStatusWithReason(NodeStatus.ReadOnly, NodeStatus.STOPPING);
+    // A pre-apply read-only check can race with this transition. Drain admitted IoTConsensus
+    // applies, including their WAL separators, and permanently reject new ones before flushing.
+    StorageEngine.getInstance().stopIoTConsensusWrites();
     // Wait all wal are flushed
     WALManager.getInstance().waitAllWALFlushed();
 

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/storageengine/StorageEngine.java

@@ -433,6 +433,7 @@ private void asyncRecoverTsFileResource() {
 
   @Override
   public void stop() {
+    stopIoTConsensusWrites();
     loadTsFileManager.stop();
     for (DataRegion dataRegion : dataRegionMap.values()) {
       if (dataRegion != null) {
@@ -453,6 +454,7 @@ public void stop() {
 
   @Override
   public void shutdown(long milliseconds) throws ShutdownException {
+    stopIoTConsensusWrites();
     loadTsFileManager.stop();
     try {
       for (DataRegion dataRegion : dataRegionMap.values()) {
@@ -524,6 +526,17 @@ public synchronized void reset() {
     dataRegionMap.clear();
   }
 
+  public void stopIoTConsensusWrites() {
+    if (!CONFIG.getDataRegionConsensusProtocolClass().equals(ConsensusFactory.IOT_CONSENSUS)) {
+      return;
+    }
+    for (DataRegion dataRegion : dataRegionMap.values()) {
+      if (dataRegion != null) {
+        dataRegion.stopIoTConsensusWrites();
+      }
+    }
+  }
+
   /** flush command Sync asyncCloseOneProcessor all file node processors. */
   public void syncCloseAllProcessor() {
     LOGGER.info(StorageEngineMessages.START_CLOSING_ALL_DB_PROCESSOR);