Skip to content

Commit d2eaf42

Browse files
d-wang-commitd-wang-commit
committed
Add Stopped node status
1 parent 8a789d0 commit d2eaf42

38 files changed

Lines changed: 1301 additions & 202 deletions

File tree

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

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@
1919

2020
package org.apache.iotdb.it.env.cluster;
2121

22+
import org.apache.iotdb.commons.cluster.NodeStatus;
2223
import org.apache.iotdb.it.framework.IoTDBTestLogger;
2324

2425
import org.apache.tsfile.external.commons.lang3.SystemUtils;
@@ -69,6 +70,17 @@
6970

7071
public class EnvUtils {
7172

73+
/**
74+
* The status a node locally stopped via {@code AbstractNodeWrapper.stop()} is expected to be in.
75+
* On Windows, {@code Process.destroy()} terminates the node process without running the JVM
76+
* shutdown hooks, so the graceful-shutdown report is never sent and the ConfigNode marks the node
77+
* Unknown by heartbeat timeout. On Unix, the shutdown hook reports the stop and the node becomes
78+
* Stopped.
79+
*/
80+
public static NodeStatus getNodeStatusAfterLocalStop() {
81+
return SystemUtils.IS_OS_WINDOWS ? NodeStatus.Unknown : NodeStatus.Stopped;
82+
}
83+
7284
public static int[] searchAvailablePorts() {
7385
int length = 10;
7486
while (true) {

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

Lines changed: 25 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -212,15 +212,35 @@ public void testIllegalNodeRestart()
212212
dataNodeRestartResp.getStatus().getCode());
213213
Assert.assertTrue(dataNodeRestartResp.getStatus().getMessage().contains("whose nodeId="));
214214

215-
// Shutdown and check
215+
// Shutdown and check. A gracefully stopped node is reported as Stopped by its shutdown
216+
// hook. A ConfigNode that was the leader at shutdown time can not report itself to another
217+
// leader, so it may remain Unknown on the newly elected leader.
216218
EnvFactory.getEnv().shutdownConfigNode(1);
217219
EnvFactory.getEnv().shutdownDataNode(0);
218220
EnvFactory.getEnv()
219221
.ensureNodeStatus(
220-
Arrays.asList(
221-
EnvFactory.getEnv().getConfigNodeWrapper(1),
222-
EnvFactory.getEnv().getDataNodeWrapper(0)),
223-
Arrays.asList(NodeStatus.Unknown, NodeStatus.Unknown));
222+
Arrays.asList(EnvFactory.getEnv().getDataNodeWrapper(0)),
223+
Arrays.asList(NodeStatus.Stopped));
224+
boolean isConfigNodeDown = false;
225+
for (int retry = 0; retry < 30; retry++) {
226+
TShowClusterResp showClusterResp = client.showCluster();
227+
for (TConfigNodeLocation configNodeLocation : showClusterResp.getConfigNodeList()) {
228+
if (configNodeLocation.getConsensusEndPoint().getPort()
229+
== registeredConfigNodeWrapper.getConsensusPort()) {
230+
String configNodeStatus =
231+
showClusterResp.getNodeStatus().get(configNodeLocation.getConfigNodeId());
232+
if (NodeStatus.Stopped.getStatus().equals(configNodeStatus)
233+
|| NodeStatus.Unknown.getStatus().equals(configNodeStatus)) {
234+
isConfigNodeDown = true;
235+
}
236+
}
237+
}
238+
if (isConfigNodeDown) {
239+
break;
240+
}
241+
Thread.sleep(1000);
242+
}
243+
Assert.assertTrue(isConfigNodeDown);
224244

225245
/* Restart and updatePeer */
226246
// TODO: Delete this IT after enable modify internal TEndPoints

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

Lines changed: 42 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,8 @@
1919

2020
package org.apache.iotdb.confignode.it.cluster;
2121

22+
import org.apache.iotdb.common.rpc.thrift.TConfigNodeLocation;
23+
import org.apache.iotdb.common.rpc.thrift.TDataNodeLocation;
2224
import org.apache.iotdb.commons.client.exception.ClientManagerException;
2325
import org.apache.iotdb.commons.client.sync.SyncConfigNodeIServiceClient;
2426
import org.apache.iotdb.commons.cluster.NodeStatus;
@@ -39,7 +41,6 @@
3941

4042
import java.io.IOException;
4143
import java.util.concurrent.TimeUnit;
42-
import java.util.concurrent.atomic.AtomicInteger;
4344

4445
@RunWith(IoTDBTestRunner.class)
4546
@Category({ClusterIT.class})
@@ -75,29 +76,56 @@ public void testNodeShutdownReporter()
7576
try (SyncConfigNodeIServiceClient client =
7677
(SyncConfigNodeIServiceClient) EnvFactory.getEnv().getLeaderConfigNodeConnection()) {
7778

78-
// The unknown Nodes should be detected immediately with the help of shutdown hook
79+
// The stopped Nodes should be detected immediately with the help of shutdown hook. A
80+
// ConfigNode whose report can not reach the newly elected leader remains Unknown instead.
81+
TShowClusterResp showClusterResp = client.showCluster();
82+
Assert.assertEquals(
83+
TSStatusCode.SUCCESS_STATUS.getStatusCode(), showClusterResp.getStatus().getCode());
84+
85+
int stoppedDataNodeId = -1;
86+
for (TDataNodeLocation dataNodeLocation : showClusterResp.getDataNodeList()) {
87+
if (dataNodeLocation.getInternalEndPoint().getPort()
88+
== EnvFactory.getEnv().getDataNodeWrapper(0).getInternalPort()) {
89+
stoppedDataNodeId = dataNodeLocation.getDataNodeId();
90+
}
91+
}
92+
Assert.assertNotEquals(-1, stoppedDataNodeId);
93+
94+
int stoppedConfigNodeId = -1;
95+
for (TConfigNodeLocation configNodeLocation : showClusterResp.getConfigNodeList()) {
96+
if (configNodeLocation.getConsensusEndPoint().getPort()
97+
== EnvFactory.getEnv().getConfigNodeWrapper(1).getConsensusPort()) {
98+
stoppedConfigNodeId = configNodeLocation.getConfigNodeId();
99+
}
100+
}
101+
Assert.assertNotEquals(-1, stoppedConfigNodeId);
102+
79103
boolean isDetected = false;
80104
for (int retry = 0; retry < 5; retry++) {
81-
TShowClusterResp showClusterResp = client.showCluster();
105+
showClusterResp = client.showCluster();
82106
Assert.assertEquals(
83107
TSStatusCode.SUCCESS_STATUS.getStatusCode(), showClusterResp.getStatus().getCode());
84-
AtomicInteger unknownNum = new AtomicInteger(0);
85-
showClusterResp
86-
.getNodeStatus()
87-
.forEach(
88-
(nodeId, nodeStatus) -> {
89-
if (NodeStatus.Unknown.getStatus().equals(nodeStatus)) {
90-
unknownNum.getAndIncrement();
91-
}
92-
});
93-
if (unknownNum.get() == 2) {
108+
109+
// The stopped DataNode must be observable as Stopped
110+
final String dataNodeStatus = showClusterResp.getNodeStatus().get(stoppedDataNodeId);
111+
final boolean isDataNodeStopped = NodeStatus.Stopped.getStatus().equals(dataNodeStatus);
112+
113+
// The stopped ConfigNode is Stopped when its report reached a leader, and may otherwise
114+
// remain Unknown until heartbeat timeout
115+
final String configNodeStatus = showClusterResp.getNodeStatus().get(stoppedConfigNodeId);
116+
final boolean isConfigNodeDetected =
117+
NodeStatus.Stopped.getStatus().equals(configNodeStatus)
118+
|| NodeStatus.Unknown.getStatus().equals(configNodeStatus);
119+
120+
if (isDataNodeStopped && isConfigNodeDetected) {
94121
isDetected = true;
95122
break;
96123
}
97124

98125
TimeUnit.SECONDS.sleep(1);
99126
}
100-
Assert.assertTrue(isDetected);
127+
Assert.assertTrue(
128+
"Timed out waiting for the stopped DataNode and ConfigNode to be detected", isDetected);
101129
}
102130
}
103131
}

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

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -208,7 +208,7 @@ public void testCFDWithUnknownStatus() throws Exception {
208208
EnvFactory.getEnv()
209209
.ensureNodeStatus(
210210
Collections.singletonList(EnvFactory.getEnv().getDataNodeWrapper(0)),
211-
Collections.singletonList(NodeStatus.Unknown));
211+
Collections.singletonList(NodeStatus.Stopped));
212212

