[To dev/1.3] Fence IoTConsensus writes before storage shutdown (#18841, #18893)

integration-test/src/test/java/org/apache/iotdb/pipe/it/autocreate/IoTDBPipeClusterIT.java

@@ -442,31 +442,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);
                 }
@@ -476,20 +477,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

@@ -31,6 +31,7 @@
 import org.apache.iotdb.commons.utils.CommonDateTimeUtils;
 import org.apache.iotdb.commons.utils.KillPoint.DataNodeKillPoints;
 import org.apache.iotdb.commons.utils.KillPoint.KillPoint;
+import org.apache.iotdb.commons.utils.StatusUtils;
 import org.apache.iotdb.consensus.IStateMachine;
 import org.apache.iotdb.consensus.common.DataSet;
 import org.apache.iotdb.consensus.common.Peer;
@@ -1010,7 +1011,13 @@ private TSStatus cacheAndInsertLatestNode(DeserializedBatchIndexedConsensusReque
         List<TSStatus> subStatus = new LinkedList<>();
         for (IConsensusRequest insertNode : request.getInsertNodes()) {
           insertNode.markAsGeneratedByRemoteConsensusLeader();
-          subStatus.add(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.
+          subStatus.add(
+              isReadOnly()
+                  ? StatusUtils.getStatus(TSStatusCode.SYSTEM_READ_ONLY)
+                  : stateMachine.write(insertNode));
         }
         long applyTime = System.nanoTime();
         ioTConsensusServerMetrics.recordApplyCost(applyTime - sortTime);

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

@@ -0,0 +1,135 @@
+/*
+ * 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.consensus.iot;
+
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.commons.utils.StatusUtils;
+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.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;
+import org.junit.rules.TemporaryFolder;
+
+import java.util.Collections;
+import java.util.TreeSet;
+import java.util.concurrent.Executors;
+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;
+import static org.junit.Assert.assertTrue;
+
+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(),
+            thisNode,
+            new TreeSet<>(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);
+    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(
+          StatusUtils.getStatus(TSStatusCode.SYSTEM_READ_ONLY).getMessage(),
+          status.get().getSubStatus().get(0).getMessage());
+      assertEquals(
+          TSStatusCode.SYSTEM_READ_ONLY.getStatusCode(),
+          status.get().getSubStatus().get(1).getCode());
+      assertTrue(stateMachine.getRequestSet().isEmpty());
+
+      // 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();
+    }
+  }
+}

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

@@ -60,7 +60,6 @@ public TSStatus visitPlan(PlanNode node, DataRegion context) {
   public TSStatus visitInsertRow(InsertRowNode node, DataRegion dataRegion) {
     try {
       dataRegion.insert(node);
-      dataRegion.insertSeparatorToWAL();
       return StatusUtils.OK;
     } catch (OutOfTTLException e) {
       LOGGER.warn("Error in executing plan node: {}, caused by {}", node, e.getMessage());
@@ -78,7 +77,6 @@ public TSStatus visitInsertRow(InsertRowNode node, DataRegion dataRegion) {
   public TSStatus visitInsertTablet(InsertTabletNode node, DataRegion dataRegion) {
     try {
       dataRegion.insertTablet(node);
-      dataRegion.insertSeparatorToWAL();
       return StatusUtils.OK;
     } catch (OutOfTTLException e) {
       LOGGER.warn("Error in executing plan node: {}, caused by {}", node, e.getMessage());
@@ -115,7 +113,6 @@ public TSStatus visitInsertTablet(InsertTabletNode node, DataRegion dataRegion)
   public TSStatus visitInsertRows(InsertRowsNode node, DataRegion dataRegion) {
     try {
       dataRegion.insert(node);
-      dataRegion.insertSeparatorToWAL();
       return StatusUtils.OK;
     } catch (WriteProcessRejectException e) {
       LOGGER.warn("Reject in executing plan node: {}, caused by {}", node, e.getMessage());
@@ -149,8 +146,9 @@ public TSStatus visitInsertRows(InsertRowsNode node, DataRegion dataRegion) {
   public TSStatus visitInsertMultiTablets(InsertMultiTabletsNode node, DataRegion dataRegion) {
     try {
       dataRegion.insertTablets(node);
-      dataRegion.insertSeparatorToWAL();
       return StatusUtils.OK;
+    } catch (WriteProcessRejectException e) {
+      return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
     } catch (BatchProcessException e) {
       LOGGER.warn("Batch failure in executing a InsertMultiTabletsNode.");
       TSStatus firstStatus = null;
@@ -181,7 +179,6 @@ public TSStatus visitInsertRowsOfOneDevice(
       InsertRowsOfOneDeviceNode node, DataRegion dataRegion) {
     try {
       dataRegion.insert(node);
-      dataRegion.insertSeparatorToWAL();
       return StatusUtils.OK;
     } catch (WriteProcessRejectException e) {
       LOGGER.warn("Reject in executing plan node: {}, caused by {}", node, e.getMessage());
@@ -244,6 +241,8 @@ public TSStatus visitDeleteData(DeleteDataNode node, DataRegion dataRegion) {
       dataRegion.insertSeparatorToWAL();
       PipeInsertionDataNodeListener.getInstance().listenToDeleteData(node);
       return StatusUtils.OK;
+    } catch (WriteProcessRejectException e) {
+      return RpcUtils.getStatus(e.getErrorCode(), e.getMessage());
     } catch (IOException | IllegalPathException e) {
       LOGGER.error("Error in executing plan node: {}", node, e);
       return new TSStatus(TSStatusCode.WRITE_PROCESS_ERROR.getStatusCode());

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

@@ -95,6 +95,9 @@ public void run() {
     // Reject write operations to make sure all tsfiles will be sealed
     CommonDescriptor.getInstance().getConfig().setStopping(true);
     CommonDescriptor.getInstance().getConfig().setNodeStatus(NodeStatus.ReadOnly);
+    // 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

@@ -398,6 +398,7 @@ private void asyncRecoverTsFileResource() {
 
   @Override
   public void stop() {
+    stopIoTConsensusWrites();
     loadTsFileManager.stop();
     for (DataRegion dataRegion : dataRegionMap.values()) {
       if (dataRegion != null) {
@@ -417,6 +418,7 @@ public void stop() {
 
   @Override
   public void shutdown(long milliseconds) throws ShutdownException {
+    stopIoTConsensusWrites();
     loadTsFileManager.stop();
     try {
       for (DataRegion dataRegion : dataRegionMap.values()) {
@@ -484,6 +486,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("Start closing all database processor");