Add Stopped node status

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

@@ -212,15 +212,35 @@ public void testIllegalNodeRestart()
           dataNodeRestartResp.getStatus().getCode());
       Assert.assertTrue(dataNodeRestartResp.getStatus().getMessage().contains("whose nodeId="));
 
-      // Shutdown and check
+      // Shutdown and check. A gracefully stopped node is reported as Stopped by its shutdown
+      // hook. A ConfigNode that was the leader at shutdown time can not report itself to another
+      // leader, so it may remain Unknown on the newly elected 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(NodeStatus.Stopped));
+      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,6 +19,8 @@
 
 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;
@@ -39,7 +41,6 @@
 
 import java.io.IOException;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.atomic.AtomicInteger;
 
 @RunWith(IoTDBTestRunner.class)
 @Category({ClusterIT.class})
@@ -75,29 +76,56 @@ public void testNodeShutdownReporter()
     try (SyncConfigNodeIServiceClient client =
         (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
 
-      // The unknown Nodes should be detected immediately with the help of shutdown hook
+      // The stopped Nodes should be detected immediately with the help of shutdown hook. A
+      // ConfigNode whose report can not reach the newly elected leader remains Unknown instead.
+      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();
+        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) {
+
+        // The stopped DataNode must be observable as Stopped
+        final String dataNodeStatus = showClusterResp.getNodeStatus().get(stoppedDataNodeId);
+        final boolean isDataNodeStopped = NodeStatus.Stopped.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);
     }
   }
 }

integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderDistributionIT.java

@@ -208,7 +208,7 @@ public void testCFDWithUnknownStatus() throws Exception {
       EnvFactory.getEnv()
           .ensureNodeStatus(
               Collections.singletonList(EnvFactory.getEnv().getDataNodeWrapper(0)),
-              Collections.singletonList(NodeStatus.Unknown));
+              Collections.singletonList(NodeStatus.Stopped));
 
       // Check leader distribution
       isDistributionBalanced = false;

integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBAutoRegionGroupExtension2IT.java

@@ -34,8 +34,10 @@
 import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
 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.itbase.env.BaseNodeWrapper;
 import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.apache.thrift.TException;