213213
// Check leader distribution
214214
isDistributionBalanced = false;

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

Lines changed: 19 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -34,8 +34,10 @@
3434
import org.apache.iotdb.confignode.rpc.thrift.TTimeSlotList;
3535
import org.apache.iotdb.consensus.ConsensusFactory;
3636
import org.apache.iotdb.it.env.EnvFactory;
37+
import org.apache.iotdb.it.env.cluster.EnvUtils;
3738
import org.apache.iotdb.it.framework.IoTDBTestRunner;
3839
import org.apache.iotdb.itbase.category.ClusterIT;
40+
import org.apache.iotdb.itbase.env.BaseNodeWrapper;
3941
import org.apache.iotdb.rpc.TSStatusCode;
4042

4143
import org.apache.thrift.TException;
@@ -49,8 +51,10 @@
4951
import org.slf4j.LoggerFactory;
5052

5153
import java.io.IOException;
54+
import java.util.ArrayList;
5255
import java.util.Collections;
5356
import java.util.HashMap;
57+
import java.util.List;
5458
import java.util.Map;
5559
import java.util.concurrent.TimeUnit;
5660
import java.util.concurrent.atomic.AtomicInteger;
@@ -104,7 +108,21 @@ public void testAutoRegionGroupExtensionPolicy2()
104108
EnvFactory.getEnv()
105109
.ensureNodeStatus(
106110
Collections.singletonList(EnvFactory.getEnv().getDataNodeWrapper(1)),
107-
Collections.singletonList(NodeStatus.Unknown));
111+
Collections.singletonList(EnvUtils.getNodeStatusAfterLocalStop()));
112+
113+
// The remaining DataNodes may transiently be ReadOnly (e.g. the disk-full flap on a busy
114+
// runner, which auto-recovers at the next disk sampling); wait for them to be Running so a
115+
// transient status does not fail the allocation below.
116+
List<BaseNodeWrapper> remainingDataNodes = new ArrayList<>();
117+
for (int i = 0; i < testDataNodeNum; i++) {
118+
if (i != 1) {
119+
remainingDataNodes.add(EnvFactory.getEnv().getDataNodeWrapper(i));
120+
}
121+
}
122+
EnvFactory.getEnv()
123+
.ensureNodeStatus(
124+
remainingDataNodes,
125+
Collections.nCopies(remainingDataNodes.size(), NodeStatus.Running));
108126

