You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by ta...@apache.org on 2024/10/30 03:20:52 UTC
(iotdb) branch master updated: Enhance Remove DataNode Test (#13809)
This is an automated email from the ASF dual-hosted git repository.
tanxinyu pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iotdb.git
The following commit(s) were added to refs/heads/master by this push:
new 063fdddaaa5 Enhance Remove DataNode Test (#13809)
063fdddaaa5 is described below
commit 063fdddaaa5b675ee17e9f79e633390a8381c21b
Author: Xiangpeng Hu <65...@users.noreply.github.com>
AuthorDate: Wed Oct 30 11:20:47 2024 +0800
Enhance Remove DataNode Test (#13809)
---
.../IoTDBRegionMigrateReliabilityITFramework.java | 24 ---
.../IoTDBRemoveDataNodeITFramework.java | 92 ++++++++++-
.../IoTDBRemoveDataNodeNormalIT.java | 4 +-
.../client/async/CnToDnAsyncRequestType.java | 2 +-
.../CnToDnInternalServiceAsyncRequestManager.java | 5 +-
.../rpc/DataNodeAsyncRequestRPCHandler.java | 2 +-
.../client/sync/CnToDnSyncRequestType.java | 2 +-
.../client/sync/SyncDataNodeClientPool.java | 3 +-
.../iotdb/confignode/manager/ProcedureManager.java | 15 +-
.../iotdb/confignode/manager/load/LoadManager.java | 4 +-
.../manager/load/cache/AbstractLoadCache.java | 2 +-
.../confignode/manager/load/cache/LoadCache.java | 10 +-
.../load/cache/consensus/ConsensusGroupCache.java | 2 +-
.../load/cache/node/AINodeHeartbeatCache.java | 2 +-
.../load/cache/node/ConfigNodeHeartbeatCache.java | 2 +-
.../load/cache/node/DataNodeHeartbeatCache.java | 4 +-
.../manager/load/cache/region/RegionCache.java | 2 +-
.../load/cache/region/RegionGroupCache.java | 2 +-
.../manager/load/service/StatisticsService.java | 2 +-
.../iotdb/confignode/manager/node/NodeManager.java | 2 +-
.../procedure/env/ConfigNodeProcedureEnv.java | 2 +-
.../procedure/env/RemoveDataNodeHandler.java | 35 ++---
.../impl/node/RemoveDataNodesProcedure.java | 22 +--
.../confignode/manager/ProcedureManagerTest.java | 174 +++++++++++++++++++++
.../confignode/manager/load/LoadManagerTest.java | 4 +-
.../router/priority/GreedyPriorityTest.java | 2 +-
.../priority/LeaderPriorityBalancerTest.java | 2 +-
.../load/cache/ConsensusGroupCacheTest.java | 2 +-
.../manager/load/cache/NodeCacheTest.java | 4 +-
.../impl/DataNodeInternalRPCServiceImpl.java | 3 +-
.../java/org/apache/iotdb/db/service/DataNode.java | 4 +
.../commons/file/SystemPropertiesHandler.java | 14 ++
.../src/main/thrift/datanode.thrift | 4 +-
33 files changed, 355 insertions(+), 100 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateReliabilityITFramework.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateReliabilityITFramework.java
index dc514b27aa5..87f1ab2fd58 100644
--- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateReliabilityITFramework.java
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionMigrateReliabilityITFramework.java
@@ -306,30 +306,6 @@ public class IoTDBRegionMigrateReliabilityITFramework {
}
}
- private void restartDataNodes(List<DataNodeWrapper> dataNodeWrappers) {
- dataNodeWrappers.parallelStream()
- .forEach(
- nodeWrapper -> {
- nodeWrapper.stop();
- Awaitility.await()
- .atMost(1, TimeUnit.MINUTES)
- .pollDelay(2, TimeUnit.SECONDS)
- .until(() -> !nodeWrapper.isAlive());
- LOGGER.info("Node {} stopped.", nodeWrapper.getId());
- nodeWrapper.start();
- Awaitility.await()
- .atMost(1, TimeUnit.MINUTES)
- .pollDelay(2, TimeUnit.SECONDS)
- .until(nodeWrapper::isAlive);
- try {
- TimeUnit.SECONDS.sleep(10);
- } catch (InterruptedException e) {
- Thread.currentThread().interrupt();
- }
- LOGGER.info("Node {} restarted.", nodeWrapper.getId());
- });
- }
-
private void setConfigNodeKillPoints(
KeySetView<String, Boolean> killConfigNodeKeywords, Consumer<KillPointContext> action) {
EnvFactory.getEnv()
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeITFramework.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeITFramework.java
index 4e253ab5ab1..490696005ea 100644
--- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeITFramework.java
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeITFramework.java
@@ -27,6 +27,7 @@ import org.apache.iotdb.confignode.rpc.thrift.TDataNodeRemoveResp;
import org.apache.iotdb.consensus.ConsensusFactory;
import org.apache.iotdb.db.queryengine.common.header.ColumnHeaderConstant;
import org.apache.iotdb.it.env.EnvFactory;
+import org.apache.iotdb.it.env.cluster.node.DataNodeWrapper;
import org.apache.iotdb.itbase.exception.InconsistentDataException;
import org.apache.iotdb.rpc.TSStatusCode;
@@ -88,7 +89,8 @@ public class IoTDBRemoveDataNodeITFramework {
final int configNodeNum,
final int dataNodeNum,
final int removeDataNodeNum,
- final int dataRegionPerDataNode)
+ final int dataRegionPerDataNode,
+ final boolean rejoinRemovedDataNode)
throws Exception {
testRemoveDataNode(
dataReplicateFactor,
@@ -97,7 +99,8 @@ public class IoTDBRemoveDataNodeITFramework {
dataNodeNum,
removeDataNodeNum,
dataRegionPerDataNode,
- true);
+ true,
+ rejoinRemovedDataNode);
}
public void failTest(
@@ -106,7 +109,8 @@ public class IoTDBRemoveDataNodeITFramework {
final int configNodeNum,
final int dataNodeNum,
final int removeDataNodeNum,
- final int dataRegionPerDataNode)
+ final int dataRegionPerDataNode,
+ final boolean rejoinRemovedDataNode)
throws Exception {
testRemoveDataNode(
dataReplicateFactor,
@@ -115,7 +119,8 @@ public class IoTDBRemoveDataNodeITFramework {
dataNodeNum,
removeDataNodeNum,
dataRegionPerDataNode,
- false);
+ false,
+ rejoinRemovedDataNode);
}
public void testRemoveDataNode(
@@ -125,7 +130,8 @@ public class IoTDBRemoveDataNodeITFramework {
final int dataNodeNum,
final int removeDataNodeNum,
final int dataRegionPerDataNode,
- final boolean expectRemoveSuccess)
+ final boolean expectRemoveSuccess,
+ final boolean rejoinRemovedDataNode)
throws Exception {
// Set up the environment
EnvFactory.getEnv()
@@ -147,6 +153,13 @@ public class IoTDBRemoveDataNodeITFramework {
ResultSet result = statement.executeQuery(SHOW_REGIONS);
Map<Integer, Set<Integer>> regionMap = getRegionMap(result);
+ regionMap.forEach(
+ (key, valueSet) -> {
+ LOGGER.info("Key: {}, Value: {}", key, valueSet);
+ if (valueSet.size() != dataReplicateFactor) {
+ Assert.fail();
+ }
+ });
// Get all data nodes
result = statement.executeQuery(SHOW_DATANODES);
@@ -159,6 +172,11 @@ public class IoTDBRemoveDataNodeITFramework {
final Set<Integer> removeDataNodes =
selectRemoveDataNodes(allDataNodeId, regionMap, removeDataNodeNum);
+ List<DataNodeWrapper> removeDataNodeWrappers =
+ removeDataNodes.stream()
+ .map(dataNodeId -> EnvFactory.getEnv().dataNodeIdToWrapper(dataNodeId).get())
+ .collect(Collectors.toList());
+
AtomicReference<SyncConfigNodeIServiceClient> clientRef = new AtomicReference<>(client);
List<TDataNodeLocation> removeDataNodeLocations =
clientRef
@@ -204,8 +222,46 @@ public class IoTDBRemoveDataNodeITFramework {
}
LOGGER.info("Remove DataNodes success");
- } catch (InconsistentDataException ignored) {
+ if (rejoinRemovedDataNode) {
+ try {
+ // Use sleep and restart to ensure that removeDataNodes restarts successfully
+ Thread.sleep(30000);
+ restartDataNodes(removeDataNodeWrappers);
+ LOGGER.info("RemoveDataNodes:{} rejoined successfully.", removeDataNodes);
+ } catch (Exception e) {
+ LOGGER.error("RemoveDataNodes rejoin failed.");
+ Assert.fail();
+ }
+ }
+ } catch (InconsistentDataException e) {
+ LOGGER.error("Unexpected error:", e);
+ }
+
+ try (final Connection connection = closeQuietly(EnvFactory.getEnv().getConnection());
+ final Statement statement = closeQuietly(connection.createStatement())) {
+
+ // Check the data region distribution after removing data nodes
+ ResultSet result = statement.executeQuery(SHOW_REGIONS);
+ Map<Integer, Set<Integer>> afterRegionMap = getRegionMap(result);
+ afterRegionMap.forEach(
+ (key, valueSet) -> {
+ LOGGER.info("Key: {}, Value: {}", key, valueSet);
+ if (valueSet.size() != dataReplicateFactor) {
+ Assert.fail();
+ }
+ });
+
+ if (rejoinRemovedDataNode) {
+ result = statement.executeQuery(SHOW_DATANODES);
+ Set<Integer> allDataNodeId = new HashSet<>();
+ while (result.next()) {
+ allDataNodeId.add(result.getInt(ColumnHeaderConstant.NODE_ID));
+ }
+ Assert.assertEquals(allDataNodeId.size(), dataNodeNum);
+ }
+ } catch (InconsistentDataException e) {
+ LOGGER.error("Unexpected error:", e);
}
}
@@ -285,4 +341,28 @@ public class IoTDBRemoveDataNodeITFramework {
LOGGER.info("DataNodes has been successfully changed to {}", lastTimeDataNodeLocations.get());
}
+
+ public void restartDataNodes(List<DataNodeWrapper> dataNodeWrappers) {
+ dataNodeWrappers.parallelStream()
+ .forEach(
+ nodeWrapper -> {
+ nodeWrapper.stopForcibly();
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .pollDelay(2, TimeUnit.SECONDS)
+ .until(() -> !nodeWrapper.isAlive());
+ LOGGER.info("Node {} stopped.", nodeWrapper.getId());
+ nodeWrapper.start();
+ Awaitility.await()
+ .atMost(1, TimeUnit.MINUTES)
+ .pollDelay(2, TimeUnit.SECONDS)
+ .until(nodeWrapper::isAlive);
+ try {
+ TimeUnit.SECONDS.sleep(10);
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ }
+ LOGGER.info("Node {} restarted.", nodeWrapper.getId());
+ });
+ }
}
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeNormalIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeNormalIT.java
index 4496c6c725e..f091499bf02 100644
--- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeNormalIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeNormalIT.java
@@ -31,11 +31,11 @@ import org.junit.runner.RunWith;
public class IoTDBRemoveDataNodeNormalIT extends IoTDBRemoveDataNodeITFramework {
@Test
public void success1C4DTest() throws Exception {
- successTest(2, 3, 1, 4, 1, 2);
+ successTest(2, 3, 1, 4, 1, 2, true);
}
@Test
public void fail1C3DTest() throws Exception {
- failTest(2, 3, 1, 3, 1, 2);
+ failTest(2, 3, 1, 3, 1, 2, false);
}
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java
index 8404db8101c..315a9a463b8 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnAsyncRequestType.java
@@ -21,7 +21,7 @@ package org.apache.iotdb.confignode.client.async;
public enum CnToDnAsyncRequestType {
// Node Maintenance
- STOP_DATA_NODE,
+ STOP_AND_CLEAR_DATA_NODE,
CLEAN_DATA_NODE_CACHE,
FLUSH,
MERGE,
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java
index 70c168574b1..8562f75f85a 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/CnToDnInternalServiceAsyncRequestManager.java
@@ -398,8 +398,9 @@ public class CnToDnInternalServiceAsyncRequestManager
client.cleanDataNodeCache(
(TCleanDataNodeCacheReq) req, (DataNodeTSStatusRPCHandler) handler));
actionMapBuilder.put(
- CnToDnAsyncRequestType.STOP_DATA_NODE,
- (req, client, handler) -> client.stopDataNode((DataNodeTSStatusRPCHandler) handler));
+ CnToDnAsyncRequestType.STOP_AND_CLEAR_DATA_NODE,
+ (req, client, handler) ->
+ client.stopAndClearDataNode((DataNodeTSStatusRPCHandler) handler));
}
@Override
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeAsyncRequestRPCHandler.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeAsyncRequestRPCHandler.java
index b850bbd54ed..b1e8753ab49 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeAsyncRequestRPCHandler.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeAsyncRequestRPCHandler.java
@@ -197,7 +197,7 @@ public abstract class DataNodeAsyncRequestRPCHandler<Response>
case CLEAR_CACHE:
case INVALIDATE_LAST_CACHE:
case CLEAN_DATA_NODE_CACHE:
- case STOP_DATA_NODE:
+ case STOP_AND_CLEAR_DATA_NODE:
case START_REPAIR_DATA:
case STOP_REPAIR_DATA:
case LOAD_CONFIGURATION:
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/CnToDnSyncRequestType.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/CnToDnSyncRequestType.java
index 1ec6b6ef4e9..21263612af2 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/CnToDnSyncRequestType.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/CnToDnSyncRequestType.java
@@ -22,7 +22,7 @@ package org.apache.iotdb.confignode.client.sync;
public enum CnToDnSyncRequestType {
// Node Maintenance
CLEAN_DATA_NODE_CACHE,
- STOP_DATA_NODE,
+ STOP_AND_CLEAR_DATA_NODE,
SET_SYSTEM_STATUS,
SHOW_CONFIGURATION,
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java
index 42fbbe6b83e..2999bbdec58 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/sync/SyncDataNodeClientPool.java
@@ -104,7 +104,8 @@ public class SyncDataNodeClientPool {
CnToDnSyncRequestType.CLEAN_DATA_NODE_CACHE,
(req, client) -> client.cleanDataNodeCache((TCleanDataNodeCacheReq) req));
actionMapBuilder.put(
- CnToDnSyncRequestType.STOP_DATA_NODE, (req, client) -> client.stopDataNode());
+ CnToDnSyncRequestType.STOP_AND_CLEAR_DATA_NODE,
+ (req, client) -> client.stopAndClearDataNode());
actionMapBuilder.put(
CnToDnSyncRequestType.SET_SYSTEM_STATUS,
(req, client) -> client.setSystemStatus((String) req));
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 bbbe7214e18..df5a905cf57 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
@@ -616,7 +616,7 @@ public class ProcedureManager {
public TSStatus checkRemoveDataNodes(List<TDataNodeLocation> dataNodeLocations) {
// 1. Only one RemoveDataNodesProcedure is allowed in the cluster
Optional<Procedure<ConfigNodeProcedureEnv>> anotherRemoveProcedure =
- this.executor.getProcedures().values().stream()
+ getExecutor().getProcedures().values().stream()
.filter(
procedure -> {
if (procedure instanceof RemoveDataNodesProcedure) {
@@ -640,11 +640,10 @@ public class ProcedureManager {
}
// 2. Check if the RemoveDataNodesProcedure conflicts with the RegionMigrateProcedure
- RemoveDataNodeHandler removeDataNodeHandler = env.getRemoveDataNodeManager();
Set<TConsensusGroupId> removedDataNodesRegionSet =
- removeDataNodeHandler.getRemovedDataNodesRegionSet(dataNodeLocations);
+ getEnv().getRemoveDataNodeHandler().getRemovedDataNodesRegionSet(dataNodeLocations);
Optional<Procedure<ConfigNodeProcedureEnv>> conflictRegionMigrateProcedure =
- this.executor.getProcedures().values().stream()
+ getExecutor().getProcedures().values().stream()
.filter(
procedure -> {
if (procedure instanceof RegionMigrateProcedure) {
@@ -675,7 +674,7 @@ public class ProcedureManager {
// 3. Check if the RegionMigrateProcedure generated by RemoveDataNodesProcedure conflicts with
// each other
List<RegionMigrationPlan> regionMigrationPlans =
- removeDataNodeHandler.getRegionMigrationPlans(dataNodeLocations);
+ getEnv().getRemoveDataNodeHandler().getRegionMigrationPlans(dataNodeLocations);
removedDataNodesRegionSet.clear();
for (RegionMigrationPlan regionMigrationPlan : regionMigrationPlans) {
if (removedDataNodesRegionSet.contains(regionMigrationPlan.getRegionId())) {
@@ -718,7 +717,7 @@ public class ProcedureManager {
String failMessage = null;
// 1. Check if the RegionMigrateProcedure has conflict with another RegionMigrateProcedure
Optional<Procedure<ConfigNodeProcedureEnv>> anotherMigrateProcedure =
- this.executor.getProcedures().values().stream()
+ getExecutor().getProcedures().values().stream()
.filter(
procedure -> {
if (procedure instanceof RegionMigrateProcedure) {
@@ -799,7 +798,7 @@ public class ProcedureManager {
// 2. Check if the RegionMigrateProcedure has conflict with RemoveDataNodesProcedure
Optional<Procedure<ConfigNodeProcedureEnv>> conflictRemoveDataNodesProcedure =
- this.executor.getProcedures().values().stream()
+ getExecutor().getProcedures().values().stream()
.filter(
procedure -> {
if (procedure instanceof RemoveDataNodesProcedure) {
@@ -810,7 +809,7 @@ public class ProcedureManager {
.findAny();
if (conflictRemoveDataNodesProcedure.isPresent()) {
- RemoveDataNodeHandler removeDataNodeHandler = env.getRemoveDataNodeManager();
+ RemoveDataNodeHandler removeDataNodeHandler = env.getRemoveDataNodeHandler();
List<TDataNodeLocation> removedDataNodes =
((RemoveDataNodesProcedure) conflictRemoveDataNodesProcedure.get()).getRemovedDataNodes();
Set<TConsensusGroupId> removedDataNodesRegionSet =
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
index 31ef0c17c3a..ee9e7530f3d 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/LoadManager.java
@@ -270,7 +270,7 @@ public class LoadManager {
default:
break;
}
- loadCache.updateNodeStatistics();
+ loadCache.updateNodeStatistics(true);
eventService.checkAndBroadcastNodeStatisticsChangeEventIfNecessary();
}
@@ -282,7 +282,7 @@ public class LoadManager {
*/
public void removeNodeCache(int nodeId) {
loadCache.removeNodeCache(nodeId);
- loadCache.updateNodeStatistics();
+ loadCache.updateNodeStatistics(true);
eventService.checkAndBroadcastNodeStatisticsChangeEventIfNecessary();
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/AbstractLoadCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/AbstractLoadCache.java
index 42411e462a1..5fe60d03309 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/AbstractLoadCache.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/AbstractLoadCache.java
@@ -78,7 +78,7 @@ public abstract class AbstractLoadCache {
/**
* Update currentStatistics based on the latest heartbeat sample that cached in the slidingWindow.
*/
- public abstract void updateCurrentStatistics();
+ public abstract void updateCurrentStatistics(boolean forceUpdate);
public AbstractStatistics getCurrentStatistics() {
return currentStatistics.get();
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java
index 137a8b7f63e..01aefa2e2ee 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/LoadCache.java
@@ -331,8 +331,10 @@ public class LoadCache {
}
/** Update the NodeStatistics of all Nodes. */
- public void updateNodeStatistics() {
- nodeCacheMap.values().forEach(BaseNodeCache::updateCurrentStatistics);
+ public void updateNodeStatistics(boolean forceUpdate) {
+ nodeCacheMap
+ .values()
+ .forEach(baseNodeCache -> baseNodeCache.updateCurrentStatistics(forceUpdate));
}
/** Update the RegionGroupStatistics of all RegionGroups. */
@@ -342,7 +344,9 @@ public class LoadCache {
/** Update the ConsensusGroupStatistics of all RegionGroups. */
public void updateConsensusGroupStatistics() {
- consensusGroupCacheMap.values().forEach(ConsensusGroupCache::updateCurrentStatistics);
+ consensusGroupCacheMap
+ .values()
+ .forEach(consensusGroupCache -> consensusGroupCache.updateCurrentStatistics(false));
}
/**
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/consensus/ConsensusGroupCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/consensus/ConsensusGroupCache.java
index 32d798c48b3..aa924dc29b5 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/consensus/ConsensusGroupCache.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/consensus/ConsensusGroupCache.java
@@ -36,7 +36,7 @@ public class ConsensusGroupCache extends AbstractLoadCache {
}
@Override
- public synchronized void updateCurrentStatistics() {
+ public synchronized void updateCurrentStatistics(boolean forceUpdate) {
ConsensusGroupHeartbeatSample lastSample;
synchronized (slidingWindow) {
lastSample = (ConsensusGroupHeartbeatSample) getLastSample();
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/AINodeHeartbeatCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/AINodeHeartbeatCache.java
index f6c845f30e6..50ad19c530e 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/AINodeHeartbeatCache.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/node/AINodeHeartbeatCache.java
@@ -34,7 +34,7 @@ public class AINodeHeartbeatCache extends BaseNodeCache {
}
@Override
- public void updateCurrentStatistics() {
+ public void updateCurrentStatistics(boolean forceUpdate) {
NodeHeartbeatSample lastSample = null;
synchronized (slidingWindow) {
if (!slidingWindow.isEmpty()) {
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 72daed53512..d6a41285c54 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
@@ -44,7 +44,7 @@ public class ConfigNodeHeartbeatCache extends BaseNodeCache {
}
@Override
- public synchronized void updateCurrentStatistics() {
+ public synchronized void updateCurrentStatistics(boolean forceUpdate) {
// Skip itself and the Removing status can not be updated
if (nodeId == CURRENT_NODE_ID || NodeStatus.Removing.equals(getNodeStatus())) {
return;
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 39454a4ccb4..cf550931c40 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
@@ -37,9 +37,9 @@ public class DataNodeHeartbeatCache extends BaseNodeCache {
}
@Override
- public synchronized void updateCurrentStatistics() {
+ public synchronized void updateCurrentStatistics(boolean forceUpdate) {
// The Removing status can not be updated
- if (NodeStatus.Removing.equals(getNodeStatus())) {
+ if (!forceUpdate && NodeStatus.Removing.equals(getNodeStatus())) {
return;
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/region/RegionCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/region/RegionCache.java
index 8facf40ba5a..ecb4aebd10b 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/region/RegionCache.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/region/RegionCache.java
@@ -34,7 +34,7 @@ public class RegionCache extends AbstractLoadCache {
}
@Override
- public synchronized void updateCurrentStatistics() {
+ public synchronized void updateCurrentStatistics(boolean forceUpdate) {
RegionHeartbeatSample lastSample;
synchronized (slidingWindow) {
lastSample = (RegionHeartbeatSample) getLastSample();
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/region/RegionGroupCache.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/region/RegionGroupCache.java
index d166d4bceae..0fa8d95d2d7 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/region/RegionGroupCache.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/cache/region/RegionGroupCache.java
@@ -94,7 +94,7 @@ public class RegionGroupCache {
* slidingWindow.
*/
public void updateCurrentStatistics() {
- regionCacheMap.values().forEach(RegionCache::updateCurrentStatistics);
+ regionCacheMap.values().forEach(regionCache -> regionCache.updateCurrentStatistics(false));
Map<Integer, RegionStatistics> regionStatisticsMap =
regionCacheMap.entrySet().stream()
.collect(
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java
index d8accbcd482..ed80cf58f57 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/load/service/StatisticsService.java
@@ -82,7 +82,7 @@ public class StatisticsService {
}
private void updateLoadStatistics() {
- loadCache.updateNodeStatistics();
+ loadCache.updateNodeStatistics(false);
loadCache.updateRegionGroupStatistics();
loadCache.updateConsensusGroupStatistics();
}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
index 433a0748b8a..d0053594ba1 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/node/NodeManager.java
@@ -383,7 +383,7 @@ public class NodeManager {
try {
// Checks if the RemoveDataNode request is valid
RemoveDataNodeHandler removeDataNodeHandler =
- configManager.getProcedureManager().getEnv().getRemoveDataNodeManager();
+ configManager.getProcedureManager().getEnv().getRemoveDataNodeHandler();
DataNodeToStatusResp preCheckStatus =
removeDataNodeHandler.checkRemoveDataNodeRequest(removeDataNodePlan);
if (preCheckStatus.getStatus().getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
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 2a16798d8a5..d9ec20316a7 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
@@ -841,7 +841,7 @@ public class ConfigNodeProcedureEnv {
return regionMaintainHandler;
}
- public RemoveDataNodeHandler getRemoveDataNodeManager() {
+ public RemoveDataNodeHandler getRemoveDataNodeHandler() {
return removeDataNodeHandler;
}
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 b80b1c560e6..44371956981 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
@@ -134,29 +134,26 @@ public class RemoveDataNodeHandler {
int dataNodeId = entry.getKey();
NodeStatus nodeStatus = nodeStatusMap.get(dataNodeId);
-
- // Force updating NodeStatus to NodeStatus.Removing
+ RegionStatus regionStatus = RegionStatus.valueOf(nodeStatus.getStatus());
+ // Force updating NodeStatus
configManager
.getLoadManager()
.forceUpdateNodeCache(
NodeType.DataNode, dataNodeId, new NodeHeartbeatSample(currentTime, nodeStatus));
- if (nodeStatus == NodeStatus.Removing) {
- Map<TConsensusGroupId, Map<Integer, RegionHeartbeatSample>> removingHeartbeatSampleMap =
- new TreeMap<>();
- // Force update RegionStatus to NodeStatus.Removing
- configManager
- .getPartitionManager()
- .getAllReplicaSets(entry.getKey())
- .forEach(
- replicaSet ->
- removingHeartbeatSampleMap.put(
- replicaSet.getRegionId(),
- Collections.singletonMap(
- dataNodeId,
- new RegionHeartbeatSample(currentTime, RegionStatus.Removing))));
- configManager.getLoadManager().forceUpdateRegionGroupCache(removingHeartbeatSampleMap);
- }
+ Map<TConsensusGroupId, Map<Integer, RegionHeartbeatSample>> heartbeatSampleMap =
+ new TreeMap<>();
+ // Force update RegionStatus
+ configManager
+ .getPartitionManager()
+ .getAllReplicaSets(entry.getKey())
+ .forEach(
+ replicaSet ->
+ heartbeatSampleMap.put(
+ replicaSet.getRegionId(),
+ Collections.singletonMap(
+ dataNodeId, new RegionHeartbeatSample(currentTime, regionStatus))));
+ configManager.getLoadManager().forceUpdateRegionGroupCache(heartbeatSampleMap);
}
}
@@ -268,7 +265,7 @@ public class RemoveDataNodeHandler {
removedDataNodes);
DataNodeAsyncRequestContext<TDataNodeLocation, TSStatus> stopDataNodesContext =
- new DataNodeAsyncRequestContext<>(CnToDnAsyncRequestType.STOP_DATA_NODE);
+ new DataNodeAsyncRequestContext<>(CnToDnAsyncRequestType.STOP_AND_CLEAR_DATA_NODE);
for (TDataNodeLocation dataNode : removedDataNodes) {
stopDataNodesContext.putRequest(dataNode.getDataNodeId(), dataNode);
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodesProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodesProcedure.java
index 57c44310ce8..aa27a0276f7 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodesProcedure.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodesProcedure.java
@@ -76,7 +76,7 @@ public class RemoveDataNodesProcedure extends AbstractNodeProcedure<RemoveDataNo
return Flow.NO_MORE_STATE;
}
- RemoveDataNodeHandler removeDataNodeHandler = env.getRemoveDataNodeManager();
+ RemoveDataNodeHandler removeDataNodeHandler = env.getRemoveDataNodeHandler();
try {
switch (state) {
case REGION_REPLICA_CHECK:
@@ -169,6 +169,7 @@ public class RemoveDataNodesProcedure extends AbstractNodeProcedure<RemoveDataNo
List<TRegionReplicaSet> replicaSets =
env.getConfigManager().getPartitionManager().getAllReplicaSets();
List<TDataNodeLocation> rollBackDataNodes = new ArrayList<>();
+ List<TDataNodeLocation> successDataNodes = new ArrayList<>();
for (TDataNodeLocation dataNode : removedDataNodes) {
List<TConsensusGroupId> migratedFailedRegions =
replicaSets.stream()
@@ -184,21 +185,24 @@ public class RemoveDataNodesProcedure extends AbstractNodeProcedure<RemoveDataNo
migratedFailedRegions);
rollBackDataNodes.add(dataNode);
} else {
- LOG.info(
- "{}, Region all migrated successfully, start to stop DataNode: {}",
- REMOVE_DATANODE_PROCESS,
- dataNode);
- env.getRemoveDataNodeManager().removeDataNodePersistence(removedDataNodes);
- env.getRemoveDataNodeManager().stopDataNodes(removedDataNodes);
+ successDataNodes.add(dataNode);
}
}
+ if (!successDataNodes.isEmpty()) {
+ LOG.info(
+ "{}, DataNodes: {} all regions migrated successfully, start to stop them.",
+ REMOVE_DATANODE_PROCESS,
+ successDataNodes);
+ env.getRemoveDataNodeHandler().removeDataNodePersistence(successDataNodes);
+ env.getRemoveDataNodeHandler().stopDataNodes(successDataNodes);
+ }
if (!rollBackDataNodes.isEmpty()) {
LOG.info(
"{}, Start to roll back the DataNodes status: {}",
REMOVE_DATANODE_PROCESS,
rollBackDataNodes);
- env.getRemoveDataNodeManager().changeDataNodeStatus(rollBackDataNodes, nodeStatusMap);
- env.getRemoveDataNodeManager().broadcastDataNodeStatusChange(rollBackDataNodes);
+ env.getRemoveDataNodeHandler().changeDataNodeStatus(rollBackDataNodes, nodeStatusMap);
+ env.getRemoveDataNodeHandler().broadcastDataNodeStatusChange(rollBackDataNodes);
LOG.info(
"{}, Roll back the DataNodes status successfully: {}",
REMOVE_DATANODE_PROCESS,
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
new file mode 100644
index 00000000000..e57cbd0ddb5
--- /dev/null
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/ProcedureManagerTest.java
@@ -0,0 +1,174 @@
+/*
+ * 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.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.procedure.Procedure;
+import org.apache.iotdb.confignode.procedure.ProcedureExecutor;
+import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import org.apache.iotdb.confignode.procedure.env.RemoveDataNodeHandler;
+import org.apache.iotdb.confignode.procedure.impl.node.RemoveDataNodesProcedure;
+import org.apache.iotdb.confignode.procedure.impl.region.RegionMigrateProcedure;
+import org.apache.iotdb.confignode.procedure.impl.region.RegionMigrationPlan;
+
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static org.apache.iotdb.db.service.RegionMigrateService.isFailed;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
+
+public class ProcedureManagerTest {
+
+ private static ProcedureManager PROCEDURE_MANAGER;
+
+ private static RemoveDataNodeHandler REMOVE_DATA_NODE_HANDLER;
+
+ private static final ConcurrentHashMap<Long, Procedure<ConfigNodeProcedureEnv>> procedureMap =
+ new ConcurrentHashMap<>();
+
+ private final TConsensusGroupId consensusGroupId =
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 1);
+
+ private final TDataNodeLocation removeDataNodeLocationA =
+ new TDataNodeLocation(
+ 10,
+ new TEndPoint("127.0.0.1", 6667),
+ new TEndPoint("127.0.0.1", 6668),
+ new TEndPoint("127.0.0.1", 6669),
+ new TEndPoint("127.0.0.1", 6670),
+ new TEndPoint("127.0.0.1", 6671));
+
+ private final TDataNodeLocation removeDataNodeLocationB =
+ new TDataNodeLocation(
+ 11,
+ new TEndPoint("127.0.0.1", 6677),
+ new TEndPoint("127.0.0.1", 6678),
+ new TEndPoint("127.0.0.1", 6679),
+ new TEndPoint("127.0.0.1", 6680),
+ new TEndPoint("127.0.0.1", 6681));
+
+ private final TDataNodeLocation toDataNodeLocation =
+ new TDataNodeLocation(
+ 12,
+ new TEndPoint("127.0.0.1", 6687),
+ new TEndPoint("127.0.0.1", 6688),
+ new TEndPoint("127.0.0.1", 6689),
+ new TEndPoint("127.0.0.1", 6690),
+ new TEndPoint("127.0.0.1", 6691));
+
+ private final TDataNodeLocation coordinatorDataNodeLocation =
+ new TDataNodeLocation(
+ 13,
+ new TEndPoint("127.0.0.1", 6697),
+ new TEndPoint("127.0.0.1", 6698),
+ new TEndPoint("127.0.0.1", 6699),
+ new TEndPoint("127.0.0.1", 6700),
+ new TEndPoint("127.0.0.1", 6701));
+
+ private final List<TDataNodeLocation> removedDataNodes =
+ new ArrayList<>(Arrays.asList(removeDataNodeLocationA, removeDataNodeLocationB));
+
+ @BeforeClass
+ public static void setUp() throws IOException {
+ IManager CONFIG_MANAGER = new ConfigManager();
+ ProcedureManager procedureManager = CONFIG_MANAGER.getProcedureManager();
+ PROCEDURE_MANAGER = spy(procedureManager);
+
+ ProcedureExecutor<ConfigNodeProcedureEnv> procedureExecutor = PROCEDURE_MANAGER.getExecutor();
+ ProcedureExecutor<ConfigNodeProcedureEnv> PROCEDURE_EXECUTOR = spy(procedureExecutor);
+
+ ConfigNodeProcedureEnv env = PROCEDURE_MANAGER.getEnv();
+ ConfigNodeProcedureEnv ENV = spy(env);
+
+ RemoveDataNodeHandler removeDataNodeHandler = ENV.getRemoveDataNodeHandler();
+ REMOVE_DATA_NODE_HANDLER = spy(removeDataNodeHandler);
+
+ when(PROCEDURE_MANAGER.getExecutor()).thenReturn(PROCEDURE_EXECUTOR);
+ when(PROCEDURE_EXECUTOR.getProcedures()).thenReturn(procedureMap);
+ when(PROCEDURE_MANAGER.getEnv()).thenReturn(ENV);
+ when(ENV.getRemoveDataNodeHandler()).thenReturn(REMOVE_DATA_NODE_HANDLER);
+ }
+
+ @Test
+ public void testCheckRemoveDataNodeWithAnotherRemoveProcedure() {
+ Map<Integer, NodeStatus> nodeStatusMap = new HashMap<>();
+ nodeStatusMap.put(10, NodeStatus.Running);
+ RemoveDataNodesProcedure anotherRemoveProcedure =
+ new RemoveDataNodesProcedure(removedDataNodes, nodeStatusMap);
+ procedureMap.put(0L, anotherRemoveProcedure);
+
+ TSStatus status = PROCEDURE_MANAGER.checkRemoveDataNodes(removedDataNodes);
+ Assert.assertTrue(isFailed(status));
+ }
+
+ @Test
+ public void testCheckRemoveDataNodeWithConflictRegionMigrateProcedure() {
+ RegionMigrateProcedure regionMigrateProcedure =
+ new RegionMigrateProcedure(
+ consensusGroupId,
+ removeDataNodeLocationA,
+ removeDataNodeLocationB,
+ coordinatorDataNodeLocation,
+ coordinatorDataNodeLocation);
+ procedureMap.put(0L, regionMigrateProcedure);
+
+ Set<TConsensusGroupId> set = new HashSet<>();
+ set.add(consensusGroupId);
+ when(REMOVE_DATA_NODE_HANDLER.getRemovedDataNodesRegionSet(removedDataNodes)).thenReturn(set);
+
+ TSStatus status = PROCEDURE_MANAGER.checkRemoveDataNodes(removedDataNodes);
+ Assert.assertTrue(isFailed(status));
+ }
+
+ @Test
+ public void testCheckRemoveDataNodeWithRegionMigrateProcedureConflictsWithEachOther() {
+ RegionMigrationPlan regionMigrationPlanA =
+ new RegionMigrationPlan(consensusGroupId, removeDataNodeLocationA);
+ regionMigrationPlanA.setToDataNode(toDataNodeLocation);
+ RegionMigrationPlan regionMigrationPlanB =
+ new RegionMigrationPlan(consensusGroupId, removeDataNodeLocationB);
+ regionMigrationPlanB.setToDataNode(toDataNodeLocation);
+
+ List<RegionMigrationPlan> regionMigrationPlans =
+ new ArrayList<>(Arrays.asList(regionMigrationPlanA, regionMigrationPlanB));
+ when(REMOVE_DATA_NODE_HANDLER.getRegionMigrationPlans(removedDataNodes))
+ .thenReturn(regionMigrationPlans);
+
+ TSStatus status = PROCEDURE_MANAGER.checkRemoveDataNodes(removedDataNodes);
+ Assert.assertTrue(isFailed(status));
+ }
+}
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/LoadManagerTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/LoadManagerTest.java
index 12b3dbb8233..62e6a24fed8 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/LoadManagerTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/LoadManagerTest.java
@@ -89,7 +89,7 @@ public class LoadManagerTest {
// Simulate update to Running status
LOAD_CACHE.cacheConfigNodeHeartbeatSample(0, new NodeHeartbeatSample(NodeStatus.Running));
LOAD_CACHE.cacheDataNodeHeartbeatSample(1, new NodeHeartbeatSample(NodeStatus.Running));
- LOAD_CACHE.updateNodeStatistics();
+ LOAD_CACHE.updateNodeStatistics(false);
LOAD_MANAGER.getEventService().checkAndBroadcastNodeStatisticsChangeEventIfNecessary();
NODE_SEMAPHORE.acquire();
Assert.assertEquals(NodeStatus.Running, LOAD_CACHE.getNodeStatus(0));
@@ -117,7 +117,7 @@ public class LoadManagerTest {
// Removing status can't be updated to any other status automatically
LOAD_CACHE.cacheDataNodeHeartbeatSample(1, new NodeHeartbeatSample(NodeStatus.ReadOnly));
- LOAD_CACHE.updateNodeStatistics();
+ LOAD_CACHE.updateNodeStatistics(false);
LOAD_MANAGER.getEventService().checkAndBroadcastNodeStatisticsChangeEventIfNecessary();
Assert.assertEquals(NodeStatus.Removing, LOAD_CACHE.getNodeStatus(1));
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/GreedyPriorityTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/GreedyPriorityTest.java
index cd578aba1f9..a18bca54f99 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/GreedyPriorityTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/GreedyPriorityTest.java
@@ -66,7 +66,7 @@ public class GreedyPriorityTest {
nodeCacheMap.put(i, new DataNodeHeartbeatCache(i));
nodeCacheMap.get(i).cacheHeartbeatSample(new NodeHeartbeatSample(currentTimeNs, statuses[i]));
}
- nodeCacheMap.values().forEach(BaseNodeCache::updateCurrentStatistics);
+ nodeCacheMap.values().forEach(baseNodeCache -> baseNodeCache.updateCurrentStatistics(false));
/* Get the loadScoreMap */
Map<Integer, Long> loadScoreMap = new ConcurrentHashMap<>();
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/LeaderPriorityBalancerTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/LeaderPriorityBalancerTest.java
index 6f77d459603..7b126ed6e84 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/LeaderPriorityBalancerTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/balancer/router/priority/LeaderPriorityBalancerTest.java
@@ -67,7 +67,7 @@ public class LeaderPriorityBalancerTest {
.cacheHeartbeatSample(new NodeHeartbeatSample(currentTimeNs, NodeStatus.Running));
}
}
- nodeCacheMap.values().forEach(BaseNodeCache::updateCurrentStatistics);
+ nodeCacheMap.values().forEach(baseNodeCache -> baseNodeCache.updateCurrentStatistics(false));
// Get the loadScoreMap
Map<Integer, Long> loadScoreMap = new ConcurrentHashMap<>();
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/ConsensusGroupCacheTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/ConsensusGroupCacheTest.java
index 7bbd366be57..caa3e839439 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/ConsensusGroupCacheTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/manager/load/cache/ConsensusGroupCacheTest.java
@@ -31,7 +31,7 @@ public class ConsensusGroupCacheTest {
ConsensusGroupCache consensusGroupCache = new ConsensusGroupCache();
ConsensusGroupHeartbeatSample sample = new ConsensusGroupHeartbeatSample(1L, 1);
consensusGroupCache.cacheHeartbeatSample(sample);
- consensusGroupCache.updateCurrentStatistics();
+ consensusGroupCache.updateCurrentStatistics(false);
Assert.assertEquals(1, consensusGroupCache.getLeaderId());
}
}
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 eef3e9b9b6d..a4006929566 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
@@ -35,7 +35,7 @@ public class NodeCacheTest {
long currentTime = System.nanoTime();
dataNodeHeartbeatCache.cacheHeartbeatSample(
new NodeHeartbeatSample(currentTime, NodeStatus.Running));
- dataNodeHeartbeatCache.updateCurrentStatistics();
+ dataNodeHeartbeatCache.updateCurrentStatistics(false);
Assert.assertEquals(NodeStatus.Running, dataNodeHeartbeatCache.getNodeStatus());
Assert.assertEquals(0, dataNodeHeartbeatCache.getLoadScore());
@@ -44,7 +44,7 @@ public class NodeCacheTest {
currentTime = System.nanoTime();
configNodeHeartbeatCache.cacheHeartbeatSample(
new NodeHeartbeatSample(currentTime, NodeStatus.Running));
- configNodeHeartbeatCache.updateCurrentStatistics();
+ configNodeHeartbeatCache.updateCurrentStatistics(false);
Assert.assertEquals(NodeStatus.Running, configNodeHeartbeatCache.getNodeStatus());
Assert.assertEquals(0, configNodeHeartbeatCache.getLoadScore());
}
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
index 732274ca811..76583044664 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/protocol/thrift/impl/DataNodeInternalRPCServiceImpl.java
@@ -2512,7 +2512,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
@SuppressWarnings("squid:S2142") // ignore Either re-interrupt this method or rethrow
@Override
- public TSStatus stopDataNode() {
+ public TSStatus stopAndClearDataNode() {
TSStatus status = new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
LOGGER.info("Execute stopDataNode RPC method");
@@ -2534,6 +2534,7 @@ public class DataNodeInternalRPCServiceImpl implements IDataNodeRPCService.Iface
try {
DataNode.getInstance().stop();
status.setMessage("stop datanode succeed");
+ DataNode.getInstance().deleteDataNodeSystemProperties();
} catch (Exception e) {
LOGGER.warn("Stop Data Node error", e);
status.setCode(TSStatusCode.DATANODE_STOP_ERROR.getStatusCode());
diff --git a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
index 9d3754764f8..334bfcdc5cf 100644
--- a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
+++ b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/service/DataNode.java
@@ -1193,6 +1193,10 @@ public class DataNode extends ServerCommandLine implements DataNodeMBean {
}
}
+ public void deleteDataNodeSystemProperties() {
+ DataNodeSystemPropertiesHandler.getInstance().delete();
+ }
+
public void stop() {
stopTriggerRelatedServices();
registerManager.deregisterAll();
diff --git a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/file/SystemPropertiesHandler.java b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/file/SystemPropertiesHandler.java
index 35aacd0b5c2..076353275e6 100644
--- a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/file/SystemPropertiesHandler.java
+++ b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/file/SystemPropertiesHandler.java
@@ -197,4 +197,18 @@ public abstract class SystemPropertiesHandler {
this.formalFile = SystemFileFactory.INSTANCE.getFile(filePath);
this.tmpFile = SystemFileFactory.INSTANCE.getFile(filePath + ".tmp");
}
+
+ public void delete() {
+ try {
+ FileUtils.deleteFile(this.formalFile);
+ } catch (IOException e) {
+ LOGGER.warn("Delete formalFile error, ", e);
+ }
+
+ try {
+ FileUtils.deleteFile(this.tmpFile);
+ } catch (IOException e) {
+ LOGGER.warn("Delete tmpFile error, ", e);
+ }
+ }
}
diff --git a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
index b79cce400b0..32168aff86d 100644
--- a/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
+++ b/iotdb-protocol/thrift-datanode/src/main/thrift/datanode.thrift
@@ -802,9 +802,9 @@ service IDataNodeRPCService {
common.TSStatus cleanDataNodeCache(TCleanDataNodeCacheReq req)
/**
- * Config node will stop the Data node.
+ * Config node will stop and clear the Data node.
*/
- common.TSStatus stopDataNode()
+ common.TSStatus stopAndClearDataNode()
/**
* ConfigNode will ask DataNode for heartbeat in every few seconds.