diff --git a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/EnvUtils.java b/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/EnvUtils.java index 63c2041199d82..b69e86c12bd42 100644 --- a/integration-test/src/main/java/org/apache/iotdb/it/env/cluster/EnvUtils.java +++ b/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) { diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterNodeErrorStartUpIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterNodeErrorStartUpIT.java index e12ab129d21c7..b6e9c8c57992b 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterNodeErrorStartUpIT.java +++ b/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 diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterNodeShutdownHookIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterNodeShutdownHookIT.java index 8917f163557be..8d543d373b72a 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/cluster/IoTDBClusterNodeShutdownHookIT.java +++ b/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); } } } diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderDistributionIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderDistributionIT.java index fc0b73d6e1c37..5cabdf7f71561 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/load/IoTDBRegionGroupLeaderDistributionIT.java +++ b/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; diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBAutoRegionGroupExtension2IT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBAutoRegionGroupExtension2IT.java index 2ca2670cbad1f..f25e5d30809cc 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBAutoRegionGroupExtension2IT.java +++ b/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 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++) { diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionCreationIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionCreationIT.java index 954f54e01bd02..20cdc991d02e9 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionCreationIT.java +++ b/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; diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionDurableIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionDurableIT.java index 8d6e4a5f83113..f44730a7fd3dd 100644 --- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionDurableIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/partition/IoTDBPartitionDurableIT.java @@ -285,49 +285,47 @@ public void testReadOnlyDataNode() throws Exception { } @Test - public void testUnknownDataNode() throws Exception { + public void testStoppedDataNode() throws Exception { // Shutdown a DataNode, the ConfigNode should still be able to create RegionGroup EnvFactory.getEnv().shutdownDataNode(testDataNodeId); EnvFactory.getEnv() .ensureNodeStatus( Collections.singletonList(EnvFactory.getEnv().getDataNodeWrapper(testDataNodeId)), - Collections.singletonList(NodeStatus.Unknown)); + Collections.singletonList(NodeStatus.Stopped)); try (SyncConfigNodeIServiceClient client = (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) { // Wait for shutdown check TShowClusterResp showClusterResp; - 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); int runningCnt = 0; - int unknownCnt = 0; + int stoppedCnt = 0; showClusterResp = client.showCluster(); for (TDataNodeLocation dataNodeLocation : showClusterResp.getDataNodeList()) { if (NodeStatus.Running.getStatus() .equals(showClusterResp.getNodeStatus().get(dataNodeLocation.getDataNodeId()))) { runningCnt += 1; - } else if (NodeStatus.Unknown.getStatus() + } else if (NodeStatus.Stopped.getStatus() .equals(showClusterResp.getNodeStatus().get(dataNodeLocation.getDataNodeId()))) { - unknownCnt += 1; + stoppedCnt += 1; } } Assert.assertEquals(2, runningCnt); - Assert.assertEquals(1, unknownCnt); + Assert.assertEquals(1, stoppedCnt); // Test getOrCreateDataPartition, ConfigNode should create DataPartition and return Map> partitionSlotsMap = ConfigNodeTestUtils.constructPartitionSlotsMap( @@ -369,7 +367,7 @@ public void testUnknownDataNode() throws Exception { // Check Region count runningCnt = 0; - unknownCnt = 0; + int regionUnknownCnt = 0; TShowRegionResp showRegionResp = client.showRegion(new TShowRegionReq()); showRegionResp .getRegionInfoList() @@ -383,12 +381,12 @@ public void testUnknownDataNode() throws Exception { if (RegionStatus.Running.getStatus().equals(regionInfo.getStatus())) { runningCnt += 1; } else if (RegionStatus.Unknown.getStatus().equals(regionInfo.getStatus())) { - unknownCnt += 1; + regionUnknownCnt += 1; } } // The runningCnt should be exactly twice as the unknownCnt // since there exists one DataNode is shutdown - Assert.assertEquals(unknownCnt * 2, runningCnt); + Assert.assertEquals(regionUnknownCnt * 2, runningCnt); // Test getOrCreateDataPartition, ConfigNode should create DataPartition and return partitionSlotsMap = @@ -430,7 +428,7 @@ public void testUnknownDataNode() throws Exception { // Check Region count and status runningCnt = 0; - unknownCnt = 0; + regionUnknownCnt = 0; showRegionResp = client.showRegion(new TShowRegionReq()); showRegionResp .getRegionInfoList() @@ -444,12 +442,12 @@ public void testUnknownDataNode() throws Exception { if (RegionStatus.Running.getStatus().equals(regionInfo.getStatus())) { runningCnt += 1; } else if (RegionStatus.Unknown.getStatus().equals(regionInfo.getStatus())) { - unknownCnt += 1; + regionUnknownCnt += 1; } } // The runningCnt should be exactly twice as the unknownCnt // since there exists one DataNode is shutdown - Assert.assertEquals(unknownCnt * 2, runningCnt); + Assert.assertEquals(regionUnknownCnt * 2, runningCnt); EnvFactory.getEnv().startDataNode(testDataNodeId); EnvFactory.getEnv() diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationQueryWithNetworkPartitionIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationQueryWithNetworkPartitionIT.java index 3e8bc96f1db0d..684976206e765 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationQueryWithNetworkPartitionIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/recent/IoTDBTableAggregationQueryWithNetworkPartitionIT.java @@ -167,7 +167,7 @@ public void test1() throws IoTDBConnectionException, StatementExecutionException EnvFactory.getEnv() .ensureNodeStatus( Collections.singletonList(dataNodeWrapper), - Collections.singletonList(NodeStatus.Unknown)); + Collections.singletonList(NodeStatus.Stopped)); } List otherNodes = new ArrayList<>(); @@ -209,7 +209,7 @@ public void test2() EnvFactory.getEnv() .ensureNodeStatus( Collections.singletonList(dataNodeWrapper), - Collections.singletonList(NodeStatus.Unknown)); + Collections.singletonList(NodeStatus.Stopped)); } List otherNodes = new ArrayList<>(); diff --git a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/view/recent/IoTDBTableViewQueryWithCachedDeviceIT.java b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/view/recent/IoTDBTableViewQueryWithCachedDeviceIT.java index 74e009afb179e..c2e2c696cc738 100644 --- a/integration-test/src/test/java/org/apache/iotdb/relational/it/query/view/recent/IoTDBTableViewQueryWithCachedDeviceIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/relational/it/query/view/recent/IoTDBTableViewQueryWithCachedDeviceIT.java @@ -97,7 +97,7 @@ public void test1() throws IoTDBConnectionException, StatementExecutionException EnvFactory.getEnv() .ensureNodeStatus( Collections.singletonList(dataNodeWrapper), - Collections.singletonList(NodeStatus.Unknown)); + Collections.singletonList(NodeStatus.Stopped)); } EnvFactory.getEnv().startAllDataNodes(); for (DataNodeWrapper dataNodeWrapper : EnvFactory.getEnv().getDataNodeWrapperList()) { diff --git a/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBConnectionsIT.java b/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBConnectionsIT.java index 18b5f897e5c2b..90795d969ffcd 100644 --- a/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBConnectionsIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/session/it/IoTDBConnectionsIT.java @@ -23,6 +23,7 @@ import org.apache.iotdb.commons.cluster.NodeStatus; import org.apache.iotdb.commons.conf.CommonDescriptor; import org.apache.iotdb.commons.schema.column.ColumnHeaderConstant; +import org.apache.iotdb.confignode.rpc.thrift.TDataNodeInfo; import org.apache.iotdb.confignode.rpc.thrift.TShowDataNodesResp; import org.apache.iotdb.it.env.EnvFactory; import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper; @@ -49,7 +50,6 @@ import java.util.HashSet; import java.util.Set; import java.util.concurrent.TimeUnit; -import java.util.concurrent.atomic.AtomicBoolean; import static org.apache.iotdb.db.it.utils.TestUtils.createUser; import static org.apache.iotdb.itbase.env.BaseEnv.TABLE_SQL_DIALECT; @@ -278,23 +278,24 @@ public void testClosedDataNodeGetConnections() throws Exception { (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) { // 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()) { + // A gracefully stopped DataNode is reported as Stopped by its shutdown hook; if + // the report fails it becomes Unknown instead + if (NodeStatus.Unknown.getStatus().equals(dataNodeInfo.getStatus()) + || NodeStatus.Stopped.getStatus().equals(dataNodeInfo.getStatus())) { + isShutdownDetected = true; + break; + } + } + if (isShutdownDetected) { break; } TimeUnit.SECONDS.sleep(1); } + Assert.assertTrue(isShutdownDetected); } int activeDataNodeId = (int) allDataNodeId.toArray()[1]; @@ -327,23 +328,25 @@ public void testClosedDataNodeGetConnections() throws Exception { try (SyncConfigNodeIServiceClient client = (SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) { // 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); } // The ConfigNode may report the restarted DataNode as Running before its client RPC service is diff --git a/integration-test/src/test/java/org/apache/iotdb/session/it/pool/SessionPoolIT.java b/integration-test/src/test/java/org/apache/iotdb/session/it/pool/SessionPoolIT.java index 0a801477866f3..0ce3be416d400 100644 --- a/integration-test/src/test/java/org/apache/iotdb/session/it/pool/SessionPoolIT.java +++ b/integration-test/src/test/java/org/apache/iotdb/session/it/pool/SessionPoolIT.java @@ -23,6 +23,7 @@ import org.apache.iotdb.isession.pool.ISessionPool; import org.apache.iotdb.isession.pool.SessionDataSetWrapper; 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.category.LocalStandaloneIT; @@ -252,7 +253,10 @@ public void tryIfTheServerIsRestart() throws InterruptedException { node.stop(); EnvFactory.getEnv() .ensureNodeStatus( - Collections.singletonList(node), Collections.singletonList(NodeStatus.Unknown)); + // A gracefully stopped DataNode is reported as Stopped by its shutdown hook (on + // Windows the process is hard-killed, so it is marked Unknown instead) + Collections.singletonList(node), + Collections.singletonList(EnvUtils.getNodeStatusAfterLocalStop())); // user does not know what happens. while (wrapper.hasNext()) { wrapper.next(); @@ -263,7 +267,10 @@ public void tryIfTheServerIsRestart() throws InterruptedException { node.stop(); EnvFactory.getEnv() .ensureNodeStatus( - Collections.singletonList(node), Collections.singletonList(NodeStatus.Unknown)); + // A gracefully stopped DataNode is reported as Stopped by its shutdown hook (on + // Windows the process is hard-killed, so it is marked Unknown instead) + Collections.singletonList(node), + Collections.singletonList(EnvUtils.getNodeStatusAfterLocalStop())); node.start(); EnvFactory.getEnv() .ensureNodeStatus( @@ -289,7 +296,10 @@ public void tryIfTheServerIsRestart() throws InterruptedException { node.stop(); EnvFactory.getEnv() .ensureNodeStatus( - Collections.singletonList(node), Collections.singletonList(NodeStatus.Unknown)); + // A gracefully stopped DataNode is reported as Stopped by its shutdown hook (on + // Windows the process is hard-killed, so it is marked Unknown instead) + Collections.singletonList(node), + Collections.singletonList(EnvUtils.getNodeStatusAfterLocalStop())); node.start(); EnvFactory.getEnv() .ensureNodeStatus( @@ -345,7 +355,10 @@ public void restart() throws InterruptedException { node.stop(); EnvFactory.getEnv() .ensureNodeStatus( - Collections.singletonList(node), Collections.singletonList(NodeStatus.Unknown)); + // A gracefully stopped DataNode is reported as Stopped by its shutdown hook (on + // Windows the process is hard-killed, so it is marked Unknown instead) + Collections.singletonList(node), + Collections.singletonList(EnvUtils.getNodeStatusAfterLocalStop())); pool = EnvFactory.getEnv().getSessionPool(1); // all this ten data will fail. write10Data(pool, false); diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java index 10ea4763af9a2..8acb86569c91e 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java @@ -469,8 +469,10 @@ public final class ManagerMessages { "Successfully transferred config region snapshot {}."; public static final String THERE_IS_NO_RUNNING_DATANODE_TO_EXECUTE_CQ = "There is no RUNNING DataNode to execute CQ {}"; - public static final String THE_CONFIGNODE_WILL_BE_SHUTDOWN_SOON_MARK_IT_AS_UNKNOWN = - "The ConfigNode-{} will be shutdown soon, mark it as Unknown"; + public static final String LOG_THE_CONFIGNODE_IS_REMOVING_SKIP_MARKING_IT_AS_STOPPED_41B041A3 = + "The ConfigNode-{} is Removing, skip marking it as Stopped"; + public static final String LOG_THE_CONFIGNODE_WILL_BE_SHUTDOWN_SOON_MARK_IT_AS_STOPPED_D2A64AFD = + "The ConfigNode-{} will be shutdown soon, mark it as Stopped"; public static final String THE_CONFIG_REGION_AIR_GAP_CONNECTOR_DOES_NOT_SUPPORT_TRANSFERRING = "The config region air gap connector does not support transferring single file piece bytes."; public static final String THE_CONFIG_REGION_SINK_DOES_NOT_SUPPORT_TRANSFERRING_SINGLE_FILE = @@ -479,8 +481,10 @@ public final class ManagerMessages { "The config region snapshots %s cannot be parsed."; public static final String THE_DATABASE_DOESN_T_EXIST_MAYBE_IT_HAS_BEEN_PRE = "The Database: {} doesn't exist. Maybe it has been pre-deleted."; - public static final String THE_DATANODE_WILL_BE_SHUTDOWN_SOON_MARK_IT_AS_UNKNOWN = - "The DataNode-{} will be shutdown soon, mark it as Unknown"; + public static final String LOG_THE_DATANODE_IS_REMOVING_SKIP_MARKING_IT_AS_STOPPED_90F95D71 = + "The DataNode-{} is Removing, skip marking it as Stopped"; + public static final String LOG_THE_DATANODE_WILL_BE_SHUTDOWN_SOON_MARK_IT_AS_STOPPED_05CF8A45 = + "The DataNode-{} will be shutdown soon, mark it as Stopped"; public static final String THE_REMOVENODEREPLICASELECT_METHOD_OF_GREEDYREGIONGROUPALLOCATOR_IS_YET = "The removeNodeReplicaSelect method of GreedyRegionGroupAllocator is yet to be implemented."; public static final String THE_REMOVENODEREPLICASELECT_METHOD_OF_PARTITEGRAPHPLACEMENTREGIONGROUPALLOCATOR = diff --git a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index e04460375f9b4..1757c49ac5886 100644 --- a/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java @@ -1218,6 +1218,9 @@ private ProcedureMessages() {} + " cache status is {}"; public static final String MESSAGE_REMOVE_CONFIGNODE_FAILED_BECAUSE_UPDATE_CONSENSUSGROUP_PEER_INFORMATION_FAILED_FCE5302B = "Remove ConfigNode failed because update ConsensusGroup peer information failed."; public static final String MESSAGE_CAN_T_REMOVE_DATANODE_LIMIT_REPLICATION_FACTOR_D960E3A6 = "Can't remove datanode due to the limit of replication factor, "; + public static final String + MESSAGE_SUBMIT_REMOVEDATANODESPROCEDURE_FAILED_BECAUSE_WHEN_THERE_ARE_OTHER_UNKNOWN_STOPPED_OR_READONLY_NODES_IN_THE_CONSENSUS_GROUP_THAT_ARE_NOT_REMOVE_NODES_THE_REMOVE_OPERATION_CANNOT_BE_PERFORMED_FOR_SECURITY_REASONS_PLEASE_CHECK_THE_STATUS_OF_THE_NODE_ARG_AND_ENSURE_IT_IS_RUNNING_5063B3F6 = + "Submit RemoveDataNodesProcedure failed, because when there are other unknown, stopped or readonly nodes in the consensus group that are not remove nodes, the remove operation cannot be performed for security reasons. Please check the status of the node %s and ensure it is running."; public static final String MESSAGE_AVAILABLEDATANODESIZE_ARG_MAXREPLICAFACTOR_ARG_MAX_ALLOWED_REMOVED_DATA_NODE_SIZE_FB8C382C = "availableDataNodeSize: %s, maxReplicaFactor: %s, max allowed removed Data Node size is: %s"; public static final String EXCEPTION_NOT_SUPPORTED_0A83F963 = " is not supported"; public static final String LOG_START_ADD_TRIGGER_ARG_TRIGGERTABLE_CONFIG_NODES_NEEDTOSAVEJAR_ARG_0C23D81E = "Start to add trigger [{}] in TriggerTable on Config Nodes, needToSaveJar[{}]"; diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java index 499e922f7a7d2..a8b58781261f8 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java @@ -464,8 +464,10 @@ public final class ManagerMessages { "成功传输 config region 快照 {}。"; public static final String THERE_IS_NO_RUNNING_DATANODE_TO_EXECUTE_CQ = "没有处于 RUNNING 状态的 DataNode 可用于执行 CQ {}"; - public static final String THE_CONFIGNODE_WILL_BE_SHUTDOWN_SOON_MARK_IT_AS_UNKNOWN = - "ConfigNode-{} 即将关闭，将其标记为 Unknown"; + public static final String LOG_THE_CONFIGNODE_IS_REMOVING_SKIP_MARKING_IT_AS_STOPPED_41B041A3 = + "ConfigNode-{} 当前处于 Removing 状态，跳过将其标记为 Stopped"; + public static final String LOG_THE_CONFIGNODE_WILL_BE_SHUTDOWN_SOON_MARK_IT_AS_STOPPED_D2A64AFD = + "ConfigNode-{} 即将关闭，将其标记为 Stopped"; public static final String THE_CONFIG_REGION_AIR_GAP_CONNECTOR_DOES_NOT_SUPPORT_TRANSFERRING = "config region air gap connector 不支持传输单文件分片字节。"; public static final String THE_CONFIG_REGION_SINK_DOES_NOT_SUPPORT_TRANSFERRING_SINGLE_FILE = @@ -474,8 +476,10 @@ public final class ManagerMessages { "无法解析 config region 快照 %s。"; public static final String THE_DATABASE_DOESN_T_EXIST_MAYBE_IT_HAS_BEEN_PRE = "Database: {} 不存在，可能已被预删除。"; - public static final String THE_DATANODE_WILL_BE_SHUTDOWN_SOON_MARK_IT_AS_UNKNOWN = - "DataNode-{} 即将关闭，将其标记为 Unknown"; + public static final String LOG_THE_DATANODE_IS_REMOVING_SKIP_MARKING_IT_AS_STOPPED_90F95D71 = + "DataNode-{} 当前处于 Removing 状态，跳过将其标记为 Stopped"; + public static final String LOG_THE_DATANODE_WILL_BE_SHUTDOWN_SOON_MARK_IT_AS_STOPPED_05CF8A45 = + "DataNode-{} 即将关闭，将其标记为 Stopped"; public static final String THE_REMOVENODEREPLICASELECT_METHOD_OF_GREEDYREGIONGROUPALLOCATOR_IS_YET = "GreedyRegionGroupAllocator 的 removeNodeReplicaSelect 方法尚未实现。"; public static final String THE_REMOVENODEREPLICASELECT_METHOD_OF_PARTITEGRAPHPLACEMENTREGIONGROUPALLOCATOR = diff --git a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java index 3014e1c0076d8..a8f07ed4ffa35 100644 --- a/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java +++ b/iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java @@ -1167,6 +1167,9 @@ private ProcedureMessages() {} "移除 ConfigNode 失败，原因：更新 ConsensusGroup peer 信息失败。"; public static final String MESSAGE_CAN_T_REMOVE_DATANODE_LIMIT_REPLICATION_FACTOR_D960E3A6 = "无法移除 DataNode，原因：受副本因子限制，"; + public static final String + MESSAGE_SUBMIT_REMOVEDATANODESPROCEDURE_FAILED_BECAUSE_WHEN_THERE_ARE_OTHER_UNKNOWN_STOPPED_OR_READONLY_NODES_IN_THE_CONSENSUS_GROUP_THAT_ARE_NOT_REMOVE_NODES_THE_REMOVE_OPERATION_CANNOT_BE_PERFORMED_FOR_SECURITY_REASONS_PLEASE_CHECK_THE_STATUS_OF_THE_NODE_ARG_AND_ENSURE_IT_IS_RUNNING_5063B3F6 = + "提交 RemoveDataNodesProcedure 失败：共识组内存在非移除目标且状态为 Unknown、Stopped 或 ReadOnly 的其他节点时，出于安全考虑无法执行移除操作。请检查节点 %s 的状态并确保其处于 Running 状态。"; public static final String MESSAGE_AVAILABLEDATANODESIZE_ARG_MAXREPLICAFACTOR_ARG_MAX_ALLOWED_REMOVED_DATA_NODE_SIZE_FB8C382C = "availableDataNodeSize：%s，maxReplicaFactor：%s，允许移除的最大 DataNode 数量为：%s"; diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java index f86681f2bbec2..ec46015a5fcc5 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java @@ -579,15 +579,21 @@ public TSStatus removeAINode() { public TSStatus reportDataNodeShutdown(TDataNodeLocation dataNodeLocation) { TSStatus status = confirmLeader(); if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - // Force updating the target DataNode's status to Unknown - getLoadManager() - .forceUpdateNodeCache( - NodeType.DataNode, - dataNodeLocation.getDataNodeId(), - new NodeHeartbeatSample(NodeStatus.Unknown)); - LOGGER.info( - ManagerMessages.THE_DATANODE_WILL_BE_SHUTDOWN_SOON_MARK_IT_AS_UNKNOWN, - dataNodeLocation.getDataNodeId()); + int dataNodeId = dataNodeLocation.getDataNodeId(); + if (NodeStatus.Removing.equals(getLoadManager().getNodeStatus(dataNodeId))) { + // Removing has the highest priority and can not be refreshed by the Stopped report + LOGGER.info( + ManagerMessages.LOG_THE_DATANODE_IS_REMOVING_SKIP_MARKING_IT_AS_STOPPED_90F95D71, + dataNodeId); + } else { + // Force updating the target DataNode's status to Stopped + getLoadManager() + .forceUpdateNodeCache( + NodeType.DataNode, dataNodeId, new NodeHeartbeatSample(NodeStatus.Stopped)); + LOGGER.info( + ManagerMessages.LOG_THE_DATANODE_WILL_BE_SHUTDOWN_SOON_MARK_IT_AS_STOPPED_05CF8A45, + dataNodeId); + } } return status; } @@ -1628,15 +1634,21 @@ public TSStatus removeConfigNode(RemoveConfigNodePlan removeConfigNodePlan) { public TSStatus reportConfigNodeShutdown(TConfigNodeLocation configNodeLocation) { TSStatus status = confirmLeader(); if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - // Force updating the target ConfigNode's status to Unknown - getLoadManager() - .forceUpdateNodeCache( - NodeType.ConfigNode, - configNodeLocation.getConfigNodeId(), - new NodeHeartbeatSample(NodeStatus.Unknown)); - LOGGER.info( - ManagerMessages.THE_CONFIGNODE_WILL_BE_SHUTDOWN_SOON_MARK_IT_AS_UNKNOWN, - configNodeLocation.getConfigNodeId()); + int configNodeId = configNodeLocation.getConfigNodeId(); + if (NodeStatus.Removing.equals(getLoadManager().getNodeStatus(configNodeId))) { + // Removing has the highest priority and can not be refreshed by the Stopped report + LOGGER.info( + ManagerMessages.LOG_THE_CONFIGNODE_IS_REMOVING_SKIP_MARKING_IT_AS_STOPPED_41B041A3, + configNodeId); + } else { + // Force updating the target ConfigNode's status to Stopped + getLoadManager() + .forceUpdateNodeCache( + NodeType.ConfigNode, configNodeId, new NodeHeartbeatSample(NodeStatus.Stopped)); + LOGGER.info( + ManagerMessages.LOG_THE_CONFIGNODE_WILL_BE_SHUTDOWN_SOON_MARK_IT_AS_STOPPED_D2A64AFD, + configNodeId); + } } return status; } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java index bae45596812e1..a61bd5df2cf82 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/IManager.java @@ -345,7 +345,9 @@ public interface IManager { /** * Report that the specified DataNode will be shutdown. * - *

The ConfigNode-leader will mark it as {@link NodeStatus#Unknown} + *

The ConfigNode-leader will mark it as {@link NodeStatus#Stopped}. A node that is currently + * {@link NodeStatus#Removing} keeps its status. If the report never reaches the leader, the node + * will be marked as {@link NodeStatus#Unknown} by heartbeat timeout instead. * * @return {@link TSStatusCode#SUCCESS_STATUS} if reporting successfully */ @@ -539,7 +541,9 @@ TPermissionInfoResp login( /** * Report that the specified ConfigNode will be shutdown. The ConfigNode-leader will mark it as - * Unknown. + * {@link NodeStatus#Stopped}. A node that is currently {@link NodeStatus#Removing} keeps its + * status. If the report never reaches the leader, the node will be marked as {@link + * NodeStatus#Unknown} by heartbeat timeout instead. * * @return {@link TSStatusCode#SUCCESS_STATUS} if reporting successfully */ diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java index 50b003d0494d4..8dd9ab01a043d 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ProcedureManager.java @@ -59,6 +59,7 @@ import org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProcedurePlan; import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan; import org.apache.iotdb.confignode.i18n.ManagerMessages; +import org.apache.iotdb.confignode.i18n.ProcedureMessages; import org.apache.iotdb.confignode.manager.partition.PartitionManager; import org.apache.iotdb.confignode.manager.subscription.SubscriptionCoordinator; import org.apache.iotdb.confignode.persistence.ProcedureInfo; @@ -777,8 +778,8 @@ public TSStatus checkRemoveDataNodes(List dataNodeLocations) removedDataNodesRegionSet.add(regionMigrationPlan.getRegionId()); } - // 4. Check if there are any other unknown or readonly DataNodes in the consensus group that are - // not the remove DataNodes + // 4. Check if there are any other unknown, stopped or readonly DataNodes in the consensus + // group that are not the remove DataNodes for (TDataNodeLocation removeDataNode : dataNodeLocations) { Set relatedDataNodes = @@ -788,13 +789,14 @@ public TSStatus checkRemoveDataNodes(List dataNodeLocations) for (TDataNodeLocation relatedDataNode : relatedDataNodes) { NodeStatus nodeStatus = getConfigManager().getLoadManager().getNodeStatus(relatedDataNode.getDataNodeId()); - if (nodeStatus == NodeStatus.Unknown || nodeStatus == NodeStatus.ReadOnly) { + // A Stopped node is handled like Unknown: it can not serve the consensus group either + if (nodeStatus == NodeStatus.Unknown + || nodeStatus == NodeStatus.Stopped + || nodeStatus == NodeStatus.ReadOnly) { failMessage = String.format( - "Submit RemoveDataNodesProcedure failed, " - + "because when there are other unknown or readonly nodes in the consensus group that are not remove nodes, " - + "the remove operation cannot be performed for security reasons. " - + "Please check the status of the node %s and ensure it is running.", + ProcedureMessages + .MESSAGE_SUBMIT_REMOVEDATANODESPROCEDURE_FAILED_BECAUSE_WHEN_THERE_ARE_OTHER_UNKNOWN_STOPPED_OR_READONLY_NODES_IN_THE_CONSENSUS_GROUP_THAT_ARE_NOT_REMOVE_NODES_THE_REMOVE_OPERATION_CANNOT_BE_PERFORMED_FOR_SECURITY_REASONS_PLEASE_CHECK_THE_STATUS_OF_THE_NODE_ARG_AND_ENSURE_IT_IS_RUNNING_5063B3F6, relatedDataNode.getDataNodeId()); } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/RetryFailedTasksThread.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/RetryFailedTasksThread.java index 2f222a93ac6d4..b128cd6c5364e 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/RetryFailedTasksThread.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/RetryFailedTasksThread.java @@ -140,7 +140,8 @@ private void triggerDetectTask() { if (nodeStatus == NodeStatus.Running) { oldUnknownNodes.remove(dataNodeLocation); } else if (!oldUnknownNodes.contains(dataNodeLocation) - && nodeStatus == NodeStatus.Unknown) { + // A Stopped node is handled like Unknown: its regions need transfer too + && (nodeStatus == NodeStatus.Unknown || nodeStatus == NodeStatus.Stopped)) { newUnknownNodes.add(dataNodeLocation); } }); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java index 73583151f9819..43350bc0f18d5 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancer.java @@ -84,16 +84,16 @@ public CreateRegionGroupsPlan genRegionGroupsAllocationPlan( throws NotEnoughDataNodeException, DatabaseNotExistsException { // Some new RegionGroups will have to occupy unknown DataNodes if the number of online - // DataNodes is insufficient (Unknown DataNodes are intentionally kept as candidates). - // However, DataNodes that an in-progress RemoveDataNodesProcedure is removing must be - // excluded: placing a new replica on a node that is about to disappear would strand that - // replica and stall the removal forever. A status filter is not enough here, because a + // DataNodes is insufficient (Unknown and Stopped DataNodes are intentionally kept as + // candidates). However, DataNodes that an in-progress RemoveDataNodesProcedure is removing + // must be excluded: placing a new replica on a node that is about to disappear would strand + // that replica and stall the removal forever. A status filter is not enough here, because a // DataNode killed (e.g. kill -9) before removal is reported as Unknown (not Removing) by the // failure detector, so we additionally drop every DataNode that is currently being removed. final Set removingDataNodeIds = getProcedureManager().getRemovingDataNodeIds(); final List availableDataNodes = getNodeManager() - .filterDataNodeThroughStatus(NodeStatus.Running, NodeStatus.Unknown) + .filterDataNodeThroughStatus(NodeStatus.Running, NodeStatus.Unknown, NodeStatus.Stopped) .stream() .filter( dataNode -> !removingDataNodeIds.contains(dataNode.getLocation().getDataNodeId())) diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/ConfigNodeHeartbeatCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/ConfigNodeHeartbeatCache.java index 4d675e1e8a442..3dd3db274b8d6 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/ConfigNodeHeartbeatCache.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/ConfigNodeHeartbeatCache.java @@ -74,6 +74,14 @@ public synchronized void updateCurrentStatistics(boolean forceUpdate) { } } + // The Stopped status is sticky: heartbeat-driven updates (heartbeat failure, failure + // detection) must not refresh a gracefully stopped node back to Unknown. A live status + // reported by a heartbeat (e.g. Running after the node restarts) still revives it. Removing + // needs no extra protection here: the guard above unconditionally keeps it. + if (NodeStatus.Stopped.equals(getNodeStatus()) && NodeStatus.Unknown.equals(status)) { + status = NodeStatus.Stopped; + } + /* Update loadScore */ // Only consider Running ConfigNode as available currently // TODO: Construct load score module diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/DataNodeHeartbeatCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/DataNodeHeartbeatCache.java index e3d6d15c8356b..bd6cdf0360b5f 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/DataNodeHeartbeatCache.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/DataNodeHeartbeatCache.java @@ -78,6 +78,22 @@ public synchronized void updateCurrentStatistics(boolean forceUpdate) { } } + if (NodeStatus.Removing.equals(getNodeStatus())) { + // Removing is the highest-priority sticky status: neither a heartbeat-driven Unknown (e.g. + // the onError path of a broken connection) nor the Stopped report may refresh it. Explicit + // management status changes (e.g. Running on rollback) still apply. + if (NodeStatus.Unknown.equals(status) || NodeStatus.Stopped.equals(status)) { + status = NodeStatus.Removing; + statusReason = null; + } + } else if (NodeStatus.Stopped.equals(getNodeStatus()) && NodeStatus.Unknown.equals(status)) { + // The Stopped status is sticky: heartbeat-driven updates (heartbeat failure, failure + // detection) must not refresh a gracefully stopped node back to Unknown. A live status + // reported by a heartbeat (e.g. Running after the node restarts) still revives it. + status = NodeStatus.Stopped; + statusReason = null; + } + /* Update loadScore */ // Only consider Running DataNode as available currently // TODO: Construct load score module diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/runtime/SubscriptionRuntimeCoordinator.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/runtime/SubscriptionRuntimeCoordinator.java index dd8fe2e70cf7e..228f3e385ada5 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/runtime/SubscriptionRuntimeCoordinator.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/subscription/runtime/SubscriptionRuntimeCoordinator.java @@ -155,6 +155,9 @@ private static NodeStatus getNodeStatus(final NodeStatistics statistics) { } private static boolean isRuntimeSensitiveStatus(final NodeStatus status) { - return status == NodeStatus.Unknown || status == NodeStatus.Removing; + // A Stopped node is handled like Unknown: its runtime leader pairs must be refreshed too + return status == NodeStatus.Unknown + || status == NodeStatus.Removing + || status == NodeStatus.Stopped; } } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java index 6f2f90b739265..4475166ea45a4 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnv.java @@ -1032,10 +1032,13 @@ && isRuntimeActiveWriterNode(oldLeaderNodeId)) { return responseMap; } - private boolean isRuntimeActiveWriterNode(final int dataNodeId) { + boolean isRuntimeActiveWriterNode(final int dataNodeId) { + final NodeStatus nodeStatus = getLoadManager().getNodeStatus(dataNodeId); return dataNodeId >= 0 - && getLoadManager().getNodeStatus(dataNodeId) != NodeStatus.Unknown - && getLoadManager().getNodeStatus(dataNodeId) != NodeStatus.Removing; + && nodeStatus != NodeStatus.Unknown + && nodeStatus != NodeStatus.Removing + // A Stopped node can not serve as an active runtime writer either + && nodeStatus != NodeStatus.Stopped; } private static Map sendPipeMetaRequest( diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java index 4def6fc78b383..be5d5f868cc44 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java @@ -309,7 +309,9 @@ public TSStatus submitDeleteOldRegionPeerTask( new TMaintainPeerReq(regionId, originalDataNode, procedureId); final NodeStatus nodeStatus = getDataNodeStatus(originalDataNode.getDataNodeId()); - final boolean useFullRetry = !NodeStatus.Unknown.equals(nodeStatus); + // A Stopped node does not respond to requests either, so it is handled like Unknown + final boolean useFullRetry = + !NodeStatus.Unknown.equals(nodeStatus) && !NodeStatus.Stopped.equals(nodeStatus); if (!useFullRetry) { LOGGER.info( ProcedureMessages.DATANODE_IS_SUBMIT_DELETE_OLD_REGION_PEER_WITH_A_SINGLE, diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RemoveDataNodeHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RemoveDataNodeHandler.java index 9455d34a0ddb4..1a1544669cc85 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RemoveDataNodeHandler.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RemoveDataNodeHandler.java @@ -111,17 +111,25 @@ public boolean checkEnoughDataNodeAfterRemoving(List removedD .size(); int removedDataNodeSize = - (int) - removedDataNodes.stream() - .filter( - x -> - configManager.getLoadManager().getNodeStatus(x.getDataNodeId()) - != NodeStatus.Unknown) - .count(); + (int) removedDataNodes.stream().filter(this::isCountedInAvailableCapacity).count(); return availableDatanodeSize - removedDataNodeSize >= NodeInfo.getMinimumDataNode(); } + /** + * Whether the specified DataNode currently occupies online capacity. Both this check and {@link + * #checkRegionReplication} subtract only the nodes that still occupy capacity from the available + * DataNode count: an Unknown or Stopped node is already down, so removing it must not consume the + * quota twice. + * + * @return true if the node is neither Unknown nor Stopped + */ + private boolean isCountedInAvailableCapacity(final TDataNodeLocation dataNodeLocation) { + final NodeStatus status = + configManager.getLoadManager().getNodeStatus(dataNodeLocation.getDataNodeId()); + return !NodeStatus.Unknown.equals(status) && !NodeStatus.Stopped.equals(status); + } + /** * Changes the status of a batch of specified DataNodes to the given status. This is done to * prevent the DataNodes from receiving read or write requests when they are being removed or are @@ -234,7 +242,9 @@ public List selectedRegionMigrationPlans( final List availableDataNodes = configManager .getNodeManager() - .filterDataNodeThroughStatus(NodeStatus.Running, NodeStatus.Unknown) + // A Stopped node is handled like Unknown: it can still serve as a migration + // destination when the cluster is short of online nodes + .filterDataNodeThroughStatus(NodeStatus.Running, NodeStatus.Unknown, NodeStatus.Stopped) .stream() .filter(node -> !removedDataNodesSet.contains(node.getLocation().getDataNodeId())) .collect(Collectors.toList()); @@ -595,10 +605,7 @@ public TSStatus checkRegionReplication(RemoveDataNodePlan removeDataNodePlan) { int removedDataNodeSize = (int) removeDataNodePlan.getDataNodeLocations().stream() - .filter( - x -> - configManager.getLoadManager().getNodeStatus(x.getDataNodeId()) - != NodeStatus.Unknown) + .filter(this::isCountedInAvailableCapacity) .count(); if (availableDatanodeSize - removedDataNodeSize < NodeInfo.getMinimumDataNode()) { status.setCode(TSStatusCode.NO_ENOUGH_DATANODE.getStatusCode()); diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java index 69a3cec44005d..69f2fda69be0b 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtils.java @@ -282,9 +282,13 @@ public static Map filterFencedDataNode( final ConfigManager configManager) { return configManager.getNodeManager().getRegisteredDataNodeLocations().entrySet().stream() .filter( - entry -> - configManager.getLoadManager().getNodeStatus(entry.getKey()) != NodeStatus.Unknown - || !DataNodeContactTracker.getInstance().isDataNodeFenced(entry.getKey())) + entry -> { + final NodeStatus status = + configManager.getLoadManager().getNodeStatus(entry.getKey()); + // An Unknown or Stopped node that is additionally fenced is unreachable and skipped + return (status != NodeStatus.Unknown && status != NodeStatus.Stopped) + || !DataNodeContactTracker.getInstance().isDataNodeFenced(entry.getKey()); + }) .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue)); } diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNodeShutdownHook.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNodeShutdownHook.java index 6b894804dc53f..0d8788bf0baa5 100644 --- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNodeShutdownHook.java +++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/service/ConfigNodeShutdownHook.java @@ -49,46 +49,44 @@ public class ConfigNodeShutdownHook extends Thread { public void run() { LOGGER.info(ConfigNodeMessages.CONFIGNODE_EXITING); - boolean isLeader = getConfigNodeInstance().getConfigManager().getConsensusManager().isLeader(); - try { ConfigNode.getInstance().deactivate(); } catch (IOException e) { LOGGER.error(ConfigNodeMessages.MEET_ERROR_WHEN_DEACTIVATE_CONFIGNODE, e); } - if (!isLeader) { - // Set and report shutdown to cluster ConfigNode-leader - CommonDescriptor.getInstance().getConfig().setNodeStatus(NodeStatus.Unknown); - boolean isReportSuccess = false; - TEndPoint seedConfigNode = CONF.getSeedConfigNode(); - for (int retry = 0; retry < SHUTDOWN_REPORT_RETRY_NUM; retry++) { - TSStatus result = - (TSStatus) - SyncConfigNodeClientPool.getInstance() - .sendSyncRequestToConfigNodeWithRetry( - seedConfigNode, - new TConfigNodeLocation( - CONF.getConfigNodeId(), - new TEndPoint(CONF.getInternalAddress(), CONF.getInternalPort()), - new TEndPoint(CONF.getInternalAddress(), CONF.getConsensusPort())), - CnToCnNodeRequestType.REPORT_CONFIG_NODE_SHUTDOWN); + // Set and report shutdown to the cluster ConfigNode-leader best-effort, regardless of + // leadership: a leader that just stepped down may still reach the newly elected leader via + // redirect. If no leader is reachable, the new leader will mark this node Unknown by + // heartbeat timeout instead. + CommonDescriptor.getInstance().getConfig().setNodeStatus(NodeStatus.Stopped); + boolean isReportSuccess = false; + TEndPoint seedConfigNode = CONF.getSeedConfigNode(); + for (int retry = 0; retry < SHUTDOWN_REPORT_RETRY_NUM; retry++) { + TSStatus result = + (TSStatus) + SyncConfigNodeClientPool.getInstance() + .sendSyncRequestToConfigNodeWithRetry( + seedConfigNode, + new TConfigNodeLocation( + CONF.getConfigNodeId(), + new TEndPoint(CONF.getInternalAddress(), CONF.getInternalPort()), + new TEndPoint(CONF.getInternalAddress(), CONF.getConsensusPort())), + CnToCnNodeRequestType.REPORT_CONFIG_NODE_SHUTDOWN); - if (result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { - // Report success - isReportSuccess = true; - break; - } else if (result.getCode() == TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) { - // Redirect - seedConfigNode = result.getRedirectNode(); - } - } - if (!isReportSuccess) { - LOGGER.error( - ConfigNodeMessages - .REPORTING_CONFIGNODE_SHUTDOWN_FAILED_THE_CLUSTER_WILL_STILL_TAKE_THE); + if (result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) { + // Report success + isReportSuccess = true; + break; + } else if (result.getCode() == TSStatusCode.REDIRECTION_RECOMMEND.getStatusCode()) { + // Redirect + seedConfigNode = result.getRedirectNode(); } } + if (!isReportSuccess) { + LOGGER.error( + ConfigNodeMessages.REPORTING_CONFIGNODE_SHUTDOWN_FAILED_THE_CLUSTER_WILL_STILL_TAKE_THE); + } if (LOGGER.isInfoEnabled()) { LOGGER.info( @@ -98,8 +96,4 @@ public void run() { Runtime.getRuntime().totalMemory() - Runtime.getRuntime().freeMemory())); } } - - protected ConfigNode getConfigNodeInstance() { - return ConfigNode.getInstance(); - } } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/ConfigManagerShutdownReportTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/ConfigManagerShutdownReportTest.java new file mode 100644 index 0000000000000..c8c5698beff56 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/ConfigManagerShutdownReportTest.java @@ -0,0 +1,145 @@ +/* + * 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.confignode.manager; + +import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.commons.cluster.NodeType; +import org.apache.iotdb.confignode.manager.load.LoadManager; +import org.apache.iotdb.confignode.manager.load.cache.node.NodeHeartbeatSample; +import org.apache.iotdb.rpc.RpcUtils; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.BeforeClass; +import org.junit.Test; + +import java.io.IOException; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyInt; +import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.clearInvocations; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class ConfigManagerShutdownReportTest { + + private static final TSStatus SUCCESS_STATUS = RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); + + private static ConfigManager CONFIG_MANAGER_SPY; + private static LoadManager LOAD_MANAGER; + + private static final int DATA_NODE_ID = 1; + private static final int CONFIG_NODE_ID = 2; + + private static final TDataNodeLocation DATA_NODE_LOCATION = + new TDataNodeLocation( + DATA_NODE_ID, + new TEndPoint("127.0.0.1", 2000), + new TEndPoint("127.0.0.1", 2001), + new TEndPoint("127.0.0.1", 2002), + new TEndPoint("127.0.0.1", 2003), + new TEndPoint("127.0.0.1", 2004)); + + private static final TConfigNodeLocation CONFIG_NODE_LOCATION = + new TConfigNodeLocation( + CONFIG_NODE_ID, new TEndPoint("127.0.0.1", 1000), new TEndPoint("127.0.0.1", 1001)); + + @BeforeClass + public static void setUp() throws IOException { + final ConfigManager configManager = new ConfigManager(); + CONFIG_MANAGER_SPY = spy(configManager); + LOAD_MANAGER = mock(LoadManager.class); + when(CONFIG_MANAGER_SPY.getLoadManager()).thenReturn(LOAD_MANAGER); + doReturn(SUCCESS_STATUS).when(CONFIG_MANAGER_SPY).confirmLeader(); + } + + @Before + public void clearPreviousInteractions() { + // The shared LoadManager mock keeps interactions from previous tests; the never() checks + // below must only consider the current test. + clearInvocations(LOAD_MANAGER); + } + + @Test + public void testReportDataNodeShutdownSkipsRemovingDataNode() { + when(LOAD_MANAGER.getNodeStatus(DATA_NODE_ID)).thenReturn(NodeStatus.Removing); + + Assert.assertEquals( + SUCCESS_STATUS.getCode(), + CONFIG_MANAGER_SPY.reportDataNodeShutdown(DATA_NODE_LOCATION).getCode()); + + // Removing has the highest priority: the Stopped report must not overwrite it. + verify(LOAD_MANAGER, never()) + .forceUpdateNodeCache(any(NodeType.class), anyInt(), any(NodeHeartbeatSample.class)); + } + + @Test + public void testReportDataNodeShutdownMarksRunningDataNodeStopped() { + when(LOAD_MANAGER.getNodeStatus(DATA_NODE_ID)).thenReturn(NodeStatus.Running); + + Assert.assertEquals( + SUCCESS_STATUS.getCode(), + CONFIG_MANAGER_SPY.reportDataNodeShutdown(DATA_NODE_LOCATION).getCode()); + + verify(LOAD_MANAGER) + .forceUpdateNodeCache( + eq(NodeType.DataNode), + eq(DATA_NODE_ID), + argThat(sample -> sample.getStatus() == NodeStatus.Stopped)); + } + + @Test + public void testReportConfigNodeShutdownSkipsRemovingConfigNode() { + when(LOAD_MANAGER.getNodeStatus(CONFIG_NODE_ID)).thenReturn(NodeStatus.Removing); + + Assert.assertEquals( + SUCCESS_STATUS.getCode(), + CONFIG_MANAGER_SPY.reportConfigNodeShutdown(CONFIG_NODE_LOCATION).getCode()); + + verify(LOAD_MANAGER, never()) + .forceUpdateNodeCache(any(NodeType.class), anyInt(), any(NodeHeartbeatSample.class)); + } + + @Test + public void testReportConfigNodeShutdownMarksRunningConfigNodeStopped() { + when(LOAD_MANAGER.getNodeStatus(CONFIG_NODE_ID)).thenReturn(NodeStatus.Running); + + Assert.assertEquals( + SUCCESS_STATUS.getCode(), + CONFIG_MANAGER_SPY.reportConfigNodeShutdown(CONFIG_NODE_LOCATION).getCode()); + + verify(LOAD_MANAGER) + .forceUpdateNodeCache( + eq(NodeType.ConfigNode), + eq(CONFIG_NODE_ID), + argThat(sample -> sample.getStatus() == NodeStatus.Stopped)); + } +} diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/ProcedureManagerTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/ProcedureManagerTest.java index 4de883b9c158f..0289d0b642097 100644 --- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/ProcedureManagerTest.java +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/ProcedureManagerTest.java @@ -111,6 +111,7 @@ public class ProcedureManagerTest { @BeforeClass public static void setUp() throws IOException { IManager CONFIG_MANAGER = new ConfigManager(); + IManager CONFIG_MANAGER_SPY = spy(CONFIG_MANAGER); ProcedureManager procedureManager = CONFIG_MANAGER.getProcedureManager(); PROCEDURE_MANAGER = spy(procedureManager); @@ -130,6 +131,10 @@ public static void setUp() throws IOException { when(PROCEDURE_EXECUTOR.getProcedures()).thenReturn(procedureMap); when(PROCEDURE_MANAGER.getEnv()).thenReturn(ENV); when(ENV.getRemoveDataNodeHandler()).thenReturn(REMOVE_DATA_NODE_HANDLER); + // Inject the LoadManager spy into the call chain used by checkRemoveDataNodes, so that the + // node-status stubs below take effect on the production path + when(PROCEDURE_MANAGER.getConfigManager()).thenReturn(CONFIG_MANAGER_SPY); + when(CONFIG_MANAGER_SPY.getLoadManager()).thenReturn(LOAD_MANAGER); } @Test @@ -197,6 +202,26 @@ public void testCheckRemoveDataNodeWithAnotherUnknownDataNode() { TSStatus status = PROCEDURE_MANAGER.checkRemoveDataNodes(removedDataNodes); Assert.assertTrue(isFailed(status)); + Assert.assertTrue(status.getMessage().contains("unknown")); + } + + @Test + public void testCheckRemoveDataNodeWithAnotherStoppedDataNode() { + Set relatedDataNodes = new HashSet<>(); + relatedDataNodes.add(removeDataNodeLocationA); + relatedDataNodes.add(coordinatorDataNodeLocation); + + when(REMOVE_DATA_NODE_HANDLER.getRelatedDataNodeLocations(removeDataNodeLocationA)) + .thenReturn(relatedDataNodes); + + when(LOAD_MANAGER.getNodeStatus(removeDataNodeLocationA.getDataNodeId())) + .thenReturn(NodeStatus.Running); + when(LOAD_MANAGER.getNodeStatus(coordinatorDataNodeLocation.getDataNodeId())) + .thenReturn(NodeStatus.Stopped); + + TSStatus status = PROCEDURE_MANAGER.checkRemoveDataNodes(removedDataNodes); + Assert.assertTrue(isFailed(status)); + Assert.assertTrue(status.getMessage().contains("stopped")); } @Test diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/RetryFailedTasksThreadTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/RetryFailedTasksThreadTest.java new file mode 100644 index 0000000000000..972421961ec02 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/RetryFailedTasksThreadTest.java @@ -0,0 +1,107 @@ +/* + * 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.confignode.manager; + +import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration; +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.confignode.manager.load.LoadManager; +import org.apache.iotdb.confignode.manager.node.NodeManager; +import org.apache.iotdb.rpc.RpcUtils; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.Test; + +import java.lang.reflect.Method; +import java.util.Collections; +import java.util.List; + +import static org.mockito.ArgumentMatchers.anyList; +import static org.mockito.ArgumentMatchers.argThat; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class RetryFailedTasksThreadTest { + + private static final int STOPPED_DATA_NODE_ID = 1; + + private final IManager configManager = mock(IManager.class); + private final NodeManager nodeManager = mock(NodeManager.class); + private final LoadManager loadManager = mock(LoadManager.class); + + private final TDataNodeConfiguration stoppedDataNode = + new TDataNodeConfiguration() + .setLocation( + new TDataNodeLocation( + STOPPED_DATA_NODE_ID, + new TEndPoint("127.0.0.1", 2000), + new TEndPoint("127.0.0.1", 2001), + new TEndPoint("127.0.0.1", 2002), + new TEndPoint("127.0.0.1", 2003), + new TEndPoint("127.0.0.1", 2004))); + + private RetryFailedTasksThread retryFailedTasksThread; + + // The thread captures its managers at construction time, so it must be created after the stubs. + private void createThreadWithStoppedDataNode() { + when(configManager.getNodeManager()).thenReturn(nodeManager); + when(configManager.getLoadManager()).thenReturn(loadManager); + when(nodeManager.getRegisteredDataNodes()) + .thenReturn(Collections.singletonList(stoppedDataNode)); + when(loadManager.getNodeStatus(STOPPED_DATA_NODE_ID)).thenReturn(NodeStatus.Stopped); + when(configManager.transfer(anyList())) + .thenReturn(RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS)); + retryFailedTasksThread = new RetryFailedTasksThread(configManager); + } + + private void invokeTriggerDetectTask() throws Exception { + final Method method = RetryFailedTasksThread.class.getDeclaredMethod("triggerDetectTask"); + method.setAccessible(true); + method.invoke(retryFailedTasksThread); + } + + @Test + public void testStoppedDataNodeTriggersRegionTransfer() throws Exception { + createThreadWithStoppedDataNode(); + + invokeTriggerDetectTask(); + + // A Stopped node is handled like Unknown: its regions are transferred. + verify(configManager) + .transfer( + argThat( + (List nodes) -> + nodes.size() == 1 && nodes.get(0).getDataNodeId() == STOPPED_DATA_NODE_ID)); + } + + @Test + public void testContinuingStoppedDataNodeIsNotTransferredTwice() throws Exception { + createThreadWithStoppedDataNode(); + + invokeTriggerDetectTask(); + invokeTriggerDetectTask(); + + // Like a continuing Unknown node, a continuing Stopped node only triggers one transfer. + verify(configManager, times(1)).transfer(anyList()); + } +} diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancerTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancerTest.java new file mode 100644 index 0000000000000..207bae7535245 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/RegionBalancerTest.java @@ -0,0 +1,134 @@ +/* + * 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.confignode.manager.load.balancer; + +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.common.rpc.thrift.TDataNodeConfiguration; +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.confignode.exception.NotEnoughDataNodeException; +import org.apache.iotdb.confignode.manager.IManager; +import org.apache.iotdb.confignode.manager.ProcedureManager; +import org.apache.iotdb.confignode.manager.load.LoadManager; +import org.apache.iotdb.confignode.manager.node.NodeManager; +import org.apache.iotdb.confignode.manager.partition.PartitionManager; +import org.apache.iotdb.confignode.manager.schema.ClusterSchemaManager; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; + +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class RegionBalancerTest { + + private static final int STOPPED_DATA_NODE_ID = 1; + private static final int REMOVING_DATA_NODE_ID = 2; + + private final IManager configManager = mock(IManager.class); + private final NodeManager nodeManager = mock(NodeManager.class); + private final ProcedureManager procedureManager = mock(ProcedureManager.class); + private final ClusterSchemaManager clusterSchemaManager = mock(ClusterSchemaManager.class); + private final PartitionManager partitionManager = mock(PartitionManager.class); + private final LoadManager loadManager = mock(LoadManager.class); + + private final TDataNodeConfiguration stoppedDataNode = + new TDataNodeConfiguration() + .setLocation( + new TDataNodeLocation( + STOPPED_DATA_NODE_ID, + new TEndPoint("127.0.0.1", 2000), + new TEndPoint("127.0.0.1", 2001), + new TEndPoint("127.0.0.1", 2002), + new TEndPoint("127.0.0.1", 2003), + new TEndPoint("127.0.0.1", 2004))); + private final TDataNodeConfiguration removingDataNode = + new TDataNodeConfiguration() + .setLocation( + new TDataNodeLocation( + REMOVING_DATA_NODE_ID, + new TEndPoint("127.0.0.1", 3000), + new TEndPoint("127.0.0.1", 3001), + new TEndPoint("127.0.0.1", 3002), + new TEndPoint("127.0.0.1", 3003), + new TEndPoint("127.0.0.1", 3004))); + + private final RegionBalancer regionBalancer = new RegionBalancer(configManager); + + @Before + public void setUp() { + when(configManager.getNodeManager()).thenReturn(nodeManager); + when(configManager.getProcedureManager()).thenReturn(procedureManager); + when(configManager.getClusterSchemaManager()).thenReturn(clusterSchemaManager); + when(configManager.getPartitionManager()).thenReturn(partitionManager); + when(configManager.getLoadManager()).thenReturn(loadManager); + } + + @Test + public void testStoppedDataNodesAreKeptAsAllocationCandidates() throws Exception { + // With an empty allotment only the candidate filtering runs. The stub below locks the exact + // status candidate list: if Stopped is dropped from the production call, the stub no longer + // matches and the test fails. + when(procedureManager.getRemovingDataNodeIds()).thenReturn(Collections.emptySet()); + when(nodeManager.filterDataNodeThroughStatus( + NodeStatus.Running, NodeStatus.Unknown, NodeStatus.Stopped)) + .thenReturn(Arrays.asList(stoppedDataNode, removingDataNode)); + + regionBalancer.genRegionGroupsAllocationPlan( + Collections.emptyMap(), TConsensusGroupType.DataRegion); + + verify(nodeManager) + .filterDataNodeThroughStatus(NodeStatus.Running, NodeStatus.Unknown, NodeStatus.Stopped); + } + + @Test + public void testRemovingDataNodesAreExcludedFromCandidates() throws Exception { + when(procedureManager.getRemovingDataNodeIds()) + .thenReturn(new HashSet<>(Collections.singletonList(REMOVING_DATA_NODE_ID))); + // The mocked status filter returns both nodes; the in-progress removal must drop the second. + when(nodeManager.filterDataNodeThroughStatus( + NodeStatus.Running, NodeStatus.Unknown, NodeStatus.Stopped)) + .thenReturn(Arrays.asList(stoppedDataNode, removingDataNode)); + when(clusterSchemaManager.getReplicationFactor(eq("db1"), eq(TConsensusGroupType.DataRegion))) + .thenReturn(3); + + try { + regionBalancer.genRegionGroupsAllocationPlan( + Collections.singletonMap("db1", 1), TConsensusGroupType.DataRegion); + Assert.fail("Expected NotEnoughDataNodeException"); + } catch (NotEnoughDataNodeException e) { + // The remaining candidates keep the Stopped node but not the node being removed. + Assert.assertTrue( + "Stopped node should be kept: " + e.getMessage(), + e.getMessage().contains("dataNodeId:" + STOPPED_DATA_NODE_ID)); + Assert.assertFalse( + "Removing node should be excluded: " + e.getMessage(), + e.getMessage().contains("dataNodeId:" + REMOVING_DATA_NODE_ID)); + } + } +} 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 a400692956621..02cd4c531797d 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 @@ -48,4 +48,101 @@ public void updateStatisticsTest() { Assert.assertEquals(NodeStatus.Running, configNodeHeartbeatCache.getNodeStatus()); Assert.assertEquals(0, configNodeHeartbeatCache.getLoadScore()); } + + @Test + public void stoppedStatusStickyAndRevivalTest() { + // Test DataNode heartbeat cache + DataNodeHeartbeatCache dataNodeHeartbeatCache = new DataNodeHeartbeatCache(1); + // A fresh Stopped report (shutdown hook) marks the node as Stopped + dataNodeHeartbeatCache.cacheHeartbeatSample( + new NodeHeartbeatSample(System.nanoTime(), NodeStatus.Stopped)); + dataNodeHeartbeatCache.updateCurrentStatistics(true); + Assert.assertEquals(NodeStatus.Stopped, dataNodeHeartbeatCache.getNodeStatus()); + Assert.assertEquals(Long.MAX_VALUE, dataNodeHeartbeatCache.getLoadScore()); + + // A forced Unknown update (e.g. heartbeat connection failure) must not refresh Stopped + dataNodeHeartbeatCache.cacheHeartbeatSample(new NodeHeartbeatSample(NodeStatus.Unknown)); + dataNodeHeartbeatCache.updateCurrentStatistics(true); + Assert.assertEquals(NodeStatus.Stopped, dataNodeHeartbeatCache.getNodeStatus()); + + // Periodic updates must not refresh Stopped to Unknown either + dataNodeHeartbeatCache.updateCurrentStatistics(false); + Assert.assertEquals(NodeStatus.Stopped, dataNodeHeartbeatCache.getNodeStatus()); + + // A live heartbeat (e.g. the node restarted) revives the node + dataNodeHeartbeatCache.cacheHeartbeatSample( + new NodeHeartbeatSample(System.nanoTime(), NodeStatus.Running)); + dataNodeHeartbeatCache.updateCurrentStatistics(false); + Assert.assertEquals(NodeStatus.Running, dataNodeHeartbeatCache.getNodeStatus()); + + // Removing has the highest priority: it refreshes Stopped + dataNodeHeartbeatCache.cacheHeartbeatSample( + new NodeHeartbeatSample(System.nanoTime(), NodeStatus.Stopped)); + dataNodeHeartbeatCache.updateCurrentStatistics(true); + Assert.assertEquals(NodeStatus.Stopped, dataNodeHeartbeatCache.getNodeStatus()); + dataNodeHeartbeatCache.cacheHeartbeatSample( + new NodeHeartbeatSample(System.nanoTime(), NodeStatus.Removing)); + dataNodeHeartbeatCache.updateCurrentStatistics(true); + Assert.assertEquals(NodeStatus.Removing, dataNodeHeartbeatCache.getNodeStatus()); + // And the Stopped report must not refresh Removing + dataNodeHeartbeatCache.cacheHeartbeatSample( + new NodeHeartbeatSample(System.nanoTime(), NodeStatus.Stopped)); + dataNodeHeartbeatCache.updateCurrentStatistics(true); + Assert.assertEquals(NodeStatus.Removing, dataNodeHeartbeatCache.getNodeStatus()); + // A forced Unknown update (e.g. heartbeat connection failure) must not refresh Removing + // either + dataNodeHeartbeatCache.cacheHeartbeatSample(new NodeHeartbeatSample(NodeStatus.Unknown)); + dataNodeHeartbeatCache.updateCurrentStatistics(true); + Assert.assertEquals(NodeStatus.Removing, dataNodeHeartbeatCache.getNodeStatus()); + // An explicit management status change (e.g. rollback to Running) still applies + dataNodeHeartbeatCache.cacheHeartbeatSample( + new NodeHeartbeatSample(System.nanoTime(), NodeStatus.Running)); + dataNodeHeartbeatCache.updateCurrentStatistics(true); + Assert.assertEquals(NodeStatus.Running, dataNodeHeartbeatCache.getNodeStatus()); + + // Test ConfigNode heartbeat cache + ConfigNodeHeartbeatCache configNodeHeartbeatCache = new ConfigNodeHeartbeatCache(2); + configNodeHeartbeatCache.cacheHeartbeatSample( + new NodeHeartbeatSample(System.nanoTime(), NodeStatus.Stopped)); + configNodeHeartbeatCache.updateCurrentStatistics(true); + Assert.assertEquals(NodeStatus.Stopped, configNodeHeartbeatCache.getNodeStatus()); + + // A forced Unknown update (e.g. heartbeat connection failure) must not refresh Stopped + configNodeHeartbeatCache.cacheHeartbeatSample(new NodeHeartbeatSample(NodeStatus.Unknown)); + configNodeHeartbeatCache.updateCurrentStatistics(true); + Assert.assertEquals(NodeStatus.Stopped, configNodeHeartbeatCache.getNodeStatus()); + + // Periodic updates must not refresh Stopped to Unknown either + configNodeHeartbeatCache.updateCurrentStatistics(false); + Assert.assertEquals(NodeStatus.Stopped, configNodeHeartbeatCache.getNodeStatus()); + + // A live heartbeat revives the ConfigNode + configNodeHeartbeatCache.cacheHeartbeatSample( + new NodeHeartbeatSample(System.nanoTime(), NodeStatus.Running)); + configNodeHeartbeatCache.updateCurrentStatistics(false); + Assert.assertEquals(NodeStatus.Running, configNodeHeartbeatCache.getNodeStatus()); + + // Removing has the highest priority: the Stopped report must not refresh it + configNodeHeartbeatCache.cacheHeartbeatSample( + new NodeHeartbeatSample(System.nanoTime(), NodeStatus.Stopped)); + configNodeHeartbeatCache.updateCurrentStatistics(true); + Assert.assertEquals(NodeStatus.Stopped, configNodeHeartbeatCache.getNodeStatus()); + configNodeHeartbeatCache.cacheHeartbeatSample( + new NodeHeartbeatSample(System.nanoTime(), NodeStatus.Removing)); + configNodeHeartbeatCache.updateCurrentStatistics(true); + Assert.assertEquals(NodeStatus.Removing, configNodeHeartbeatCache.getNodeStatus()); + configNodeHeartbeatCache.cacheHeartbeatSample( + new NodeHeartbeatSample(System.nanoTime(), NodeStatus.Stopped)); + configNodeHeartbeatCache.updateCurrentStatistics(true); + Assert.assertEquals(NodeStatus.Removing, configNodeHeartbeatCache.getNodeStatus()); + // Unlike the DataNode cache, the ConfigNode cache unconditionally keeps Removing against any + // update (pre-existing guard), including forced ones + configNodeHeartbeatCache.cacheHeartbeatSample(new NodeHeartbeatSample(NodeStatus.Unknown)); + configNodeHeartbeatCache.updateCurrentStatistics(true); + Assert.assertEquals(NodeStatus.Removing, configNodeHeartbeatCache.getNodeStatus()); + configNodeHeartbeatCache.cacheHeartbeatSample( + new NodeHeartbeatSample(System.nanoTime(), NodeStatus.Running)); + configNodeHeartbeatCache.updateCurrentStatistics(true); + Assert.assertEquals(NodeStatus.Removing, configNodeHeartbeatCache.getNodeStatus()); + } } diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/subscription/runtime/SubscriptionRuntimeCoordinatorTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/subscription/runtime/SubscriptionRuntimeCoordinatorTest.java new file mode 100644 index 0000000000000..f3c0aab1a767c --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/subscription/runtime/SubscriptionRuntimeCoordinatorTest.java @@ -0,0 +1,119 @@ +/* + * 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.confignode.manager.subscription.runtime; + +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.commons.subscription.meta.topic.TopicMeta; +import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.confignode.manager.ProcedureManager; +import org.apache.iotdb.confignode.manager.load.LoadManager; +import org.apache.iotdb.confignode.manager.load.cache.node.NodeStatistics; +import org.apache.iotdb.confignode.manager.load.subscriber.NodeStatisticsChangeEvent; +import org.apache.iotdb.confignode.manager.subscription.SubscriptionCoordinator; +import org.apache.iotdb.confignode.manager.subscription.SubscriptionManager; +import org.apache.iotdb.confignode.persistence.subscription.SubscriptionInfo; +import org.apache.iotdb.rpc.subscription.config.TopicConfig; + +import org.apache.tsfile.utils.Pair; +import org.junit.Test; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyLong; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +public class SubscriptionRuntimeCoordinatorTest { + + private static final int DATA_NODE_ID = 1; + + private final ConfigManager configManager = mock(ConfigManager.class); + private final SubscriptionManager subscriptionManager = mock(SubscriptionManager.class); + private final SubscriptionCoordinator subscriptionCoordinator = + mock(SubscriptionCoordinator.class); + private final SubscriptionInfo subscriptionInfo = mock(SubscriptionInfo.class); + private final TopicMeta topicMeta = mock(TopicMeta.class); + private final TopicConfig topicConfig = mock(TopicConfig.class); + private final LoadManager loadManager = mock(LoadManager.class); + private final ProcedureManager procedureManager = mock(ProcedureManager.class); + + private final SubscriptionRuntimeCoordinator coordinator = + new SubscriptionRuntimeCoordinator(configManager); + + private void stubConsensusBasedTopic() { + when(configManager.getSubscriptionManager()).thenReturn(subscriptionManager); + when(subscriptionManager.getSubscriptionCoordinator()).thenReturn(subscriptionCoordinator); + when(subscriptionCoordinator.getSubscriptionInfo()).thenReturn(subscriptionInfo); + when(subscriptionInfo.getAllTopicMeta()).thenReturn(Collections.singletonList(topicMeta)); + when(topicMeta.getConfig()).thenReturn(topicConfig); + when(topicConfig.isIncrementalMode()).thenReturn(true); + + when(configManager.getLoadManager()).thenReturn(loadManager); + // A seeded DataRegion leader pair makes the refresh map non-empty. + when(loadManager.getRegionLeaderMap()) + .thenReturn( + Collections.singletonMap( + new TConsensusGroupId(TConsensusGroupType.DataRegion, 1), DATA_NODE_ID)); + when(configManager.getProcedureManager()).thenReturn(procedureManager); + } + + private NodeStatisticsChangeEvent eventOf( + final NodeStatus oldStatus, final NodeStatus newStatus) { + final Map> map = new HashMap<>(); + map.put(DATA_NODE_ID, new Pair<>(new NodeStatistics(oldStatus), new NodeStatistics(newStatus))); + return new NodeStatisticsChangeEvent(map); + } + + @Test + public void testStoppedStatusTriggersRuntimeRefresh() { + stubConsensusBasedTopic(); + + coordinator.handleNodeStatisticsChange(eventOf(NodeStatus.Running, NodeStatus.Stopped)); + + // A Stopped node is handled like Unknown/Removing: its runtime leader pairs are refreshed. + verify(procedureManager).subscriptionHandleLeaderChange(any(Map.class), anyLong()); + } + + @Test + public void testUnknownStatusTriggersRuntimeRefresh() { + stubConsensusBasedTopic(); + + coordinator.handleNodeStatisticsChange(eventOf(NodeStatus.Running, NodeStatus.Unknown)); + + verify(procedureManager).subscriptionHandleLeaderChange(any(Map.class), anyLong()); + } + + @Test + public void testReadOnlyStatusDoesNotTriggerRuntimeRefresh() { + stubConsensusBasedTopic(); + + coordinator.handleNodeStatisticsChange(eventOf(NodeStatus.Running, NodeStatus.ReadOnly)); + + // ReadOnly is not runtime-sensitive: the runtime leader pairs stay untouched. + verify(procedureManager, never()).subscriptionHandleLeaderChange(any(Map.class), anyLong()); + } +} diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnvTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnvTest.java new file mode 100644 index 0000000000000..7691eb45848ee --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/ConfigNodeProcedureEnvTest.java @@ -0,0 +1,93 @@ +/* + * 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.confignode.procedure.env; + +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.confignode.manager.load.LoadManager; +import org.apache.iotdb.confignode.procedure.scheduler.ProcedureScheduler; + +import org.junit.Assert; +import org.junit.Before; +import org.junit.Test; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class ConfigNodeProcedureEnvTest { + + private static final int DATA_NODE_ID = 1; + + private final ConfigManager configManager = mock(ConfigManager.class); + private final LoadManager loadManager = mock(LoadManager.class); + private final ConfigNodeProcedureEnv env = + new ConfigNodeProcedureEnv(configManager, mock(ProcedureScheduler.class)); + + @Before + public void setUp() { + when(configManager.getLoadManager()).thenReturn(loadManager); + } + + @Test + public void testStoppedNodeIsNotRuntimeActiveWriter() { + when(loadManager.getNodeStatus(DATA_NODE_ID)).thenReturn(NodeStatus.Stopped); + + // A Stopped node is handled like Unknown: it can not serve as an active runtime writer. + Assert.assertFalse(env.isRuntimeActiveWriterNode(DATA_NODE_ID)); + } + + @Test + public void testUnknownNodeIsNotRuntimeActiveWriter() { + when(loadManager.getNodeStatus(DATA_NODE_ID)).thenReturn(NodeStatus.Unknown); + + Assert.assertFalse(env.isRuntimeActiveWriterNode(DATA_NODE_ID)); + } + + @Test + public void testRemovingNodeIsNotRuntimeActiveWriter() { + when(loadManager.getNodeStatus(DATA_NODE_ID)).thenReturn(NodeStatus.Removing); + + Assert.assertFalse(env.isRuntimeActiveWriterNode(DATA_NODE_ID)); + } + + @Test + public void testRunningNodeIsRuntimeActiveWriter() { + when(loadManager.getNodeStatus(DATA_NODE_ID)).thenReturn(NodeStatus.Running); + + Assert.assertTrue(env.isRuntimeActiveWriterNode(DATA_NODE_ID)); + } + + @Test + public void testReadOnlyNodeIsRuntimeActiveWriter() { + when(loadManager.getNodeStatus(DATA_NODE_ID)).thenReturn(NodeStatus.ReadOnly); + + // ReadOnly still responds and can serve as a runtime writer, unlike Unknown/Stopped/Removing. + Assert.assertTrue(env.isRuntimeActiveWriterNode(DATA_NODE_ID)); + } + + @Test + public void testNegativeNodeIdIsNotRuntimeActiveWriter() { + // The node id guard dominates the status check: even a Running status can not turn an + // invalid node id into an active writer. + when(loadManager.getNodeStatus(-1)).thenReturn(NodeStatus.Running); + + Assert.assertFalse(env.isRuntimeActiveWriterNode(-1)); + } +} diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerTest.java new file mode 100644 index 0000000000000..11ca557928bc7 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandlerTest.java @@ -0,0 +1,120 @@ +/* + * 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.confignode.procedure.env; + +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupId; +import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType; +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.common.rpc.thrift.TSStatus; +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.confignode.client.sync.CnToDnSyncRequestType; +import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.rpc.RpcUtils; +import org.apache.iotdb.rpc.TSStatusCode; + +import org.junit.Assert; +import org.junit.Test; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.eq; +import static org.mockito.Mockito.doReturn; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.spy; +import static org.mockito.Mockito.verify; + +public class RegionMaintainHandlerTest { + + private static final int DATA_NODE_ID = 1; + + private final ConfigManager configManager = mock(ConfigManager.class); + private final RegionMaintainHandler handler = spy(new RegionMaintainHandler(configManager)); + + private final TDataNodeLocation dataNodeLocation = + new TDataNodeLocation( + DATA_NODE_ID, + new TEndPoint("127.0.0.1", 2000), + new TEndPoint("127.0.0.1", 2001), + new TEndPoint("127.0.0.1", 2002), + new TEndPoint("127.0.0.1", 2003), + new TEndPoint("127.0.0.1", 2004)); + + private final TConsensusGroupId regionId = + new TConsensusGroupId(TConsensusGroupType.DataRegion, 1); + + private void stubStatusAndSubmit(final NodeStatus status, final TSStatus submitResult) { + doReturn(status).when(handler).getDataNodeStatus(DATA_NODE_ID); + doReturn(submitResult) + .when(handler) + .submitDataNodeSyncRequest( + any(TEndPoint.class), + any(Object.class), + any(CnToDnSyncRequestType.class), + any(Boolean.class)); + } + + @Test + public void testDeleteOldRegionPeerUsesSingleRetryForStoppedDataNode() { + final TSStatus success = RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); + stubStatusAndSubmit(NodeStatus.Stopped, success); + + final TSStatus status = handler.submitDeleteOldRegionPeerTask(1L, dataNodeLocation, regionId); + + Assert.assertEquals(success.getCode(), status.getCode()); + // A Stopped node does not respond to requests either: like Unknown, it gets a single retry. + verify(handler) + .submitDataNodeSyncRequest( + any(TEndPoint.class), + any(Object.class), + eq(CnToDnSyncRequestType.DELETE_OLD_REGION_PEER), + eq(false)); + } + + @Test + public void testDeleteOldRegionPeerUsesFullRetryForRunningDataNode() { + final TSStatus success = RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); + stubStatusAndSubmit(NodeStatus.Running, success); + + handler.submitDeleteOldRegionPeerTask(1L, dataNodeLocation, regionId); + + verify(handler) + .submitDataNodeSyncRequest( + any(TEndPoint.class), + any(Object.class), + eq(CnToDnSyncRequestType.DELETE_OLD_REGION_PEER), + eq(true)); + } + + @Test + public void testDeleteOldRegionPeerUsesFullRetryForReadOnlyDataNode() { + final TSStatus success = RpcUtils.getStatus(TSStatusCode.SUCCESS_STATUS); + stubStatusAndSubmit(NodeStatus.ReadOnly, success); + + handler.submitDeleteOldRegionPeerTask(1L, dataNodeLocation, regionId); + + // Only Unknown and Stopped are treated as down; ReadOnly still responds. + verify(handler) + .submitDataNodeSyncRequest( + any(TEndPoint.class), + any(Object.class), + eq(CnToDnSyncRequestType.DELETE_OLD_REGION_PEER), + eq(true)); + } +} diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtilsTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtilsTest.java new file mode 100644 index 0000000000000..005cd93211f43 --- /dev/null +++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/schema/SchemaUtilsTest.java @@ -0,0 +1,91 @@ +/* + * 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.confignode.procedure.impl.schema; + +import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation; +import org.apache.iotdb.common.rpc.thrift.TEndPoint; +import org.apache.iotdb.commons.cluster.NodeStatus; +import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor; +import org.apache.iotdb.confignode.manager.ConfigManager; +import org.apache.iotdb.confignode.manager.load.LoadManager; +import org.apache.iotdb.confignode.manager.node.NodeManager; + +import org.junit.Assert; +import org.junit.Test; + +import java.util.HashMap; +import java.util.Map; + +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; + +public class SchemaUtilsTest { + + private static final int STOPPED_DATA_NODE_ID = 1; + private static final int UNKNOWN_DATA_NODE_ID = 2; + private static final int RUNNING_DATA_NODE_ID = 3; + + private static TDataNodeLocation locationOf(final int dataNodeId) { + return new TDataNodeLocation( + dataNodeId, + new TEndPoint("127.0.0.1", dataNodeId * 1000), + new TEndPoint("127.0.0.1", dataNodeId * 1000 + 1), + new TEndPoint("127.0.0.1", dataNodeId * 1000 + 2), + new TEndPoint("127.0.0.1", dataNodeId * 1000 + 3), + new TEndPoint("127.0.0.1", dataNodeId * 1000 + 4)); + } + + @Test + public void testFilterFencedDataNodeTreatsStoppedLikeUnknown() { + final ConfigManager configManager = mock(ConfigManager.class); + final NodeManager nodeManager = mock(NodeManager.class); + final LoadManager loadManager = mock(LoadManager.class); + when(configManager.getNodeManager()).thenReturn(nodeManager); + when(configManager.getLoadManager()).thenReturn(loadManager); + + final Map registered = new HashMap<>(); + registered.put(STOPPED_DATA_NODE_ID, locationOf(STOPPED_DATA_NODE_ID)); + registered.put(UNKNOWN_DATA_NODE_ID, locationOf(UNKNOWN_DATA_NODE_ID)); + registered.put(RUNNING_DATA_NODE_ID, locationOf(RUNNING_DATA_NODE_ID)); + when(nodeManager.getRegisteredDataNodeLocations()).thenReturn(registered); + when(loadManager.getNodeStatus(STOPPED_DATA_NODE_ID)).thenReturn(NodeStatus.Stopped); + when(loadManager.getNodeStatus(UNKNOWN_DATA_NODE_ID)).thenReturn(NodeStatus.Unknown); + when(loadManager.getNodeStatus(RUNNING_DATA_NODE_ID)).thenReturn(NodeStatus.Running); + + // A negative fence threshold makes every never-contacted DataNode read as fenced, so the + // status condition decides alone. + final long originalFenceMs = + ConfigNodeDescriptor.getInstance().getConf().getMetadataLeaseFenceMs(); + try { + ConfigNodeDescriptor.getInstance().getConf().setMetadataLeaseFenceMs(-100_000L); + + final Map filtered = + SchemaUtils.filterFencedDataNode(configManager); + + // An additionally fenced Stopped node is skipped, exactly like an additionally fenced + // Unknown node. A Running node is kept even when fenced. + Assert.assertFalse(filtered.containsKey(STOPPED_DATA_NODE_ID)); + Assert.assertFalse(filtered.containsKey(UNKNOWN_DATA_NODE_ID)); + Assert.assertTrue(filtered.containsKey(RUNNING_DATA_NODE_ID)); + } finally { + ConfigNodeDescriptor.getInstance().getConf().setMetadataLeaseFenceMs(originalFenceMs); + } + } +} diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java index 518a9faaed2ec..7de0c8392d6fd 100644 --- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java +++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/cluster/NodeStatus.java @@ -33,7 +33,11 @@ public enum NodeStatus { Removing("Removing"), /** Only query statements are permitted */ - ReadOnly("ReadOnly"); + ReadOnly("ReadOnly"), + + /** Node was stopped intentionally and reported its shutdown */ + Stopped("Stopped"); + public static final String DISK_FULL = "DiskFull"; private final String status; @@ -67,6 +71,7 @@ public static boolean isReadable(NodeStatus status) { case ReadOnly: return true; case Unknown: + case Stopped: return false; default: throw new UnsupportedOperationException(