109127
// Create 3 DataPartitions to extend 3 DataRegionGroups
110128
for (int i = 0; i < testMinDataRegionGroupNum; i++) {

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

Lines changed: 29 additions & 28 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import org.apache.iotdb.commons.cluster.RegionStatus;
2929
import org.apache.iotdb.commons.pipe.config.constant.SystemConstant;
3030
import org.apache.iotdb.confignode.it.utils.ConfigNodeTestUtils;
31+
import org.apache.iotdb.confignode.rpc.thrift.TDataNodeInfo;
3132
import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionReq;
3233
import org.apache.iotdb.confignode.rpc.thrift.TDataPartitionTableResp;
3334
import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
@@ -318,36 +319,34 @@ public void testPartitionAllocation() throws Exception {
318319

319320
// Shutdown 1 DataNode
320321
// Current cluster: 1C5D
321-
// DataNode status: Running, Running, Removing, ReadOnly, Unknown
322+
// DataNode status: Running, Running, Removing, ReadOnly, Stopped
322323
// Region distribution: [0, 1, 2], [0, 1, 2], [0], [1], [2]
323324
EnvFactory.getEnv().shutdownDataNode(4);
324325
// Wait for shutdown check
325-
while (true) {
326-
AtomicBoolean containUnknown = new AtomicBoolean(false);
326+
boolean isShutdownDetected = false;
327+
for (int retry = 0; retry < 60; retry++) {
327328
TShowDataNodesResp showDataNodesResp = client.showDataNodes();
328-
showDataNodesResp
329-
.getDataNodesInfoList()
330-
.forEach(
331-
dataNodeInfo -> {
332-
if (NodeStatus.Unknown.getStatus().equals(dataNodeInfo.getStatus())) {
333-
containUnknown.set(true);
334-
}
335-
});
336-
337-
if (containUnknown.get()) {
329+
for (TDataNodeInfo dataNodeInfo : showDataNodesResp.getDataNodesInfoList()) {
330+
if (NodeStatus.Stopped.getStatus().equals(dataNodeInfo.getStatus())) {
331+
isShutdownDetected = true;
332+
break;
333+
}
334+
}
335+
if (isShutdownDetected) {
338336
break;
339337
}
340338
TimeUnit.SECONDS.sleep(1);
341339
}
340+
Assert.assertTrue(isShutdownDetected);
342341

343342
// Register 1 DataNode and Create 1 DataPartition to extend 1 DataRegionGroup
344343
// The new DataRegions wouldn't be allocated to the Removing and ReadOnly DataNode
345-
// But the new DataRegion can be allocated to the Unknown DataNode
344+
// But the new DataRegion can be allocated to the Stopped DataNode
346345
// Current cluster: 1C6D
347-
// Status: Running, Running, Removing, ReadOnly, Unknown, Running
346+
// Status: Running, Running, Removing, ReadOnly, Stopped, Running
348347
// RegionGroup: [0, 1, 2, 3], [0, 1, 2], [0], [1], [2, 3], [3]
349348
EnvFactory.getEnv().registerNewDataNode(false);
350-
// Use thread sleep to replace verifying because the Unknown DataNode can not pass the
349+
// Use thread sleep to replace verifying because the Stopped DataNode can not pass the
351350
// connection check
352351
TimeUnit.SECONDS.sleep(25);
353352
partitionSlotsMap =
@@ -446,23 +445,25 @@ public void testPartitionAllocation() throws Exception {
446445
// RegionGroup: [0, 1, 2, 3], [0, 1, 2], [0], [1], [2, 3], [3]
447446
EnvFactory.getEnv().startDataNode(4);
448447
// Wait for restart check
449-
while (true) {
450-
AtomicBoolean containUnknown = new AtomicBoolean(false);
448+
boolean isRestartDetected = false;
449+
for (int retry = 0; retry < 60; retry++) {
451450
TShowDataNodesResp showDataNodesResp = client.showDataNodes();
452-
showDataNodesResp
453-
.getDataNodesInfoList()
454-
.forEach(
455-
dataNodeInfo -> {
456-
if (NodeStatus.Unknown.getStatus().equals(dataNodeInfo.getStatus())) {
457-
containUnknown.set(true);
458-
}
459-
});
460-
461-
if (!containUnknown.get()) {
451+
boolean containDown = false;
452+
for (TDataNodeInfo dataNodeInfo : showDataNodesResp.getDataNodesInfoList()) {
453+
// The restarted DataNode keeps Stopped until its first heartbeat revives it
454+
if (NodeStatus.Unknown.getStatus().equals(dataNodeInfo.getStatus())
455+
|| NodeStatus.Stopped.getStatus().equals(dataNodeInfo.getStatus())) {
456+
containDown = true;
457+
break;
458+
}
459+
}
460+
if (!containDown) {
461+
isRestartDetected = true;
462462
break;
463463
}
464464
TimeUnit.SECONDS.sleep(1);
465465
}
466+
Assert.assertTrue(isRestartDetected);
466467
// Check Region count and status
467468
for (int i = 0; i < 30; i++) {
468469
runningCnt = 0;

0 commit comments

Comments
 (0)