Extend the node state machine and persist node statuses

integration-test/src/main/java/org/apache/iotdb/it/env/cluster/EnvUtils.java

@@ -19,6 +19,7 @@
 
 package org.apache.iotdb.it.env.cluster;
 
+import org.apache.iotdb.commons.cluster.NodeStatus;
 import org.apache.iotdb.it.framework.IoTDBTestLogger;
 
 import org.apache.tsfile.external.commons.lang3.SystemUtils;
@@ -69,6 +70,17 @@
 
 public class EnvUtils {
 
+  /**
+   * The status a node locally stopped via {@code AbstractNodeWrapper.stop()} is expected to be in.
+   * On Windows, {@code Process.destroy()} terminates the node process without running the JVM
+   * shutdown hooks, so the graceful-shutdown report is never sent and the ConfigNode marks the node
+   * Unknown by heartbeat timeout. On Unix, the shutdown hook reports the stop and the node becomes
+   * Stopped.
+   */
+  public static NodeStatus getNodeStatusAfterLocalStop() {
+    return SystemUtils.IS_OS_WINDOWS ? NodeStatus.Unknown : NodeStatus.Stopped;
+  }
+
   public static int[] searchAvailablePorts() {
     int length = 10;
     while (true) {

integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterNodeErrorStartUpIT.java

@@ -34,6 +34,7 @@
 import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp;
 import org.apache.iotdb.consensus.ConsensusFactory;
 import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.EnvUtils;
 import org.apache.iotdb.it.env.cluster.config.MppBaseConfig;
 import org.apache.iotdb.it.env.cluster.config.MppCommonConfig;
 import org.apache.iotdb.it.env.cluster.node.ConfigNodeWrapper;
@@ -212,15 +213,35 @@ public void testIllegalNodeRestart()
           dataNodeRestartResp.getStatus().getCode());
       Assert.assertTrue(dataNodeRestartResp.getStatus().getMessage().contains("whose nodeId="));
 
-      // Shutdown and check
+      // Local stop runs shutdown hooks on Unix, but terminates the process directly on Windows.
+      // A ConfigNode that was the leader at shutdown time can also remain Unknown if its report
+      // cannot reach another leader.
       EnvFactory.getEnv().shutdownConfigNode(1);
       EnvFactory.getEnv().shutdownDataNode(0);
       EnvFactory.getEnv()
           .ensureNodeStatus(
-              Arrays.asList(
-                  EnvFactory.getEnv().getConfigNodeWrapper(1),
-                  EnvFactory.getEnv().getDataNodeWrapper(0)),
-              Arrays.asList(NodeStatus.Unknown, NodeStatus.Unknown));
+              Arrays.asList(EnvFactory.getEnv().getDataNodeWrapper(0)),
+              Arrays.asList(EnvUtils.getNodeStatusAfterLocalStop()));
+      boolean isConfigNodeDown = false;
+      for (int retry = 0; retry < 30; retry++) {
+        TShowClusterResp showClusterResp = client.showCluster();
+        for (TConfigNodeLocation configNodeLocation : showClusterResp.getConfigNodeList()) {
+          if (configNodeLocation.getConsensusEndPoint().getPort()
+              == registeredConfigNodeWrapper.getConsensusPort()) {
+            String configNodeStatus =
+                showClusterResp.getNodeStatus().get(configNodeLocation.getConfigNodeId());
+            if (NodeStatus.Stopped.getStatus().equals(configNodeStatus)
+                || NodeStatus.Unknown.getStatus().equals(configNodeStatus)) {
+              isConfigNodeDown = true;
+            }
+          }
+        }
+        if (isConfigNodeDown) {
+          break;
+        }
+        Thread.sleep(1000);
+      }
+      Assert.assertTrue(isConfigNodeDown);
 
       /* Restart and updatePeer */
       // TODO: Delete this IT after enable modify internal TEndPoints

integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterNodeShutdownHookIT.java

@@ -19,12 +19,15 @@
 
 package org.apache.iotdb.confignode.it.cluster;
 
+import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
+import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
 import org.apache.iotdb.commons.client.exception.ClientManagerException;
 import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
 import org.apache.iotdb.commons.cluster.NodeStatus;
 import org.apache.iotdb.confignode.rpc.thrift.TShowClusterResp;
 import org.apache.iotdb.consensus.ConsensusFactory;
 import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.EnvUtils;
 import org.apache.iotdb.it.framework.IoTDBTestRunner;
 import org.apache.iotdb.itbase.category.ClusterIT;
 import org.apache.iotdb.rpc.TSStatusCode;
@@ -39,7 +42,6 @@
 
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
 
 @RunWith(IoTDBTestRunner.class)
 @Category({ClusterIT.class})
@@ -75,29 +77,57 @@ public void testNodeShutdownReporter()
     try (SyncConfigNodeIServiceClient client =
         (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
 
-      // The unknown Nodes should be detected immediately with the help of shutdown hook
+      // Unix shutdown hooks report Stopped; Windows local stop is detected as Unknown.
+      // A ConfigNode whose report cannot reach the newly elected leader remains Unknown too.
+      TShowClusterResp showClusterResp = client.showCluster();
+      Assert.assertEquals(
+          TSStatusCode.SUCCESS_STATUS.getStatusCode(), showClusterResp.getStatus().getCode());
+
+      int stoppedDataNodeId = -1;
+      for (TDataNodeLocation dataNodeLocation : showClusterResp.getDataNodeList()) {
+        if (dataNodeLocation.getInternalEndPoint().getPort()
+            == EnvFactory.getEnv().getDataNodeWrapper(0).getInternalPort()) {
+          stoppedDataNodeId = dataNodeLocation.getDataNodeId();
+        }
+      }
+      Assert.assertNotEquals(-1, stoppedDataNodeId);
+
+      int stoppedConfigNodeId = -1;
+      for (TConfigNodeLocation configNodeLocation : showClusterResp.getConfigNodeList()) {
+        if (configNodeLocation.getConsensusEndPoint().getPort()
+            == EnvFactory.getEnv().getConfigNodeWrapper(1).getConsensusPort()) {
+          stoppedConfigNodeId = configNodeLocation.getConfigNodeId();
+        }
+      }
+      Assert.assertNotEquals(-1, stoppedConfigNodeId);
+
       boolean isDetected = false;
-      for (int retry = 0; retry < 5; retry++) {
-        TShowClusterResp showClusterResp = client.showCluster();
+      for (int retry = 0; retry < 30; retry++) {
+        showClusterResp = client.showCluster();
         Assert.assertEquals(
             TSStatusCode.SUCCESS_STATUS.getStatusCode(), showClusterResp.getStatus().getCode());
-        AtomicInteger unknownNum = new AtomicInteger(0);
-        showClusterResp
-            .getNodeStatus()
-            .forEach(
-                (nodeId, nodeStatus) -> {
-                  if (NodeStatus.Unknown.getStatus().equals(nodeStatus)) {
-                    unknownNum.getAndIncrement();
-                  }
-                });
-        if (unknownNum.get() == 2) {
+
+        // Windows does not run the shutdown hook, so allow heartbeat failure detection to finish.
+        final String dataNodeStatus = showClusterResp.getNodeStatus().get(stoppedDataNodeId);
+        final boolean isDataNodeStopped =
+            EnvUtils.getNodeStatusAfterLocalStop().getStatus().equals(dataNodeStatus);
+
+        // The stopped ConfigNode is Stopped when its report reached a leader, and may otherwise
+        // remain Unknown until heartbeat timeout
+        final String configNodeStatus = showClusterResp.getNodeStatus().get(stoppedConfigNodeId);
+        final boolean isConfigNodeDetected =
+            NodeStatus.Stopped.getStatus().equals(configNodeStatus)
+                || NodeStatus.Unknown.getStatus().equals(configNodeStatus);
+
+        if (isDataNodeStopped && isConfigNodeDetected) {
           isDetected = true;
           break;
         }
 
         TimeUnit.SECONDS.sleep(1);
       }
-      Assert.assertTrue(isDetected);
+      Assert.assertTrue(
+          "Timed out waiting for the stopped DataNode and ConfigNode to be detected", isDetected);
     }
   }
 }