@@ -49,8 +51,10 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
@@ -104,7 +108,21 @@ public void testAutoRegionGroupExtensionPolicy2()
       EnvFactory.getEnv()
           .ensureNodeStatus(
               Collections.singletonList(EnvFactory.getEnv().getDataNodeWrapper(1)),
-              Collections.singletonList(NodeStatus.Unknown));
+              Collections.singletonList(EnvUtils.getNodeStatusAfterLocalStop()));
+
+      // The remaining DataNodes may transiently be ReadOnly (e.g. the disk-full flap on a busy
+      // runner, which auto-recovers at the next disk sampling); wait for them to be Running so a
+      // transient status does not fail the allocation below.
+      List<BaseNodeWrapper> remainingDataNodes = new ArrayList<>();
+      for (int i = 0; i < testDataNodeNum; i++) {
+        if (i != 1) {
+          remainingDataNodes.add(EnvFactory.getEnv().getDataNodeWrapper(i));
+        }
+      }
+      EnvFactory.getEnv()
+          .ensureNodeStatus(
+              remainingDataNodes,
+              Collections.nCopies(remainingDataNodes.size(), NodeStatus.Running));
 
       // Create 3 DataPartitions to extend 3 DataRegionGroups
       for (int i = 0; i < testMinDataRegionGroupNum; i++) {

integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionCreationIT.java

@@ -28,6 +28,7 @@
 import org.apache.iotdb.commons.cluster.RegionStatus;
 import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
 import org.apache.iotdb.confignode.it.utils.ConfigNodeTestUtils;
+import org.apache.iotdb.confignode.rpc.thrift.TDataNodeInfo;
 import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionReq;
 import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionTableResp;
 import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
@@ -318,36 +319,34 @@ public void testPartitionAllocation() throws Exception {
 
       // Shutdown 1 DataNode
       // Current cluster: 1C5D
-      // DataNode status: Running, Running, Removing, ReadOnly, Unknown
+      // DataNode status: Running, Running, Removing, ReadOnly, Stopped
       // Region distribution: [0, 1, 2], [0, 1, 2], [0], [1], [2]
       EnvFactory.getEnv().shutdownDataNode(4);
       // Wait for shutdown check
-      while (true) {
-        AtomicBoolean containUnknown = new AtomicBoolean(false);
+      boolean isShutdownDetected = false;
+      for (int retry = 0; retry < 60; retry++) {
         TShowDataNodesResp showDataNodesResp = client.showDataNodes();
-        showDataNodesResp
-            .getDataNodesInfoList()
-            .forEach(
-                dataNodeInfo -> {
-                  if (NodeStatus.Unknown.getStatus().equals(dataNodeInfo.getStatus())) {
-                    containUnknown.set(true);
-                  }
-                });
-
-        if (containUnknown.get()) {
+        for (TDataNodeInfo dataNodeInfo : showDataNodesResp.getDataNodesInfoList()) {
+          if (NodeStatus.Stopped.getStatus().equals(dataNodeInfo.getStatus())) {
+            isShutdownDetected = true;
+            break;
+          }
+        }
+        if (isShutdownDetected) {
           break;
         }
         TimeUnit.SECONDS.sleep(1);
       }
+      Assert.assertTrue(isShutdownDetected);
 
       // Register 1 DataNode and Create 1 DataPartition to extend 1 DataRegionGroup
       // The new DataRegions wouldn't be allocated to the Removing and ReadOnly DataNode
-      // But the new DataRegion can be allocated to the Unknown DataNode
+      // But the new DataRegion can be allocated to the Stopped DataNode
       // Current cluster: 1C6D
-      // Status: Running, Running, Removing, ReadOnly, Unknown, Running
+      // Status: Running, Running, Removing, ReadOnly, Stopped, Running
       // RegionGroup: [0, 1, 2, 3], [0, 1, 2], [0], [1], [2, 3], [3]
       EnvFactory.getEnv().registerNewDataNode(false);
-      // Use thread sleep to replace verifying because the Unknown DataNode can not pass the
+      // Use thread sleep to replace verifying because the Stopped DataNode can not pass the
       // connection check
       TimeUnit.SECONDS.sleep(25);
       partitionSlotsMap =
@@ -446,23 +445,25 @@ public void testPartitionAllocation() throws Exception {
       // RegionGroup: [0, 1, 2, 3], [0, 1, 2], [0], [1], [2, 3], [3]
       EnvFactory.getEnv().startDataNode(4);
       // Wait for restart check
-      while (true) {
-        AtomicBoolean containUnknown = new AtomicBoolean(false);
+      boolean isRestartDetected = false;
+      for (int retry = 0; retry < 60; retry++) {
         TShowDataNodesResp showDataNodesResp = client.showDataNodes();
-        showDataNodesResp
-            .getDataNodesInfoList()
-            .forEach(
-                dataNodeInfo -> {
-                  if (NodeStatus.Unknown.getStatus().equals(dataNodeInfo.getStatus())) {
-                    containUnknown.set(true);
-                  }
-                });
-
-        if (!containUnknown.get()) {
+        boolean containDown = false;
+        for (TDataNodeInfo dataNodeInfo : showDataNodesResp.getDataNodesInfoList()) {
+          // The restarted DataNode keeps Stopped until its first heartbeat revives it
+          if (NodeStatus.Unknown.getStatus().equals(dataNodeInfo.getStatus())
+              || NodeStatus.Stopped.getStatus().equals(dataNodeInfo.getStatus())) {
+            containDown = true;
+            break;
+          }
+        }
+        if (!containDown) {
+          isRestartDetected = true;
           break;
         }
         TimeUnit.SECONDS.sleep(1);
       }
+      Assert.assertTrue(isRestartDetected);
       // Check Region count and status
       for (int i = 0; i < 30; i++) {
         runningCnt = 0;