diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/NodeStatistics.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/NodeStatistics.java
index 04ed6c5028901..679c2fafa9fcb 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/NodeStatistics.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/NodeStatistics.java
@@ -56,7 +56,8 @@ public static NodeStatistics generateDefaultNodeStatistics() {
   }
 
   /**
-   * Resolves the requested statistics through {@link NodeStatus#transition}. If the requested
+   * Resolves the requested statistics through {@link NodeStatus#transition}, additionally
+   * preventing an unforced ReadOnly(Stopping) update from reviving a Stopped node. If the requested
    * status is accepted, its reason and load score are kept. Otherwise, the final status must not
    * carry the rejected status's reason or load score, e.g. Removing must not inherit ReadOnly's
    * DiskFull reason. The requested statistics timestamp is retained in both cases.
@@ -64,6 +65,14 @@ public static NodeStatistics generateDefaultNodeStatistics() {
   public static NodeStatistics transition(
       NodeStatistics previous, NodeStatistics requested, boolean force) {
     NodeStatus status = NodeStatus.transition(previous.status, requested.status, force);
+    if (!force
+        && previous.status == NodeStatus.Stopped
+        && requested.status == NodeStatus.ReadOnly
+        && NodeStatus.STOPPING.equals(requested.statusReason)) {
+      // A heartbeat during shutdown can be newer than the shutdown report without proving a
+      // restart.
+      status = NodeStatus.Stopped;
+    }
     if (status == requested.status) {
       return requested;
     }
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/LoadCachePersistedNodeStatusTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/LoadCachePersistedNodeStatusTest.java
index c8b9c84134fb5..cf9c8f101f204 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/LoadCachePersistedNodeStatusTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/LoadCachePersistedNodeStatusTest.java
@@ -387,6 +387,80 @@ public void testStoppingReasonIsClearedByShutdownAndCannotOverrideRemoving() thr
         nodeInfo.getPersistedNodeStatus(REMOVING_DATA_NODE_ID));
   }
 
+  @Test
+  public void testStoppingHeartbeatCannotOverwritePersistedShutdown() throws Exception {
+    loadCache.initHeartbeatCache(configManager);
+    Assert.assertEquals(
+        success().getCode(),
+        loadCache.trySetNodeStatus(DATA_NODE_ID, NodeStatus.Stopped, null, false).getCode());
+
+    for (int initialization = 0; initialization < 2; initialization++) {
+      // Also protect a Stopped record restored by a new leader, without a local shutdown sample.
+      if (initialization > 0) {
+        loadCache.initHeartbeatCache(configManager);
+      }
+      loadCache.cacheDataNodeHeartbeatSample(
+          DATA_NODE_ID, readOnlySample(System.nanoTime(), NodeStatus.STOPPING));
+      for (int refresh = 0; refresh < 2; refresh++) {
+        Assert.assertTrue(loadCache.updateNodeStatistics());
+        Assert.assertEquals(NodeStatus.Stopped, loadCache.getNodeStatus(DATA_NODE_ID));
+        Assert.assertNull(loadCache.getNodeStatusReason(DATA_NODE_ID));
+        Assert.assertEquals(
+            new Pair<>(NodeStatus.Stopped, null), nodeInfo.getPersistedNodeStatus(DATA_NODE_ID));
+      }
+    }
+    verify(consensusManager, never())
+        .write(new UpdateNodeStatusPlan(DATA_NODE_ID, NodeStatus.ReadOnly, NodeStatus.STOPPING));
+  }
+
+  @Test
+  public void testOtherReadOnlyReasonsCanReviveStoppedNode() throws Exception {
+    loadCache.initHeartbeatCache(configManager);
+    for (String reason :
+        Arrays.asList(
+            null,
+            NodeStatus.MANUAL,
+            NodeStatus.DISK_FULL,
+            NodeStatus.UNRECOVERABLE_ERROR + ": storage failure",
+            "CustomReason")) {
+      Assert.assertEquals(
+          success().getCode(),
+          loadCache.trySetNodeStatus(DATA_NODE_ID, NodeStatus.Stopped, null, false).getCode());
+      loadCache.cacheDataNodeHeartbeatSample(
+          DATA_NODE_ID, readOnlySample(System.nanoTime(), reason));
+      Assert.assertTrue(loadCache.updateNodeStatistics());
+      Assert.assertEquals(NodeStatus.ReadOnly, loadCache.getNodeStatus(DATA_NODE_ID));
+      Assert.assertEquals(reason, loadCache.getNodeStatusReason(DATA_NODE_ID));
+      Assert.assertEquals(
+          new Pair<>(NodeStatus.ReadOnly, reason), nodeInfo.getPersistedNodeStatus(DATA_NODE_ID));
+    }
+  }
+
+  @Test
+  public void testForcedStoppingUpdateCanReplaceStoppedNode() throws Exception {
+    loadCache.initHeartbeatCache(configManager);
+    Assert.assertEquals(
+        success().getCode(),
+        loadCache.trySetNodeStatus(DATA_NODE_ID, NodeStatus.Stopped, null, false).getCode());
+    Assert.assertEquals(
+        success().getCode(),
+        loadCache
+            .trySetNodeStatus(DATA_NODE_ID, NodeStatus.ReadOnly, NodeStatus.STOPPING, true)
+            .getCode());
+    Assert.assertEquals(NodeStatus.ReadOnly, loadCache.getNodeStatus(DATA_NODE_ID));
+    Assert.assertEquals(NodeStatus.STOPPING, loadCache.getNodeStatusReason(DATA_NODE_ID));
+    Assert.assertEquals(
+        new Pair<>(NodeStatus.ReadOnly, NodeStatus.STOPPING),
+        nodeInfo.getPersistedNodeStatus(DATA_NODE_ID));
+
+    Assert.assertTrue(loadCache.updateNodeStatistics());
+    Assert.assertEquals(NodeStatus.ReadOnly, loadCache.getNodeStatus(DATA_NODE_ID));
+    Assert.assertEquals(NodeStatus.STOPPING, loadCache.getNodeStatusReason(DATA_NODE_ID));
+    Assert.assertEquals(
+        new Pair<>(NodeStatus.ReadOnly, NodeStatus.STOPPING),
+        nodeInfo.getPersistedNodeStatus(DATA_NODE_ID));
+  }
+
   @Test
   public void testStoppedStatisticsUpdateAfterFailedPersistenceReturns() throws Exception {
     loadCache.initHeartbeatCache(configManager);
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/NodeCacheTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/NodeCacheTest.java
index 05262d3955d1f..a396babeed57b 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/NodeCacheTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/NodeCacheTest.java
@@ -131,6 +131,27 @@ public void stoppedStatusStickyAndRevivalTest() {
     }
   }
 
+  @Test
+  public void testNewerStoppingHeartbeatCannotReviveStoppedNode() {
+    DataNodeHeartbeatCache cache = new DataNodeHeartbeatCache(1);
+    cache.trySetNodeStatus(NodeStatus.Stopped, null, false);
+    NodeHeartbeatSample stoppingSample =
+        new NodeHeartbeatSample(
+            new TDataNodeHeartbeatResp()
+                .setHeartbeatTimestamp(cache.getLastSample().getSampleLogicalTimestamp() + 1)
+                .setStatus(NodeStatus.ReadOnly.getStatus())
+                .setStatusReason(NodeStatus.STOPPING));
+    cache.cacheHeartbeatSample(stoppingSample);
+    // Exercise the transition rule, not the filter for heartbeats older than the shutdown report.
+    Assert.assertSame(stoppingSample, cache.getLastSample());
+    for (int i = 0; i < 2; i++) {
+      cache.updateNodeStatistics();
+      Assert.assertEquals(NodeStatus.Stopped, cache.getNodeStatus());
+      Assert.assertNull(cache.getNodeStatusReason());
+      Assert.assertEquals(Long.MAX_VALUE, cache.getLoadScore());
+    }
+  }
+
   @Test
   public void testConnectionFailureIsNotUndoneByPreviousRunningHeartbeat() {
     for (BaseNodeCache cache :