You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iotdb.apache.org by yo...@apache.org on 2026/10/19 13:28:58 UTC
(iotdb) branch master updated: Fix stale IoTConsensus peers after restart during removal (#18892)
This is an automated email from the ASF dual-hosted git repository.
CRZbulabula 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 9bb1f4d14e2 Fix stale IoTConsensus peers after restart during removal (#18892)
9bb1f4d14e2 is described below
commit 9bb1f4d14e2eb1e9bde4b5a3784e7668b054ee21
Author: d-wang-commit <d....@gmail.com>
AuthorDate: Mon Oct 19 21:28:52 2026 +0800
Fix stale IoTConsensus peers after restart during removal (#18892)
* Fix stale IoTConsensus peers after restart during removal
Commit removal metadata before reconciling IoTConsensus V1 peer lists.
Retry metadata updates and V1 resets up to three times before failing.
Cover restart recovery and filter V1-only crash points in Ratis/V2 ITs.
* Clean up DataNode removal after child procedure failure
Finalize drained nodes or restore previous statuses during rollback.
Retry failed cleanup without losing the original reconciliation failure.
Cover retry exhaustion and delayed cleanup with ProcedureExecutor tests.
---
...DBRegionMigrateConfigNodeCrashIoTV2BatchIT.java | 2 +
...BRegionMigrateConfigNodeCrashIoTV2StreamIT.java | 2 +
...oTDBRegionMigrateConfigNodeCrashForRatisIT.java | 2 +
.../iotdb/confignode/i18n/ProcedureMessages.java | 11 +
.../iotdb/confignode/i18n/ProcedureMessages.java | 11 +
.../procedure/env/RegionMaintainHandler.java | 55 +-
.../impl/node/RemoveDataNodesProcedure.java | 21 +-
.../impl/region/RemoveRegionPeerProcedure.java | 77 ++-
.../procedure/state/RemoveRegionPeerState.java | 2 +
.../impl/node/RemoveDataNodesProcedureTest.java | 302 ++++++++++
.../region/RemovePeerRestartConsensusTest.java | 634 +++++++++++++++++++++
.../impl/region/RestartDuringRemovePeerTest.java | 450 +++++++++++++++
.../iot/StalePeerListRestartReproductionTest.java | 272 +++++++++
13 files changed, 1827 insertions(+), 14 deletions(-)
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/batch/IoTDBRegionMigrateConfigNodeCrashIoTV2BatchIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/batch/IoTDBRegionMigrateConfigNodeCrashIoTV2BatchIT.java
index 5b81272acb9..c484931e750 100644
--- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/batch/IoTDBRegionMigrateConfigNodeCrashIoTV2BatchIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/batch/IoTDBRegionMigrateConfigNodeCrashIoTV2BatchIT.java
@@ -173,6 +173,8 @@ public class IoTDBRegionMigrateConfigNodeCrashIoTV2BatchIT
.collect(Collectors.toList()));
killConfigNodeKeywords.addAll(
Arrays.stream(RemoveRegionPeerState.values())
+ // IoTConsensus V2 finishes removal before the V1-only reset state.
+ .filter(state -> state != RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST)
.map(KillPoint::enumToString)
.collect(Collectors.toList()));
successTest(1, 1, 1, 2, killConfigNodeKeywords, noKillPoints(), KillNode.CONFIG_NODE);
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/stream/IoTDBRegionMigrateConfigNodeCrashIoTV2StreamIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/stream/IoTDBRegionMigrateConfigNodeCrashIoTV2StreamIT.java
index fc87df6eb83..a826b528b23 100644
--- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/stream/IoTDBRegionMigrateConfigNodeCrashIoTV2StreamIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/stream/IoTDBRegionMigrateConfigNodeCrashIoTV2StreamIT.java
@@ -187,6 +187,8 @@ public class IoTDBRegionMigrateConfigNodeCrashIoTV2StreamIT
.collect(Collectors.toList()));
killConfigNodeKeywords.addAll(
Arrays.stream(RemoveRegionPeerState.values())
+ // IoTConsensus V2 finishes removal before the V1-only reset state.
+ .filter(state -> state != RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST)
.map(KillPoint::enumToString)
.collect(Collectors.toList()));
successTest(1, 1, 1, 2, killConfigNodeKeywords, noKillPoints(), KillNode.CONFIG_NODE);
diff --git a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/ratis/IoTDBRegionMigrateConfigNodeCrashForRatisIT.java b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/ratis/IoTDBRegionMigrateConfigNodeCrashForRatisIT.java
index 975e1611109..304ad079dfe 100644
--- a/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/ratis/IoTDBRegionMigrateConfigNodeCrashForRatisIT.java
+++ b/integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/ratis/IoTDBRegionMigrateConfigNodeCrashForRatisIT.java
@@ -183,6 +183,8 @@ public class IoTDBRegionMigrateConfigNodeCrashForRatisIT
.collect(Collectors.toList()));
killConfigNodeKeywords.addAll(
Arrays.stream(RemoveRegionPeerState.values())
+ // Ratis finishes removal before the IoTConsensus V1-only reset state.
+ .filter(state -> state != RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST)
.map(KillPoint::enumToString)
.collect(Collectors.toList()));
successTest(1, 1, 1, 2, killConfigNodeKeywords, noKillPoints(), KillNode.CONFIG_NODE);
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 18aca012439..3874fbaac0d 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
@@ -847,6 +847,17 @@ public final class ProcedureMessages {
"RemoveRegionLocation remove region {} from DataNode {}, result is {}";
public static final String REMOVEREGIONPEER_STATE_FAILED = "RemoveRegionPeer state {} failed";
public static final String REMOVEREGIONPEER_STATE_SUCCESS = "RemoveRegionPeer state {} success";
+ public static final String
+ LOG_CANNOT_RECONCILE_PEER_LIST_FOR_ARG_AFTER_REMOVING_DATANODE_ARG_THE_CURRENT_REPLICA_SET_STILL_CONTAINS_THE_REMOVED_DATANODE_DBBF45DC =
+ "Cannot reconcile peer list for {} after removing DataNode {}: the current replica set still contains the removed DataNode.";
+ public static final String
+ LOG_FAILED_TO_RESET_PEER_LIST_FOR_ARG_ON_DATANODE_ARG_TO_ARG_ARG_C8D872B4 =
+ "Failed to reset peer list for {} on DataNode {} to {}: {}";
+ public static final String LOG_PIDARG_REMOVEREGION_STATE_ARG_FAILED_RETRY_ARG_ARG_IN_ARG_MS_41B8ECB5 =
+ "[pid{}][RemoveRegion] state {} failed, retry {}/{} in {} ms.";
+ public static final String
+ EXCEPTION_FAILED_TO_FINALIZE_REGION_PEER_REMOVAL_FOR_ARG_AT_STATE_ARG_AFTER_ARG_ATTEMPTS_382EA168 =
+ "Failed to finalize Region peer removal for %s at state %s after %d attempts.";
public static final String REMOVEREGION_RATIS_TRANSFER_LEADER_FAIL_BUT_PROCEDURE_WILL_CONTINUE =
"[RemoveRegion] Ratis transfer leader fail, but procedure will continue.";
public static final String REMOVE_CONFIG_NODE = "Remove Config Node";
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 14e52d72b9e..394480f6543 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
@@ -806,6 +806,17 @@ public final class ProcedureMessages {
"RemoveRegionLocation：从 DataNode {} 移除 region {}，结果为 {}";
public static final String REMOVEREGIONPEER_STATE_FAILED = "RemoveRegionPeer 状态 {} 失败";
public static final String REMOVEREGIONPEER_STATE_SUCCESS = "RemoveRegionPeer 状态 {} 成功";
+ public static final String
+ LOG_CANNOT_RECONCILE_PEER_LIST_FOR_ARG_AFTER_REMOVING_DATANODE_ARG_THE_CURRENT_REPLICA_SET_STILL_CONTAINS_THE_REMOVED_DATANODE_DBBF45DC =
+ "无法纠正 {} 在移除 DataNode {} 后的 peer 列表：当前副本集仍包含已移除的 DataNode。";
+ public static final String
+ LOG_FAILED_TO_RESET_PEER_LIST_FOR_ARG_ON_DATANODE_ARG_TO_ARG_ARG_C8D872B4 =
+ "将 {} 在 DataNode {} 上的 peer 列表重置为 {} 失败：{}";
+ public static final String LOG_PIDARG_REMOVEREGION_STATE_ARG_FAILED_RETRY_ARG_ARG_IN_ARG_MS_41B8ECB5 =
+ "[pid{}][RemoveRegion] 状态 {} 失败，第 {}/{} 次重试将在 {} 毫秒后进行。";
+ public static final String
+ EXCEPTION_FAILED_TO_FINALIZE_REGION_PEER_REMOVAL_FOR_ARG_AT_STATE_ARG_AFTER_ARG_ATTEMPTS_382EA168 =
+ "Region %s 的 peer 移除收尾在状态 %s 尝试 %d 次后失败。";
public static final String REMOVEREGION_RATIS_TRANSFER_LEADER_FAIL_BUT_PROCEDURE_WILL_CONTINUE =
"[RemoveRegion] Ratis 转换 leader 失败，但 procedure 将继续。";
public static final String REMOVE_CONFIG_NODE = "移除 ConfigNode";
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 ea08fbbb6c6..04100d05f9c 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
@@ -362,6 +362,58 @@ public class RegionMaintainHandler {
return clientHandler.getResponseMap();
}
+ public boolean isIoTConsensusV1DataRegion(TConsensusGroupId regionId) {
+ return TConsensusGroupType.DataRegion.equals(regionId.getType())
+ && IOT_CONSENSUS.equals(CONF.getDataRegionConsensusProtocolClass());
+ }
+
+ /**
+ * Reconcile retained IoTConsensus V1 replicas after the removed location has been committed. A
+ * restarting replica may have received the previous peer list while the removal was in progress.
+ * The existing reset RPC waits for consensus startup before applying this final list.
+ */
+ public boolean resetPeerListAfterRemoveForIoTConsensusV1(
+ TConsensusGroupId regionId, TDataNodeLocation removedDataNode) {
+ if (!isIoTConsensusV1DataRegion(regionId)) {
+ return true;
+ }
+ Optional<TRegionReplicaSet> replicaSet = getRegionReplicaSet(regionId);
+ if (!replicaSet.isPresent()) {
+ // Database deletion may have removed the entire Region from ConfigNode metadata. Its
+ // deletion tasks own the cleanup, and there are no retained replicas left to reconcile.
+ return true;
+ }
+ if (replicaSet.get().getDataNodeLocations().stream()
+ .anyMatch(location -> location.getDataNodeId() == removedDataNode.getDataNodeId())) {
+ LOGGER.warn(
+ ProcedureMessages
+ .LOG_CANNOT_RECONCILE_PEER_LIST_FOR_ARG_AFTER_REMOVING_DATANODE_ARG_THE_CURRENT_REPLICA_SET_STILL_CONTAINS_THE_REMOVED_DATANODE_DBBF45DC,
+ regionId,
+ removedDataNode.getDataNodeId());
+ return false;
+ }
+ List<TDataNodeLocation> correctLocations = replicaSet.get().getDataNodeLocations();
+ Map<Integer, TDataNodeLocation> retainedNodes = new HashMap<>();
+ correctLocations.forEach(location -> retainedNodes.put(location.getDataNodeId(), location));
+ // Include offline replicas so the caller can retry unsuccessful resets within its retry limit.
+ Map<Integer, TSStatus> responses = resetPeerList(regionId, correctLocations, retainedNodes);
+ boolean allSucceeded = true;
+ for (TDataNodeLocation location : correctLocations) {
+ TSStatus response = responses == null ? null : responses.get(location.getDataNodeId());
+ if (response == null || response.getCode() != TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+ LOGGER.warn(
+ ProcedureMessages
+ .LOG_FAILED_TO_RESET_PEER_LIST_FOR_ARG_ON_DATANODE_ARG_TO_ARG_ARG_C8D872B4,
+ regionId,
+ location.getDataNodeId(),
+ correctLocations,
+ response);
+ allSucceeded = false;
+ }
+ }
+ return allSucceeded;
+ }
+
// TODO: will use 'procedure yield' to refactor later
public TRegionMigrateResult waitTaskFinish(long taskId, TDataNodeLocation dataNodeLocation) {
return waitTaskFinish(taskId, dataNodeLocation, null);
@@ -451,7 +503,7 @@ public class RegionMaintainHandler {
.forceUpdateRegionCache(regionId, newLocation.getDataNodeId(), regionStatus);
}
- public void removeRegionLocation(
+ public TSStatus removeRegionLocation(
TConsensusGroupId regionId, TDataNodeLocation deprecatedLocation) {
RemoveRegionLocationPlan req = new RemoveRegionLocationPlan(regionId, deprecatedLocation);
TSStatus status = configManager.getPartitionManager().removeRegionLocation(req);
@@ -462,6 +514,7 @@ public class RegionMaintainHandler {
status);
configManager.getLoadManager().removeRegionCache(regionId, deprecatedLocation.getDataNodeId());
configManager.getLoadManager().getRouteBalancer().balanceRegionLeaderAndPriority();
+ return status;
}
/**
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 ff63a8531fb..06986a72458 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
@@ -54,6 +54,7 @@ import static org.apache.iotdb.confignode.conf.ConfigNodeConstant.REMOVE_DATANOD
public class RemoveDataNodesProcedure extends AbstractNodeProcedure<RemoveDataNodeState> {
private static final Logger LOG = LoggerFactory.getLogger(RemoveDataNodesProcedure.class);
private static final int RETRY_THRESHOLD = 5;
+ private static final long ROLLBACK_RETRY_INTERVAL_MS = 5_000;
private List<TDataNodeLocation> removedDataNodes;
@@ -86,6 +87,12 @@ public class RemoveDataNodesProcedure extends AbstractNodeProcedure<RemoveDataNo
}
}
+ @Override
+ protected void waitForLock(ConfigNodeProcedureEnv env) {
+ // This procedure does not acquire the node lock. Failed cleanup is rescheduled by the timeout
+ // executor; waiting on an unowned node lock would requeue it immediately.
+ }
+
@Override
protected void releaseLock(ConfigNodeProcedureEnv configNodeProcedureEnv) {
configNodeProcedureEnv.getSchedulerLock().lock();
@@ -268,7 +275,19 @@ public class RemoveDataNodesProcedure extends AbstractNodeProcedure<RemoveDataNo
@Override
protected void rollbackState(ConfigNodeProcedureEnv env, RemoveDataNodeState state)
- throws IOException, InterruptedException, ProcedureException {}
+ throws IOException {
+ if (state == RemoveDataNodeState.STOP_DATA_NODE) {
+ // The parent advances here before waiting for migration children. A failed child skips
+ // normal execution, so finish node cleanup without undoing migrations or clearing its
+ // failure.
+ checkRegionStatusAndStopDataNode(env);
+ }
+ }
+
+ @Override
+ protected long getRollbackRetryTimeout() {
+ return ROLLBACK_RETRY_INTERVAL_MS;
+ }
@Override
protected boolean isRollbackSupported(RemoveDataNodeState state) {
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionPeerProcedure.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionPeerProcedure.java
index 390eb63534c..01993f91440 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionPeerProcedure.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/region/RemoveRegionPeerProcedure.java
@@ -49,12 +49,15 @@ import static org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState.
import static org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState.DROP_CONSENSUS_PIPES;
import static org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE;
import static org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState.REMOVE_REGION_PEER;
+import static org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST;
import static org.apache.iotdb.rpc.TSStatusCode.SUCCESS_STATUS;
public class RemoveRegionPeerProcedure extends RegionOperationProcedure<RemoveRegionPeerState> {
private static final Logger LOGGER = LoggerFactory.getLogger(RemoveRegionPeerProcedure.class);
private static final int MAX_DELETE_OLD_REGION_PEER_RETRY = 3;
private static final long DELETE_OLD_REGION_PEER_RETRY_INTERVAL_MS = 5_000;
+ private static final int MAX_FINALIZATION_RETRY = 3;
+ private static final long FINALIZATION_RETRY_INTERVAL_MS = 5_000;
private TDataNodeLocation coordinator;
private TDataNodeLocation targetDataNode;
private transient int deleteOldRegionPeerAttempted = 0;
@@ -193,30 +196,80 @@ public class RemoveRegionPeerProcedure extends RegionOperationProcedure<RemoveRe
setNextState(REMOVE_REGION_LOCATION_CACHE);
break;
case REMOVE_REGION_LOCATION_CACHE:
- handler.removeRegionLocation(regionId, targetDataNode);
+ tsStatus = handler.removeRegionLocation(regionId, targetDataNode);
+ if (tsStatus == null || tsStatus.getCode() != SUCCESS_STATUS.getStatusCode()) {
+ return retryStep(state);
+ }
+ if (!handler.isIoTConsensusV1DataRegion(regionId)) {
+ return finishRemoval(state);
+ }
setKillPoint(state);
- LOGGER.info(ProcedureMessages.REMOVEREGIONPEER_STATE_SUCCESS, state);
- LOGGER.info(
- ProcedureMessages
- .PID_REMOVEREGION_SUCCESS_REGION_HAS_BEEN_REMOVED_FROM_DATANODE_PROCEDURE,
- getProcId(),
- regionId.getId(),
- targetDataNode.getDataNodeId(),
- CommonDateTimeUtils.convertMillisecondToDurationStr(
- System.currentTimeMillis() - getSubmittedTime()),
- DateTimeUtils.convertLongToDate(getSubmittedTime(), "ms"));
- return Flow.NO_MORE_STATE;
+ setNextState(RESET_IOT_CONSENSUS_V1_PEER_LIST);
+ break;
+ case RESET_IOT_CONSENSUS_V1_PEER_LIST:
+ if (!handler.resetPeerListAfterRemoveForIoTConsensusV1(regionId, targetDataNode)) {
+ return retryStep(state);
+ }
+ return finishRemoval(state);
default:
throw new ProcedureException(ProcedureMessages.UNSUPPORTED_STATE + state.name());
}
} catch (Exception e) {
LOGGER.error(ProcedureMessages.REMOVEREGIONPEER_STATE_FAILED, state, e);
+ if (state == REMOVE_REGION_LOCATION_CACHE || state == RESET_IOT_CONSENSUS_V1_PEER_LIST) {
+ return retryStep(state);
+ }
return Flow.NO_MORE_STATE;
}
LOGGER.info(ProcedureMessages.PID_REMOVEREGION_STATE_SUCCESS, getProcId(), state);
return Flow.HAS_MORE_STATE;
}
+ private Flow retryStep(RemoveRegionPeerState state) {
+ if (getCycles() >= MAX_FINALIZATION_RETRY) {
+ ProcedureException failure =
+ new ProcedureException(
+ String.format(
+ ProcedureMessages
+ .EXCEPTION_FAILED_TO_FINALIZE_REGION_PEER_REMOVAL_FOR_ARG_AT_STATE_ARG_AFTER_ARG_ATTEMPTS_382EA168,
+ regionId,
+ state,
+ getCycles() + 1));
+ LOGGER.error(ProcedureMessages.REMOVEREGIONPEER_STATE_FAILED, state, failure);
+ setFailure(failure);
+ return Flow.NO_MORE_STATE;
+ }
+ setNextState(state);
+ LOGGER.warn(
+ ProcedureMessages.LOG_PIDARG_REMOVEREGION_STATE_ARG_FAILED_RETRY_ARG_ARG_IN_ARG_MS_41B8ECB5,
+ getProcId(),
+ state,
+ getCycles() + 1,
+ MAX_FINALIZATION_RETRY,
+ FINALIZATION_RETRY_INTERVAL_MS);
+ try {
+ Thread.sleep(FINALIZATION_RETRY_INTERVAL_MS);
+ } catch (InterruptedException e) {
+ // Preserve the step for recovery when this ConfigNode stops or loses leadership.
+ Thread.currentThread().interrupt();
+ }
+ return Flow.HAS_MORE_STATE;
+ }
+
+ private Flow finishRemoval(RemoveRegionPeerState state) {
+ setKillPoint(state);
+ LOGGER.info(ProcedureMessages.REMOVEREGIONPEER_STATE_SUCCESS, state);
+ LOGGER.info(
+ ProcedureMessages.PID_REMOVEREGION_SUCCESS_REGION_HAS_BEEN_REMOVED_FROM_DATANODE_PROCEDURE,
+ getProcId(),
+ regionId.getId(),
+ targetDataNode.getDataNodeId(),
+ CommonDateTimeUtils.convertMillisecondToDurationStr(
+ System.currentTimeMillis() - getSubmittedTime()),
+ DateTimeUtils.convertLongToDate(getSubmittedTime(), "ms"));
+ return Flow.NO_MORE_STATE;
+ }
+
@Override
protected void rollbackState(ConfigNodeProcedureEnv env, RemoveRegionPeerState state)
throws IOException, InterruptedException, ProcedureException {}
diff --git a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RemoveRegionPeerState.java b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RemoveRegionPeerState.java
index a9c463cbafe..35e7cc956ec 100644
--- a/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RemoveRegionPeerState.java
+++ b/iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/state/RemoveRegionPeerState.java
@@ -25,4 +25,6 @@ public enum RemoveRegionPeerState {
DELETE_OLD_REGION_PEER,
REMOVE_REGION_LOCATION_CACHE,
DROP_CONSENSUS_PIPES,
+ // Reconcile retained DataRegion replicas after removal, only for IoTConsensus V1.
+ RESET_IOT_CONSENSUS_V1_PEER_LIST,
}
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodesProcedureTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodesProcedureTest.java
index 0bfe782048a..2769701c0d4 100644
--- a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodesProcedureTest.java
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodesProcedureTest.java
@@ -22,9 +22,35 @@ 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.TRegionMaintainTaskStatus;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
import org.apache.iotdb.commons.cluster.NodeStatus;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
+import org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSchemaPlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.AddRegionLocationPlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.RemoveRegionLocationPlan;
+import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan;
+import org.apache.iotdb.confignode.manager.ConfigManager;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.persistence.partition.PartitionInfo;
+import org.apache.iotdb.confignode.procedure.NoopProcedureStore;
+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.RegionMaintainHandler;
+import org.apache.iotdb.confignode.procedure.env.RemoveDataNodeHandler;
+import org.apache.iotdb.confignode.procedure.impl.region.RegionMigrateProcedure;
import org.apache.iotdb.confignode.procedure.impl.region.RegionMigrationPlan;
+import org.apache.iotdb.confignode.procedure.impl.region.RemoveRegionPeerProcedure;
+import org.apache.iotdb.confignode.procedure.scheduler.LockQueue;
+import org.apache.iotdb.confignode.procedure.state.ProcedureState;
+import org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState;
import org.apache.iotdb.confignode.procedure.store.ProcedureFactory;
+import org.apache.iotdb.confignode.procedure.util.ProcedureTestUtil;
+import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
+import org.apache.iotdb.mpp.rpc.thrift.TRegionMigrateResult;
+import org.apache.iotdb.rpc.TSStatusCode;
import org.apache.tsfile.utils.PublicBAOS;
import org.junit.Assert;
@@ -33,10 +59,32 @@ import org.junit.Test;
import java.io.DataOutputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReentrantLock;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.Mockito.RETURNS_DEEP_STUBS;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
public class RemoveDataNodesProcedureTest {
@@ -87,4 +135,258 @@ public class RemoveDataNodesProcedureTest {
Assert.assertEquals(regionMigrationPlan, RegionMigrationPlan.deserialize(buffer));
}
}
+
+ @Test
+ public void metadataRetryExhaustionCleansUpNodeRemoval() throws Exception {
+ assertFinalizationFailureCleanup(RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE, false);
+ }
+
+ @Test
+ public void peerListRetryExhaustionRetriesFailedNodeStatusCleanup() throws Exception {
+ assertFinalizationFailureCleanup(RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST, true);
+ }
+
+ private void assertFinalizationFailureCleanup(
+ RemoveRegionPeerState failingState, boolean failFirstStatusRestoration) throws Exception {
+ TDataNodeLocation runningNode = fromDataNodeLocation;
+ TDataNodeLocation readOnlyNode = location(12);
+ TDataNodeLocation drainedNode = location(13);
+ TDataNodeLocation retainedNode = location(14);
+ List<TDataNodeLocation> removalTargets = Arrays.asList(runningNode, readOnlyNode, drainedNode);
+ Map<Integer, NodeStatus> originalStatuses = new HashMap<>();
+ originalStatuses.put(runningNode.getDataNodeId(), NodeStatus.Running);
+ originalStatuses.put(readOnlyNode.getDataNodeId(), NodeStatus.ReadOnly);
+ originalStatuses.put(drainedNode.getDataNodeId(), NodeStatus.Running);
+ Map<Integer, NodeStatus> currentStatuses = new ConcurrentHashMap<>(originalStatuses);
+ Set<Integer> registeredNodes = ConcurrentHashMap.newKeySet();
+ registeredNodes.addAll(originalStatuses.keySet());
+ Set<Integer> stoppedNodes = ConcurrentHashMap.newKeySet();
+ AtomicInteger restorationAttempts = new AtomicInteger();
+ AtomicLong firstRestorationFailureTime = new AtomicLong();
+ AtomicLong restorationRetryDelay = new AtomicLong();
+
+ TConsensusGroupId pendingRunningRegion =
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 2);
+ TConsensusGroupId pendingReadOnlyRegion =
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 3);
+ PartitionInfo partitionInfo = new PartitionInfo();
+ partitionInfo.createDatabase(
+ new DatabaseSchemaPlan(
+ ConfigPhysicalPlanType.CreateDatabase, new TDatabaseSchema("root.remove")));
+ CreateRegionGroupsPlan groups = new CreateRegionGroupsPlan();
+ groups.addRegionGroup(
+ "root.remove",
+ new TRegionReplicaSet(
+ consensusGroupId, new ArrayList<>(Arrays.asList(runningNode, retainedNode))));
+ groups.addRegionGroup(
+ "root.remove",
+ new TRegionReplicaSet(
+ pendingRunningRegion, new ArrayList<>(Arrays.asList(runningNode, retainedNode))));
+ groups.addRegionGroup(
+ "root.remove",
+ new TRegionReplicaSet(
+ pendingReadOnlyRegion, new ArrayList<>(Arrays.asList(readOnlyNode, retainedNode))));
+ partitionInfo.createRegionGroups(groups);
+
+ ConfigManager configManager = mock(ConfigManager.class, RETURNS_DEEP_STUBS);
+ PartitionManager partitionManager = mock(PartitionManager.class);
+ when(configManager.getPartitionManager()).thenReturn(partitionManager);
+ when(partitionManager.getAllReplicaSets())
+ .thenAnswer(ignored -> partitionInfo.getAllReplicaSets());
+ when(partitionManager.isDataNodeContainsRegion(anyInt(), any()))
+ .thenAnswer(
+ call ->
+ partitionInfo.getAllReplicaSets((Integer) call.getArgument(0)).stream()
+ .anyMatch(replica -> replica.getRegionId().equals(call.getArgument(1))));
+
+ RegionMaintainHandler regionHandler = mock(RegionMaintainHandler.class);
+ when(regionHandler.filterDataNodeWithOtherRegionReplica(any(), any(TDataNodeLocation.class)))
+ .thenReturn(Optional.of(retainedNode));
+ doAnswer(
+ call ->
+ partitionInfo.addRegionLocation(
+ new AddRegionLocationPlan(call.getArgument(0), call.getArgument(1))))
+ .when(regionHandler)
+ .addRegionLocation(any(), any());
+ when(regionHandler.createNewRegionPeer(any(), any())).thenReturn(success());
+ when(regionHandler.submitAddRegionPeerTask(anyLong(), any(), any(), any()))
+ .thenReturn(success());
+ when(regionHandler.submitRemoveRegionPeerTask(anyLong(), any(), any(), any()))
+ .thenAnswer(
+ call -> {
+ removalTargets.forEach(
+ node ->
+ Assert.assertEquals(
+ NodeStatus.Removing, currentStatuses.get(node.getDataNodeId())));
+ return success();
+ });
+ when(regionHandler.submitDeleteOldRegionPeerTask(anyLong(), any(), any()))
+ .thenReturn(success());
+ TRegionMigrateResult taskSuccess = new TRegionMigrateResult(TRegionMaintainTaskStatus.SUCCESS);
+ when(regionHandler.waitTaskFinish(anyLong(), any())).thenReturn(taskSuccess);
+ when(regionHandler.waitTaskFinish(anyLong(), any(), any())).thenReturn(taskSuccess);
+ when(regionHandler.isIoTConsensusV1DataRegion(consensusGroupId)).thenReturn(true);
+ when(regionHandler.removeRegionLocation(consensusGroupId, runningNode))
+ .thenAnswer(
+ ignored ->
+ failingState == RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE
+ ? new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode())
+ : partitionInfo.removeRegionLocation(
+ new RemoveRegionLocationPlan(consensusGroupId, runningNode)));
+ when(regionHandler.resetPeerListAfterRemoveForIoTConsensusV1(consensusGroupId, runningNode))
+ .thenReturn(false);
+
+ RemoveDataNodeHandler nodeHandler = mock(RemoveDataNodeHandler.class);
+ when(nodeHandler.checkEnoughDataNodeAfterRemoving(removalTargets)).thenReturn(true);
+ RegionMigrationPlan migration = new RegionMigrationPlan(consensusGroupId, runningNode);
+ migration.setToDataNode(toDataNodeLocation);
+ // No destination was selected for the other regions, so they remain on the removal targets.
+ when(nodeHandler.selectedRegionMigrationPlans(removalTargets))
+ .thenReturn(Collections.singletonList(migration));
+ doAnswer(
+ call -> {
+ List<TDataNodeLocation> nodes = call.getArgument(0);
+ Map<Integer, NodeStatus> statuses = call.getArgument(1);
+ boolean restoring = statuses.get(runningNode.getDataNodeId()) == NodeStatus.Running;
+ if (restoring) {
+ int attempt = restorationAttempts.incrementAndGet();
+ Assert.assertTrue(registeredNodes.contains(readOnlyNode.getDataNodeId()));
+ if (failFirstStatusRestoration && attempt == 1) {
+ currentStatuses.put(runningNode.getDataNodeId(), NodeStatus.Running);
+ firstRestorationFailureTime.set(System.nanoTime());
+ throw new IOException("Injected partial node status restoration failure");
+ }
+ if (failFirstStatusRestoration) {
+ restorationRetryDelay.set(System.nanoTime() - firstRestorationFailureTime.get());
+ }
+ }
+ nodes.forEach(
+ node ->
+ currentStatuses.put(
+ node.getDataNodeId(), statuses.get(node.getDataNodeId())));
+ return null;
+ })
+ .when(nodeHandler)
+ .changeDataNodeStatus(anyList(), anyMap());
+ doAnswer(
+ call -> {
+ List<TDataNodeLocation> nodes = call.getArgument(0);
+ nodes.forEach(node -> registeredNodes.remove(node.getDataNodeId()));
+ return null;
+ })
+ .when(nodeHandler)
+ .removeDataNodePersistence(anyList());
+ doAnswer(
+ call -> {
+ List<TDataNodeLocation> nodes = call.getArgument(0);
+ nodes.forEach(node -> stoppedNodes.add(node.getDataNodeId()));
+ return null;
+ })
+ .when(nodeHandler)
+ .stopDataNodes(anyList());
+
+ ConfigNodeProcedureEnv env = mock(ConfigNodeProcedureEnv.class);
+ when(env.getConfigManager()).thenReturn(configManager);
+ when(env.getRegionMaintainHandler()).thenReturn(regionHandler);
+ when(env.getRemoveDataNodeHandler()).thenReturn(nodeHandler);
+ when(env.getSchedulerLock()).thenReturn(new ReentrantLock());
+ when(env.getNodeLock()).thenReturn(new LockQueue());
+ RecordingProcedureStore store = new RecordingProcedureStore();
+ ProcedureExecutor<ConfigNodeProcedureEnv> executor = new ProcedureExecutor<>(env, store);
+ when(env.getScheduler()).thenReturn(executor.getScheduler());
+ store.start();
+ executor.init(1);
+ executor.startWorkers();
+ try {
+ RemoveDataNodesProcedure root =
+ new RemoveDataNodesProcedure(removalTargets, originalStatuses);
+ long rootId = executor.submitProcedure(root);
+ ProcedureTestUtil.waitForProcedure(executor, rootId);
+ Assert.assertTrue("Node removal must finish failure cleanup", executor.isFinished(rootId));
+ Assert.assertEquals(ProcedureState.ROLLEDBACK, root.getState());
+ Assert.assertFalse(root.isSuccess());
+ RemoveRegionPeerProcedure failedChild =
+ store.recorded.values().stream()
+ .filter(RemoveRegionPeerProcedure.class::isInstance)
+ .map(RemoveRegionPeerProcedure.class::cast)
+ .findFirst()
+ .get();
+ Procedure<?> migrationParent = store.recorded.get(failedChild.getParentProcId());
+ Assert.assertTrue(migrationParent instanceof RegionMigrateProcedure);
+ Assert.assertEquals(rootId, migrationParent.getParentProcId());
+ Assert.assertSame(failedChild.getException(), root.getException());
+ Assert.assertTrue(root.getException().getMessage().contains(failingState.name()));
+ if (failingState == RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE) {
+ verify(regionHandler, times(4)).removeRegionLocation(consensusGroupId, runningNode);
+ verify(regionHandler, never()).resetPeerListAfterRemoveForIoTConsensusV1(any(), any());
+ } else {
+ verify(regionHandler).removeRegionLocation(consensusGroupId, runningNode);
+ verify(regionHandler, times(4))
+ .resetPeerListAfterRemoveForIoTConsensusV1(consensusGroupId, runningNode);
+ }
+
+ // Failed children bypass STOP_DATA_NODE: root rollback must still finalize node statuses.
+ Assert.assertEquals(NodeStatus.Running, currentStatuses.get(runningNode.getDataNodeId()));
+ Assert.assertEquals(NodeStatus.ReadOnly, currentStatuses.get(readOnlyNode.getDataNodeId()));
+ Assert.assertTrue(registeredNodes.contains(runningNode.getDataNodeId()));
+ Assert.assertTrue(registeredNodes.contains(readOnlyNode.getDataNodeId()));
+ Assert.assertFalse(registeredNodes.contains(drainedNode.getDataNodeId()));
+ Assert.assertEquals(Collections.singleton(drainedNode.getDataNodeId()), stoppedNodes);
+ Assert.assertEquals(failFirstStatusRestoration ? 2 : 1, restorationAttempts.get());
+ if (failFirstStatusRestoration) {
+ // A node-lock wakeup must not bypass the five-second rollback retry timeout.
+ Assert.assertTrue(
+ "Status restoration must wait for the rollback retry timeout",
+ restorationRetryDelay.get() >= TimeUnit.SECONDS.toNanos(4));
+ }
+ verify(nodeHandler).broadcastDataNodeStatusChange(Arrays.asList(runningNode, readOnlyNode));
+
+ // Cleanup must not undo the completed add peer or a committed removal.
+ Assert.assertTrue(
+ partitionManager.isDataNodeContainsRegion(
+ toDataNodeLocation.getDataNodeId(), consensusGroupId));
+ Assert.assertEquals(
+ failingState == RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE,
+ partitionManager.isDataNodeContainsRegion(runningNode.getDataNodeId(), consensusGroupId));
+ Assert.assertTrue(
+ partitionManager.isDataNodeContainsRegion(
+ runningNode.getDataNodeId(), pendingRunningRegion));
+ Assert.assertTrue(
+ partitionManager.isDataNodeContainsRegion(
+ readOnlyNode.getDataNodeId(), pendingReadOnlyRegion));
+ } finally {
+ executor.stop();
+ executor.join();
+ store.stop();
+ store.cleanup();
+ }
+ }
+
+ private static TSStatus success() {
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+
+ private static TDataNodeLocation location(int id) {
+ return new TDataNodeLocation(
+ id,
+ new TEndPoint("127.0.0.1", 6000 + id),
+ new TEndPoint("127.0.0.1", 7000 + id),
+ new TEndPoint("127.0.0.1", 8000 + id),
+ new TEndPoint("127.0.0.1", 9000 + id),
+ new TEndPoint("127.0.0.1", 10000 + id));
+ }
+
+ private static class RecordingProcedureStore extends NoopProcedureStore {
+ private final Map<Long, Procedure<?>> recorded = new ConcurrentHashMap<>();
+
+ @Override
+ public void update(Procedure procedure) {
+ recorded.put(procedure.getProcId(), procedure);
+ }
+
+ @Override
+ public void update(Procedure[] procedures) {
+ Arrays.stream(procedures).forEach(this::update);
+ }
+ }
}
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RemovePeerRestartConsensusTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RemovePeerRestartConsensusTest.java
new file mode 100644
index 00000000000..b03f06c95a2
--- /dev/null
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RemovePeerRestartConsensusTest.java
@@ -0,0 +1,634 @@
+/*
+ * 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.region;
+
+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.TRegionMaintainTaskStatus;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.conf.CommonDescriptor;
+import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.commons.request.IConsensusRequest;
+import org.apache.iotdb.confignode.client.async.CnToDnInternalServiceAsyncRequestManager;
+import org.apache.iotdb.confignode.conf.ConfigNodeConfig;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
+import org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSchemaPlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.RemoveRegionLocationPlan;
+import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan;
+import org.apache.iotdb.confignode.manager.ConfigManager;
+import org.apache.iotdb.confignode.persistence.partition.PartitionInfo;
+import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import org.apache.iotdb.confignode.procedure.env.RegionMaintainHandler;
+import org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState;
+import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
+import org.apache.iotdb.consensus.ConsensusFactory;
+import org.apache.iotdb.consensus.IStateMachine;
+import org.apache.iotdb.consensus.common.DataSet;
+import org.apache.iotdb.consensus.common.Peer;
+import org.apache.iotdb.consensus.common.request.BatchIndexedConsensusRequest;
+import org.apache.iotdb.consensus.common.request.DeserializedBatchIndexedConsensusRequest;
+import org.apache.iotdb.consensus.common.request.IndexedConsensusRequest;
+import org.apache.iotdb.consensus.config.ConsensusConfig;
+import org.apache.iotdb.consensus.config.IoTConsensusConfig;
+import org.apache.iotdb.consensus.iot.IoTConsensus;
+import org.apache.iotdb.consensus.iot.IoTConsensusServerImpl;
+import org.apache.iotdb.consensus.iot.log.ConsensusReqReader;
+import org.apache.iotdb.consensus.iot.log.GetConsensusReqReaderPlan;
+import org.apache.iotdb.consensus.iot.logdispatcher.LogDispatcher;
+import org.apache.iotdb.consensus.iot.logdispatcher.LogDispatcher.LogDispatcherThread;
+import org.apache.iotdb.mpp.rpc.thrift.IDataNodeRPCService;
+import org.apache.iotdb.mpp.rpc.thrift.TRegionMigrateResult;
+import org.apache.iotdb.mpp.rpc.thrift.TResetPeerListReq;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.apache.thrift.protocol.TBinaryProtocol;
+import org.apache.thrift.protocol.TCompactProtocol;
+import org.apache.thrift.server.TThreadPoolServer;
+import org.apache.thrift.transport.TServerSocket;
+import org.apache.thrift.transport.layered.TFramedTransport;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+import org.mockito.Mockito;
+
+import java.io.File;
+import java.lang.reflect.Field;
+import java.net.ServerSocket;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+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 java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+
+/**
+ * Runs the ConfigNode removal state machine against real IoTConsensus replication services. Only
+ * ConfigNode persistence transport and the DataNode internal-service facade are adapted: metadata
+ * changes use PartitionInfo, and the production reset RPC crosses a loopback Thrift connection. The
+ * in-memory state machine stands in for DataRegion/WAL recovery; no full DataNode JVM is started.
+ */
+public class RemovePeerRestartConsensusTest {
+
+ private static final DataRegionId GROUP = new DataRegionId(92);
+ private static final TConsensusGroupId REGION = GROUP.convertToTConsensusGroupId();
+ private static final int INITIAL_WRITES = 8;
+ private static final int WRITES_DURING_STALE_MEMBERSHIP = 8;
+
+ @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Test
+ public void removalProcedureRepairsRestartedPeerAndWritesContinue() throws Exception {
+ final ConfigNodeConfig cnConfig = ConfigNodeDescriptor.getInstance().getConf();
+ final String originalProtocol = cnConfig.getDataRegionConsensusProtocolClass();
+ cnConfig.setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS);
+ final ConcurrentMap<Integer, IoTConsensus> servers = new ConcurrentHashMap<>();
+ final List<LoopbackInternalService> internalServices = new ArrayList<>();
+ final List<Peer> peers = new ArrayList<>();
+ final List<TDataNodeLocation> locations = new ArrayList<>();
+ final List<File> storageDirectories = new ArrayList<>();
+ final List<MemoryStateMachine> stateMachines = new ArrayList<>();
+ final ConcurrentMap<Integer, TResetPeerListReq> resetsReceived = new ConcurrentHashMap<>();
+ try {
+ for (int nodeId = 1; nodeId <= 3; nodeId++) {
+ final LoopbackInternalService internal =
+ new LoopbackInternalService(nodeId, servers, resetsReceived);
+ internalServices.add(internal);
+ final TEndPoint consensusEndpoint = new TEndPoint("127.0.0.1", unusedPort());
+ locations.add(location(nodeId, internal.getEndpoint(), consensusEndpoint));
+ peers.add(new Peer(GROUP, nodeId, consensusEndpoint));
+ storageDirectories.add(temporaryFolder.newFolder("peer-" + nodeId));
+ stateMachines.add(new MemoryStateMachine());
+ }
+ for (int index = 0; index < peers.size(); index++) {
+ final IoTConsensus server =
+ newServer(peers.get(index), storageDirectories.get(index), stateMachines.get(index));
+ servers.put(peers.get(index).getNodeId(), server);
+ server.start();
+ server.createLocalPeer(GROUP, peers);
+ }
+ final Peer writer = peers.get(0);
+ final Peer removed = peers.get(1);
+ final Peer coordinator = peers.get(2);
+ final List<Peer> finalPeers = Arrays.asList(writer, coordinator);
+ final IoTConsensus originalWriter = servers.get(writer.getNodeId());
+ write(originalWriter, INITIAL_WRITES);
+ await("initial replication", () -> originalWriter.getImpl(GROUP).getSyncLag() == 0);
+ final LogDispatcherThread originalRemovedThread =
+ dispatcher(originalWriter.getImpl(GROUP), removed);
+
+ final PartitionInfo partitionInfo = newPartitionInfo(locations);
+ final ConfigManager configManager =
+ Mockito.mock(ConfigManager.class, Mockito.RETURNS_DEEP_STUBS);
+ Mockito.when(configManager.getPartitionManager().getAllReplicaSets())
+ .thenAnswer(ignored -> partitionInfo.getAllReplicaSets());
+ Mockito.when(
+ configManager
+ .getPartitionManager()
+ .removeRegionLocation(any(RemoveRegionLocationPlan.class)))
+ .thenAnswer(invocation -> partitionInfo.removeRegionLocation(invocation.getArgument(0)));
+ final RegionMaintainHandler handler = Mockito.spy(new RegionMaintainHandler(configManager));
+ // These adapters replace asynchronous DN task submission, not the procedure's final reset.
+ Mockito.doNothing().when(handler).transferRegionLeader(any(), any(), any());
+ Mockito.doAnswer(
+ invocation -> {
+ servers.get(coordinator.getNodeId()).removeRemotePeer(GROUP, removed);
+ return success();
+ })
+ .when(handler)
+ .submitRemoveRegionPeerTask(anyLong(), any(), any(), any());
+ Mockito.doAnswer(
+ invocation -> {
+ servers.get(removed.getNodeId()).deleteLocalPeer(GROUP);
+ return success();
+ })
+ .when(handler)
+ .submitDeleteOldRegionPeerTask(anyLong(), any(), any());
+ Mockito.doReturn(new TRegionMigrateResult(TRegionMaintainTaskStatus.SUCCESS))
+ .when(handler)
+ .waitTaskFinish(anyLong(), any(TDataNodeLocation.class));
+ final ConfigNodeProcedureEnv env = Mockito.mock(ConfigNodeProcedureEnv.class);
+ Mockito.when(env.getRegionMaintainHandler()).thenReturn(handler);
+ final RemoveRegionPeerProcedure procedure =
+ new RemoveRegionPeerProcedure(REGION, locations.get(2), locations.get(1));
+ procedure.executeFromState(env, RemoveRegionPeerState.TRANSFER_REGION_LEADER);
+ procedure.executeFromState(env, RemoveRegionPeerState.REMOVE_REGION_PEER);
+ assertEquals(finalPeers, originalWriter.getImpl(GROUP).getConfiguration());
+ assertEquals(
+ Collections.singletonList(coordinator), dispatcherPeers(originalWriter.getImpl(GROUP)));
+ assertTrue(originalRemovedThread.isStopped());
+
+ // Obtain the same persisted replica-set source used for DataNode restart responses while
+ // the procedure is still between REMOVE_REGION_PEER and REMOVE_REGION_LOCATION_CACHE.
+ final List<TDataNodeLocation> staleRestartLocations =
+ partitionInfo.getAllReplicaSets(writer.getNodeId()).get(0).getDataNodeLocations();
+ assertTrue(staleRestartLocations.contains(locations.get(1)));
+ assertTrue(resetsReceived.isEmpty());
+ procedure.executeFromState(env, RemoveRegionPeerState.DELETE_OLD_REGION_PEER);
+ procedure.executeFromState(env, RemoveRegionPeerState.DROP_CONSENSUS_PIPES);
+ originalWriter.stop();
+ servers.remove(writer.getNodeId());
+ final IoTConsensus restartedWriter =
+ newServer(writer, storageDirectories.get(0), stateMachines.get(0));
+ servers.put(writer.getNodeId(), restartedWriter);
+ restartedWriter.recordCorrectPeerListBeforeStarting(
+ Collections.singletonMap(GROUP, toPeers(staleRestartLocations)));
+ restartedWriter.start();
+ final IoTConsensusServerImpl recovered = restartedWriter.getImpl(GROUP);
+ assertEquals(peers, recovered.getConfiguration());
+ final LogDispatcherThread recreatedRemovedThread = dispatcher(recovered, removed);
+ assertNotSame(originalRemovedThread, recreatedRemovedThread);
+ assertFalse(recreatedRemovedThread.isStopped());
+ write(restartedWriter, WRITES_DURING_STALE_MEMBERSHIP);
+ final LogDispatcherThread healthyThread = dispatcher(recovered, coordinator);
+ final long expectedIndex = INITIAL_WRITES + WRITES_DURING_STALE_MEMBERSHIP;
+ await(
+ "healthy peer receives new writes",
+ () -> healthyThread.getCurrentSyncIndex() == expectedIndex);
+ await(
+ "removed peer fills its five-batch pipeline",
+ () -> {
+ synchronized (recreatedRemovedThread.getSyncStatus()) {
+ return recreatedRemovedThread.getSyncStatus().getPendingBatches().size() == 5;
+ }
+ });
+ assertEquals(INITIAL_WRITES, recreatedRemovedThread.getCurrentSyncIndex());
+ assertEquals(WRITES_DURING_STALE_MEMBERSHIP, recovered.getSyncLag());
+ printState("STALE_RESTART_BEFORE_PROCEDURE_FINALIZATION", recovered);
+
+ // Only this production state transition may trigger reset. The test body never invokes it.
+ procedure.executeFromState(env, RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE);
+ assertTrue(resetsReceived.isEmpty());
+ procedure.executeFromState(env, RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST);
+ assertEquals(
+ new HashSet<>(Arrays.asList(writer.getNodeId(), coordinator.getNodeId())),
+ resetsReceived.keySet());
+ for (TResetPeerListReq request : resetsReceived.values()) {
+ assertEquals(REGION, request.getRegionId());
+ assertEquals(finalPeers, toPeers(request.getCorrectLocations()));
+ }
+ assertEquals(finalPeers, recovered.getConfiguration());
+ assertEquals(Collections.singletonList(coordinator), dispatcherPeers(recovered));
+ assertTrue(recreatedRemovedThread.isStopped());
+ assertEquals(0, recovered.getSyncLag());
+ assertFalse(
+ partitionInfo
+ .getAllReplicaSets(writer.getNodeId())
+ .get(0)
+ .getDataNodeLocations()
+ .contains(locations.get(1)));
+ // The earlier response remains stale; correctness comes from the production reset RPC.
+ assertTrue(staleRestartLocations.contains(locations.get(1)));
+ printState("PROCEDURE_FINALIZATION_REPAIRED_MEMBERSHIP", recovered);
+
+ write(restartedWriter, 2);
+ await("writes after procedure finalization", () -> recovered.getSyncLag() == 0);
+ assertEquals(expectedIndex + 2, stateMachines.get(2).values.size());
+ printState("WRITES_CONTINUE_AFTER_REPAIR", recovered);
+ restartedWriter.stop();
+ servers.remove(writer.getNodeId());
+ final IoTConsensus freshWriter =
+ newServer(writer, storageDirectories.get(0), stateMachines.get(0));
+ servers.put(writer.getNodeId(), freshWriter);
+ freshWriter.recordCorrectPeerListBeforeStarting(
+ Collections.singletonMap(
+ GROUP,
+ toPeers(
+ partitionInfo
+ .getAllReplicaSets(writer.getNodeId())
+ .get(0)
+ .getDataNodeLocations())));
+ freshWriter.start();
+ assertEquals(finalPeers, freshWriter.getImpl(GROUP).getConfiguration());
+ assertEquals(
+ Collections.singletonList(coordinator), dispatcherPeers(freshWriter.getImpl(GROUP)));
+ write(freshWriter, 1);
+ await(
+ "second restart uses committed membership",
+ () -> freshWriter.getImpl(GROUP).getSyncLag() == 0);
+ assertEquals(expectedIndex + 3, stateMachines.get(2).values.size());
+ printState("SECOND_RESTART_DOES_NOT_RECREATE_REMOVED_PEER", freshWriter.getImpl(GROUP));
+ } finally {
+ servers.values().parallelStream().forEach(IoTConsensus::stop);
+ for (LoopbackInternalService internal : internalServices) {
+ internal.close();
+ }
+ cnConfig.setDataRegionConsensusProtocolClass(originalProtocol);
+ }
+ }
+
+ private static PartitionInfo newPartitionInfo(List<TDataNodeLocation> locations) {
+ final PartitionInfo partitionInfo = new PartitionInfo();
+ partitionInfo.createDatabase(
+ new DatabaseSchemaPlan(
+ ConfigPhysicalPlanType.CreateDatabase, new TDatabaseSchema("root.restart_repair")));
+ final CreateRegionGroupsPlan groups = new CreateRegionGroupsPlan();
+ groups.addRegionGroup(
+ "root.restart_repair", new TRegionReplicaSet(REGION, new ArrayList<>(locations)));
+ partitionInfo.createRegionGroups(groups);
+ return partitionInfo;
+ }
+
+ private static TDataNodeLocation location(int nodeId, TEndPoint internal, TEndPoint consensus) {
+ return new TDataNodeLocation(
+ nodeId,
+ new TEndPoint("127.0.0.1", 6667),
+ internal,
+ new TEndPoint("127.0.0.1", 10740),
+ consensus,
+ new TEndPoint("127.0.0.1", 10760));
+ }
+
+ private static List<Peer> toPeers(List<TDataNodeLocation> locations) {
+ return locations.stream()
+ .map(location -> Peer.valueOf(REGION, location))
+ .sorted()
+ .collect(Collectors.toList());
+ }
+
+ private static IoTConsensus newServer(
+ Peer peer, File directory, MemoryStateMachine stateMachine) {
+ return new IoTConsensus(
+ ConsensusConfig.newBuilder()
+ .setThisNodeId(peer.getNodeId())
+ .setThisNode(peer.getEndpoint())
+ .setStorageDir(directory.getAbsolutePath())
+ .setConsensusGroupType(TConsensusGroupType.DataRegion)
+ .setIoTConsensusConfig(
+ IoTConsensusConfig.newBuilder()
+ .setRpc(
+ IoTConsensusConfig.RPC.newBuilder().setConnectionTimeoutInMs(500).build())
+ .setReplication(
+ IoTConsensusConfig.Replication.newBuilder()
+ .setCheckpointGap(1)
+ .setMaxLogEntriesNumPerBatch(1)
+ .setMaxPendingBatchesNum(5)
+ .setBasicRetryWaitTimeMs(100)
+ .setMaxRetryWaitTimeMs(500)
+ .build())
+ .build())
+ .build(),
+ ignored -> stateMachine);
+ }
+
+ private static TSStatus success() {
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+
+ private static int unusedPort() throws Exception {
+ try (ServerSocket socket = new ServerSocket(0)) {
+ return socket.getLocalPort();
+ }
+ }
+
+ private static void write(IoTConsensus server, int count) throws Exception {
+ final int previousCount = (int) server.getImpl(GROUP).getSearchIndex();
+ for (int index = 1; index <= count; index++) {
+ assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ server.write(GROUP, new IntegerRequest(previousCount + index)).getCode());
+ }
+ }
+
+ private static void await(String description, BooleanSupplier condition) throws Exception {
+ final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(25);
+ while (!condition.getAsBoolean() && System.nanoTime() < deadline) {
+ Thread.sleep(20);
+ }
+ assertTrue(description, condition.getAsBoolean());
+ }
+
+ @SuppressWarnings("unchecked")
+ private static List<LogDispatcherThread> dispatchers(IoTConsensusServerImpl server)
+ throws Exception {
+ final LogDispatcher dispatcher = server.getLogDispatcher();
+ final Field field = LogDispatcher.class.getDeclaredField("threads");
+ field.setAccessible(true);
+ synchronized (dispatcher) {
+ return new ArrayList<>((List<LogDispatcherThread>) field.get(dispatcher));
+ }
+ }
+
+ private static List<Peer> dispatcherPeers(IoTConsensusServerImpl server) throws Exception {
+ return dispatchers(server).stream()
+ .map(LogDispatcherThread::getPeer)
+ .collect(Collectors.toList());
+ }
+
+ private static LogDispatcherThread dispatcher(IoTConsensusServerImpl server, Peer peer)
+ throws Exception {
+ return dispatchers(server).stream()
+ .filter(thread -> thread.getPeer().equals(peer))
+ .findFirst()
+ .get();
+ }
+
+ private static void printState(String stage, IoTConsensusServerImpl server) throws Exception {
+ System.out.printf(
+ "CN_REPAIR %s searchIndex=%d minSyncIndex=%d syncLag=%d dispatchers=%s%n",
+ stage,
+ server.getSearchIndex(),
+ server.getMinSyncIndex(),
+ server.getSyncLag(),
+ dispatchers(server).stream()
+ .map(thread -> thread.getPeer().getNodeId() + ":" + thread.getCurrentSyncIndex())
+ .collect(Collectors.toList()));
+ }
+
+ /** A real Thrift endpoint forwarding only resetPeerList to the selected consensus instance. */
+ private static class LoopbackInternalService implements AutoCloseable {
+ private final TEndPoint endpoint;
+ private final ExecutorService workers;
+ private final TThreadPoolServer service;
+ private final Thread serverThread;
+
+ private LoopbackInternalService(
+ int nodeId,
+ Map<Integer, IoTConsensus> servers,
+ Map<Integer, TResetPeerListReq> resetsReceived)
+ throws Exception {
+ final ServerSocket socket = new ServerSocket(0);
+ endpoint = new TEndPoint("127.0.0.1", socket.getLocalPort());
+ final IDataNodeRPCService.Iface facade = Mockito.mock(IDataNodeRPCService.Iface.class);
+ Mockito.when(facade.resetPeerList(any(TResetPeerListReq.class)))
+ .thenAnswer(
+ invocation -> {
+ final TResetPeerListReq request = invocation.getArgument(0);
+ servers.get(nodeId).resetPeerList(GROUP, toPeers(request.getCorrectLocations()));
+ resetsReceived.put(nodeId, request.deepCopy());
+ return success();
+ });
+ workers =
+ Executors.newCachedThreadPool(
+ runnable -> {
+ final Thread thread = new Thread(runnable, "reset-peer-facade-" + nodeId);
+ thread.setDaemon(true);
+ return thread;
+ });
+ service =
+ new TThreadPoolServer(
+ new TThreadPoolServer.Args(new TServerSocket(socket, 1000))
+ .processor(new IDataNodeRPCService.Processor<>(facade))
+ .protocolFactory(
+ CommonDescriptor.getInstance().getConfig().isRpcThriftCompressionEnabled()
+ ? new TCompactProtocol.Factory()
+ : new TBinaryProtocol.Factory())
+ .transportFactory(new TFramedTransport.Factory())
+ .executorService(workers)
+ .stopTimeoutVal(1)
+ .stopTimeoutUnit(TimeUnit.SECONDS));
+ serverThread = new Thread(service::serve, "reset-peer-server-" + nodeId);
+ serverThread.setDaemon(true);
+ serverThread.start();
+ await("internal Thrift facade starts", service::isServing);
+ }
+
+ private TEndPoint getEndpoint() {
+ return endpoint;
+ }
+
+ @Override
+ public void close() throws Exception {
+ CnToDnInternalServiceAsyncRequestManager.getInstance().resetClient(endpoint);
+ service.stop();
+ workers.shutdownNow();
+ serverThread.join(TimeUnit.SECONDS.toMillis(3));
+ }
+ }
+
+ private static class IntegerRequest implements IConsensusRequest {
+ private final int value;
+
+ private IntegerRequest(int value) {
+ this.value = value;
+ }
+
+ @Override
+ public ByteBuffer serializeToByteBuffer() {
+ return ByteBuffer.allocate(Integer.BYTES).putInt(value).flip();
+ }
+ }
+
+ /** Keeps local WAL entries across consensus-instance restarts and records replicated values. */
+ private static class MemoryStateMachine implements IStateMachine, ConsensusReqReader, DataSet {
+ private final Map<Long, IndexedConsensusRequest> localRequests = new HashMap<>();
+ private final Set<Integer> values = ConcurrentHashMap.newKeySet();
+ private final AtomicLong localIndex = new AtomicLong();
+
+ @Override
+ public void start() {}
+
+ @Override
+ public void stop() {}
+
+ @Override
+ public TSStatus write(IConsensusRequest request) {
+ if (request instanceof DeserializedBatchIndexedConsensusRequest) {
+ final List<TSStatus> statuses = new ArrayList<>();
+ for (IConsensusRequest element :
+ ((DeserializedBatchIndexedConsensusRequest) request).getInsertNodes()) {
+ write(element);
+ statuses.add(success());
+ }
+ return success().setSubStatus(statuses);
+ }
+ final IndexedConsensusRequest indexed = (IndexedConsensusRequest) request;
+ final List<IConsensusRequest> copied = new ArrayList<>();
+ for (IConsensusRequest element : indexed.getRequests()) {
+ final int value = element.serializeToByteBuffer().duplicate().getInt();
+ values.add(value);
+ copied.add(new IntegerRequest(value));
+ }
+ if (indexed.getSearchIndex() != DEFAULT_SEARCH_INDEX) {
+ synchronized (localRequests) {
+ localRequests.put(
+ indexed.getSearchIndex(),
+ new IndexedConsensusRequest(indexed.getSearchIndex(), copied));
+ localIndex.set(indexed.getSearchIndex());
+ localRequests.notifyAll();
+ }
+ }
+ return success();
+ }
+
+ @Override
+ public IConsensusRequest deserializeRequest(IConsensusRequest request) {
+ if (!(request instanceof BatchIndexedConsensusRequest)) {
+ return request;
+ }
+ final BatchIndexedConsensusRequest batch = (BatchIndexedConsensusRequest) request;
+ final DeserializedBatchIndexedConsensusRequest result =
+ new DeserializedBatchIndexedConsensusRequest(
+ batch.getStartSyncIndex(),
+ batch.getEndSyncIndex(),
+ batch.getRequests().size(),
+ batch.getSourcePeerId(),
+ 0);
+ batch.getRequests().forEach(result::add);
+ return result;
+ }
+
+ @Override
+ public DataSet read(IConsensusRequest request) {
+ return request instanceof GetConsensusReqReaderPlan ? this : null;
+ }
+
+ @Override
+ public boolean takeSnapshot(File directory) {
+ return true;
+ }
+
+ @Override
+ public boolean loadSnapshot(File directory) {
+ return true;
+ }
+
+ @Override
+ public void setSafelyDeletedSearchIndex(long index) {}
+
+ @Override
+ public long getCurrentSearchIndex() {
+ return localIndex.get();
+ }
+
+ @Override
+ public long getCurrentWALFileVersion() {
+ return 0;
+ }
+
+ @Override
+ public long getTotalSize() {
+ return 0;
+ }
+
+ @Override
+ public ReqIterator getReqIterator(long startIndex) {
+ return new ReqIterator() {
+ private long nextIndex = startIndex;
+
+ @Override
+ public boolean hasNext() {
+ synchronized (localRequests) {
+ return localRequests.containsKey(nextIndex);
+ }
+ }
+
+ @Override
+ public IndexedConsensusRequest next() {
+ synchronized (localRequests) {
+ final IndexedConsensusRequest request = localRequests.get(nextIndex++);
+ return new IndexedConsensusRequest(request.getSearchIndex(), request.getRequests());
+ }
+ }
+
+ @Override
+ public void waitForNextReady() throws InterruptedException {
+ synchronized (localRequests) {
+ while (!hasNext()) {
+ localRequests.wait();
+ }
+ }
+ }
+
+ @Override
+ public void waitForNextReady(long timeout, TimeUnit unit)
+ throws InterruptedException, TimeoutException {
+ final long deadline = System.nanoTime() + unit.toNanos(timeout);
+ synchronized (localRequests) {
+ while (!hasNext()) {
+ final long remaining = deadline - System.nanoTime();
+ if (remaining <= 0) {
+ throw new TimeoutException();
+ }
+ TimeUnit.NANOSECONDS.timedWait(localRequests, remaining);
+ }
+ }
+ }
+
+ @Override
+ public void skipTo(long targetIndex) {
+ nextIndex = targetIndex;
+ }
+ };
+ }
+ }
+}
diff --git a/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RestartDuringRemovePeerTest.java b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RestartDuringRemovePeerTest.java
new file mode 100644
index 00000000000..be868283edc
--- /dev/null
+++ b/iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/procedure/impl/region/RestartDuringRemovePeerTest.java
@@ -0,0 +1,450 @@
+/*
+ * 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.region;
+
+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.TRegionMaintainTaskStatus;
+import org.apache.iotdb.common.rpc.thrift.TRegionReplicaSet;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.commons.cluster.RegionStatus;
+import org.apache.iotdb.confignode.conf.ConfigNodeDescriptor;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
+import org.apache.iotdb.confignode.consensus.request.write.database.DatabaseSchemaPlan;
+import org.apache.iotdb.confignode.consensus.request.write.partition.RemoveRegionLocationPlan;
+import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan;
+import org.apache.iotdb.confignode.manager.ConfigManager;
+import org.apache.iotdb.confignode.manager.load.LoadManager;
+import org.apache.iotdb.confignode.manager.partition.PartitionManager;
+import org.apache.iotdb.confignode.persistence.partition.PartitionInfo;
+import org.apache.iotdb.confignode.procedure.Procedure;
+import org.apache.iotdb.confignode.procedure.env.ConfigNodeProcedureEnv;
+import org.apache.iotdb.confignode.procedure.env.RegionMaintainHandler;
+import org.apache.iotdb.confignode.procedure.state.ProcedureState;
+import org.apache.iotdb.confignode.procedure.state.RemoveRegionPeerState;
+import org.apache.iotdb.confignode.procedure.store.ProcedureType;
+import org.apache.iotdb.confignode.rpc.thrift.TDatabaseSchema;
+import org.apache.iotdb.mpp.rpc.thrift.TRegionMigrateResult;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.io.ByteArrayOutputStream;
+import java.io.DataOutputStream;
+import java.nio.ByteBuffer;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static org.apache.iotdb.consensus.ConsensusFactory.IOT_CONSENSUS;
+import static org.apache.iotdb.consensus.ConsensusFactory.IOT_CONSENSUS_V2;
+import static org.apache.iotdb.consensus.ConsensusFactory.RATIS_CONSENSUS;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyList;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyMap;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/** Regression coverage for the stale restart-list window and bounded CN reconciliation. */
+public class RestartDuringRemovePeerTest {
+ private final TConsensusGroupId regionId =
+ new TConsensusGroupId(TConsensusGroupType.DataRegion, 6);
+ private final TDataNodeLocation restartingNode = location(3, "11.101.10.8");
+ private final TDataNodeLocation coordinator = location(5, "11.101.10.9");
+ private final TDataNodeLocation removedNode = location(6, "11.101.10.10");
+ private final List<TDataNodeLocation> finalPeers = Arrays.asList(restartingNode, coordinator);
+ private PartitionInfo partitionInfo;
+ private PartitionManager partitionManager;
+ private LoadManager loadManager;
+ private RegionMaintainHandler handler;
+ private ConfigNodeProcedureEnv env;
+ private String previousProtocol;
+
+ @Before
+ public void setUp() throws Exception {
+ previousProtocol =
+ ConfigNodeDescriptor.getInstance().getConf().getDataRegionConsensusProtocolClass();
+ ConfigNodeDescriptor.getInstance().getConf().setDataRegionConsensusProtocolClass(IOT_CONSENSUS);
+ partitionInfo = new PartitionInfo();
+ partitionInfo.createDatabase(
+ new DatabaseSchemaPlan(
+ ConfigPhysicalPlanType.CreateDatabase, new TDatabaseSchema("root.restart")));
+ CreateRegionGroupsPlan groups = new CreateRegionGroupsPlan();
+ groups.addRegionGroup(
+ "root.restart",
+ new TRegionReplicaSet(
+ regionId, new ArrayList<>(Arrays.asList(restartingNode, coordinator, removedNode))));
+ partitionInfo.createRegionGroups(groups);
+
+ ConfigManager configManager = Mockito.mock(ConfigManager.class);
+ partitionManager = Mockito.mock(PartitionManager.class);
+ loadManager = Mockito.mock(LoadManager.class, Mockito.RETURNS_DEEP_STUBS);
+ when(configManager.getPartitionManager()).thenReturn(partitionManager);
+ when(configManager.getLoadManager()).thenReturn(loadManager);
+ when(partitionManager.getAllReplicaSets())
+ .thenAnswer(ignored -> partitionInfo.getAllReplicaSets());
+ when(partitionManager.removeRegionLocation(any()))
+ .thenAnswer(call -> partitionInfo.removeRegionLocation(call.getArgument(0)));
+ handler = Mockito.spy(new RegionMaintainHandler(configManager));
+ env = Mockito.mock(ConfigNodeProcedureEnv.class);
+ when(env.getRegionMaintainHandler()).thenReturn(handler);
+ doNothing().when(handler).forceUpdateRegionCache(any(), any(), any());
+ doNothing().when(handler).transferRegionLeader(any(), any(), any());
+ doNothing().when(handler).dropConsensusPipesForRemovePeer(any(), any());
+ doReturn(success()).when(handler).submitRemoveRegionPeerTask(anyLong(), any(), any(), any());
+ doReturn(success()).when(handler).submitDeleteOldRegionPeerTask(anyLong(), any(), any());
+ doReturn(new TRegionMigrateResult(TRegionMaintainTaskStatus.SUCCESS))
+ .when(handler)
+ .waitTaskFinish(anyLong(), any(TDataNodeLocation.class));
+ doReturn(successfulResponses()).when(handler).resetPeerList(any(), anyList(), anyMap());
+ }
+
+ @After
+ public void restoreProtocol() {
+ ConfigNodeDescriptor.getInstance()
+ .getConf()
+ .setDataRegionConsensusProtocolClass(previousProtocol);
+ }
+
+ @Test
+ public void staleRestartSnapshotIsFollowedByFinalReconciliation() throws Exception {
+ AccessibleRemoveProcedure procedure = newProcedure();
+ procedure.doExecute(env);
+ procedure.doExecute(env);
+ List<TRegionReplicaSet> startupSnapshot =
+ partitionInfo.getAllReplicaSets(restartingNode.getDataNodeId());
+ assertTrue(startupSnapshot.get(0).getDataNodeLocations().contains(removedNode));
+ verify(handler, Mockito.atLeastOnce())
+ .forceUpdateRegionCache(regionId, removedNode, RegionStatus.Removing);
+ verify(handler, never()).removeRegionLocation(regionId, removedNode);
+ procedure.doExecute(env);
+ procedure.doExecute(env);
+
+ doAnswer(
+ call -> {
+ assertEquals(
+ finalPeers, partitionInfo.getAllReplicaSets().get(0).getDataNodeLocations());
+ assertEquals(finalPeers, call.getArgument(1));
+ Map<Integer, TDataNodeLocation> recipients = call.getArgument(2);
+ assertEquals(2, recipients.size());
+ assertEquals(restartingNode, recipients.get(restartingNode.getDataNodeId()));
+ assertEquals(coordinator, recipients.get(coordinator.getDataNodeId()));
+ assertFalse(recipients.containsKey(removedNode.getDataNodeId()));
+ return successfulResponses();
+ })
+ .when(handler)
+ .resetPeerList(eq(regionId), anyList(), anyMap());
+
+ assertEquals(1, procedure.doExecute(env).length);
+ assertEquals(RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST, procedure.currentState());
+ verify(handler, never()).resetPeerList(any(), anyList(), anyMap());
+ assertEquals(0, procedure.doExecute(env).length);
+ assertFalse(procedure.isWaiting());
+ // Correcting future startup responses cannot change a snapshot already returned to a DN.
+ assertTrue(startupSnapshot.get(0).getDataNodeLocations().contains(removedNode));
+ verify(handler).resetPeerList(eq(regionId), eq(finalPeers), anyMap());
+ }
+
+ @Test
+ public void metadataFailureRetriesBeforeCorrectingPeers() throws Exception {
+ AccessibleRemoveProcedure procedure = beforeMetadataRemoval();
+ doReturn(new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode()))
+ .when(partitionManager)
+ .removeRegionLocation(any());
+ assertEquals(1, procedure.doExecute(env).length);
+ assertEquals(ProcedureState.RUNNABLE, procedure.getState());
+ assertEquals(RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE, procedure.currentState());
+ assertTrue(
+ partitionInfo.getAllReplicaSets().get(0).getDataNodeLocations().contains(removedNode));
+ verify(handler, never()).resetPeerList(any(), anyList(), anyMap());
+ verify(loadManager).removeRegionCache(regionId, removedNode.getDataNodeId());
+ verify(loadManager.getRouteBalancer()).balanceRegionLeaderAndPriority();
+
+ doAnswer(call -> partitionInfo.removeRegionLocation(call.getArgument(0)))
+ .when(partitionManager)
+ .removeRegionLocation(any());
+ assertEquals(1, procedure.doExecute(env).length);
+ assertEquals(0, procedure.doExecute(env).length);
+ assertFalse(procedure.isWaiting());
+ verify(handler).resetPeerList(eq(regionId), eq(finalPeers), anyMap());
+ verify(loadManager, times(2)).removeRegionCache(regionId, removedNode.getDataNodeId());
+ verify(loadManager.getRouteBalancer(), times(2)).balanceRegionLeaderAndPriority();
+ }
+
+ @Test
+ public void v2MetadataFailureRetriesBeforeCompletingWithoutResettingPeers() throws Exception {
+ ConfigNodeDescriptor.getInstance()
+ .getConf()
+ .setDataRegionConsensusProtocolClass(IOT_CONSENSUS_V2);
+ AccessibleRemoveProcedure procedure = beforeMetadataRemoval();
+ doReturn(new TSStatus(TSStatusCode.MIGRATE_REGION_ERROR.getStatusCode()))
+ .doAnswer(call -> partitionInfo.removeRegionLocation(call.getArgument(0)))
+ .when(partitionManager)
+ .removeRegionLocation(any());
+
+ assertEquals(1, procedure.doExecute(env).length);
+ assertEquals(RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE, procedure.currentState());
+ assertFalse(procedure.isFailed());
+ assertTrue(
+ partitionInfo.getAllReplicaSets().get(0).getDataNodeLocations().contains(removedNode));
+
+ assertEquals(0, procedure.doExecute(env).length);
+ assertFalse(procedure.isFailed());
+ assertEquals(finalPeers, partitionInfo.getAllReplicaSets().get(0).getDataNodeLocations());
+ verify(handler, times(2)).removeRegionLocation(regionId, removedNode);
+ verify(handler, never()).resetPeerList(any(), anyList(), anyMap());
+ }
+
+ @Test
+ public void missingAndNotInitializedResponsesCanRetryAfterRecovery() throws Exception {
+ AccessibleRemoveProcedure procedure = beforePeerListReset();
+ doReturn(Collections.singletonMap(coordinator.getDataNodeId(), success()))
+ .when(handler)
+ .resetPeerList(any(), anyList(), anyMap());
+ assertEquals(1, procedure.doExecute(env).length);
+ assertEquals(ProcedureState.RUNNABLE, procedure.getState());
+ assertEquals(finalPeers, partitionInfo.getAllReplicaSets().get(0).getDataNodeLocations());
+
+ AccessibleRemoveProcedure restored = restore(serialize(procedure));
+ assertEquals(ProcedureState.RUNNABLE, restored.getState());
+ assertEquals(RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST, restored.currentState());
+ Map<Integer, TSStatus> notReady = successfulResponses();
+ notReady.put(
+ restartingNode.getDataNodeId(),
+ new TSStatus(TSStatusCode.CONSENSUS_NOT_INITIALIZED.getStatusCode()));
+ doReturn(notReady).when(handler).resetPeerList(any(), anyList(), anyMap());
+ assertEquals(1, restored.doExecute(env).length);
+ assertEquals(ProcedureState.RUNNABLE, restored.getState());
+
+ doReturn(successfulResponses()).when(handler).resetPeerList(any(), anyList(), anyMap());
+ assertEquals(0, restored.doExecute(env).length);
+ assertFalse(restored.isWaiting());
+ verify(handler, times(3)).resetPeerList(eq(regionId), eq(finalPeers), anyMap());
+ verify(handler).removeRegionLocation(regionId, removedNode);
+ }
+
+ @Test
+ public void exceptionDuringCorrectionIsRetriedInsteadOfCompleting() throws Exception {
+ AccessibleRemoveProcedure procedure = beforePeerListReset();
+ doThrow(new IllegalStateException("injected RPC failure"))
+ .when(handler)
+ .resetPeerList(any(), anyList(), anyMap());
+ assertEquals(1, procedure.doExecute(env).length);
+ assertEquals(ProcedureState.RUNNABLE, procedure.getState());
+ assertFalse(procedure.isFailed());
+ doReturn(successfulResponses()).when(handler).resetPeerList(any(), anyList(), anyMap());
+ assertEquals(0, procedure.doExecute(env).length);
+ assertFalse(procedure.isWaiting());
+ }
+
+ @Test
+ public void replayAfterSuccessfulCorrectionIsIdempotent() throws Exception {
+ AccessibleRemoveProcedure procedure = beforePeerListReset();
+ byte[] checkpoint = serialize(procedure);
+ assertEquals(0, procedure.doExecute(env).length);
+ // Crash after successful RPCs, before persisting procedure completion.
+ AccessibleRemoveProcedure restored = restore(checkpoint);
+ assertEquals(0, restored.doExecute(env).length);
+ assertEquals(finalPeers, partitionInfo.getAllReplicaSets().get(0).getDataNodeLocations());
+ verify(handler, times(2)).resetPeerList(eq(regionId), eq(finalPeers), anyMap());
+ }
+
+ @Test
+ public void offlineReplicaFailsAfterThreeRetriesInsteadOfWaitingForever() throws Exception {
+ AccessibleRemoveProcedure procedure = beforePeerListReset();
+ doReturn(Collections.singletonMap(coordinator.getDataNodeId(), success()))
+ .when(handler)
+ .resetPeerList(any(), anyList(), anyMap());
+ for (int retry = 0; retry < 3; retry++) {
+ assertEquals(1, procedure.doExecute(env).length);
+ assertEquals(ProcedureState.RUNNABLE, procedure.getState());
+ assertFalse(procedure.isFailed());
+ }
+ assertEquals(0, procedure.doExecute(env).length);
+ assertTrue(procedure.isFailed());
+ assertTrue(procedure.getException().getMessage().contains("4"));
+ verify(handler, times(4)).resetPeerList(eq(regionId), eq(finalPeers), anyMap());
+ verify(handler).removeRegionLocation(regionId, removedNode);
+ }
+
+ @Test
+ public void metadataFailureAlsoStopsAtRetryLimitWithoutResettingPeers() throws Exception {
+ AccessibleRemoveProcedure procedure = beforeMetadataRemoval();
+ doReturn(null).when(partitionManager).removeRegionLocation(any());
+ for (int retry = 0; retry < 3; retry++) {
+ assertEquals(1, procedure.doExecute(env).length);
+ assertFalse(procedure.isWaiting());
+ }
+ assertEquals(0, procedure.doExecute(env).length);
+ assertTrue(procedure.isFailed());
+ verify(handler, times(4)).removeRegionLocation(regionId, removedNode);
+ verify(handler, never()).resetPeerList(any(), anyList(), anyMap());
+ }
+
+ @Test
+ public void interruptedRetryRetainsItsStepForRecovery() throws Exception {
+ AccessibleRemoveProcedure procedure = beforePeerListReset();
+ doReturn(Collections.emptyMap()).when(handler).resetPeerList(any(), anyList(), anyMap());
+ try {
+ Thread.currentThread().interrupt();
+ assertEquals(1, procedure.doExecute(env).length);
+ assertTrue(Thread.currentThread().isInterrupted());
+ assertFalse(procedure.isFailed());
+ } finally {
+ Thread.interrupted();
+ }
+ AccessibleRemoveProcedure restored = restore(serialize(procedure));
+ assertEquals(RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST, restored.currentState());
+ doReturn(successfulResponses()).when(handler).resetPeerList(any(), anyList(), anyMap());
+ assertEquals(0, restored.doExecute(env).length);
+ }
+
+ @Test
+ public void correctionRejectsStaleMetadataAndNullResponsesButAcceptsDeletedRegion() {
+ assertFalse(handler.resetPeerListAfterRemoveForIoTConsensusV1(regionId, removedNode));
+ verify(handler, never()).resetPeerList(any(), anyList(), anyMap());
+ partitionInfo.removeRegionLocation(new RemoveRegionLocationPlan(regionId, removedNode));
+ doReturn(null).when(handler).resetPeerList(any(), anyList(), anyMap());
+ assertFalse(handler.resetPeerListAfterRemoveForIoTConsensusV1(regionId, removedNode));
+ when(partitionManager.getAllReplicaSets()).thenReturn(Collections.emptyList());
+ assertTrue(handler.resetPeerListAfterRemoveForIoTConsensusV1(regionId, removedNode));
+ }
+
+ @Test
+ public void correctionIsLimitedToIoTConsensusV1DataRegions() throws Exception {
+ // New states must be appended so existing persisted state IDs remain valid.
+ assertEquals(3, RemoveRegionPeerState.REMOVE_REGION_LOCATION_CACHE.ordinal());
+ assertEquals(4, RemoveRegionPeerState.DROP_CONSENSUS_PIPES.ordinal());
+ assertEquals(5, RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST.ordinal());
+ assertTrue(handler.isIoTConsensusV1DataRegion(regionId));
+ assertFalse(
+ handler.isIoTConsensusV1DataRegion(
+ new TConsensusGroupId(TConsensusGroupType.SchemaRegion, 6)));
+ for (String protocol : Arrays.asList(IOT_CONSENSUS_V2, RATIS_CONSENSUS)) {
+ ConfigNodeDescriptor.getInstance().getConf().setDataRegionConsensusProtocolClass(protocol);
+ assertFalse(handler.isIoTConsensusV1DataRegion(regionId));
+ AccessibleRemoveProcedure procedure = beforeMetadataRemoval();
+ assertEquals(0, procedure.doExecute(env).length);
+ }
+ verify(handler, never()).resetPeerList(any(), anyList(), anyMap());
+ }
+
+ private AccessibleRemoveProcedure newProcedure() {
+ AccessibleRemoveProcedure procedure =
+ new AccessibleRemoveProcedure(regionId, coordinator, removedNode);
+ procedure.setProcId(1);
+ procedure.setState(ProcedureState.RUNNABLE);
+ return procedure;
+ }
+
+ private AccessibleRemoveProcedure beforeMetadataRemoval() throws Exception {
+ AccessibleRemoveProcedure procedure = newProcedure();
+ for (int i = 0; i < 4; i++) {
+ assertEquals(1, procedure.doExecute(env).length);
+ }
+ return procedure;
+ }
+
+ private AccessibleRemoveProcedure beforePeerListReset() throws Exception {
+ AccessibleRemoveProcedure procedure = beforeMetadataRemoval();
+ assertEquals(1, procedure.doExecute(env).length);
+ assertEquals(RemoveRegionPeerState.RESET_IOT_CONSENSUS_V1_PEER_LIST, procedure.currentState());
+ return procedure;
+ }
+
+ private static byte[] serialize(AccessibleRemoveProcedure procedure) throws Exception {
+ ByteArrayOutputStream bytes = new ByteArrayOutputStream();
+ procedure.serialize(new DataOutputStream(bytes));
+ return bytes.toByteArray();
+ }
+
+ private static AccessibleRemoveProcedure restore(byte[] bytes) throws Exception {
+ ByteBuffer buffer = ByteBuffer.wrap(bytes);
+ assertEquals(ProcedureType.REMOVE_REGION_PEER_PROCEDURE.getTypeCode(), buffer.getShort());
+ AccessibleRemoveProcedure restored = new AccessibleRemoveProcedure();
+ restored.deserialize(buffer);
+ return restored;
+ }
+
+ private Map<Integer, TSStatus> successfulResponses() {
+ Map<Integer, TSStatus> responses = new HashMap<>();
+ for (TDataNodeLocation peer : finalPeers) {
+ responses.put(peer.getDataNodeId(), success());
+ }
+ return responses;
+ }
+
+ private static TSStatus success() {
+ return new TSStatus(TSStatusCode.SUCCESS_STATUS.getStatusCode());
+ }
+
+ private static class AccessibleRemoveProcedure extends RemoveRegionPeerProcedure {
+ private AccessibleRemoveProcedure() {}
+
+ private AccessibleRemoveProcedure(
+ TConsensusGroupId group, TDataNodeLocation coordinator, TDataNodeLocation target) {
+ super(group, coordinator, target);
+ }
+
+ @Override
+ public Procedure<ConfigNodeProcedureEnv>[] doExecute(ConfigNodeProcedureEnv env)
+ throws InterruptedException {
+ return super.doExecute(env);
+ }
+
+ @Override
+ public void setState(ProcedureState state) {
+ super.setState(state);
+ }
+
+ private RemoveRegionPeerState currentState() {
+ return getCurrentState();
+ }
+ }
+
+ private static TDataNodeLocation location(int id, String ip) {
+ return new TDataNodeLocation(
+ id,
+ new TEndPoint(ip, 6667),
+ new TEndPoint(ip, 10730),
+ new TEndPoint(ip, 10740),
+ new TEndPoint(ip, 10750),
+ new TEndPoint(ip, 10760));
+ }
+}
diff --git a/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/StalePeerListRestartReproductionTest.java b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/StalePeerListRestartReproductionTest.java
new file mode 100644
index 00000000000..4400d6665e7
--- /dev/null
+++ b/iotdb-core/consensus/src/test/java/org/apache/iotdb/consensus/iot/StalePeerListRestartReproductionTest.java
@@ -0,0 +1,272 @@
+/*
+ * 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.consensus.iot;
+
+import org.apache.iotdb.common.rpc.thrift.TConsensusGroupType;
+import org.apache.iotdb.common.rpc.thrift.TEndPoint;
+import org.apache.iotdb.commons.consensus.DataRegionId;
+import org.apache.iotdb.consensus.common.Peer;
+import org.apache.iotdb.consensus.config.ConsensusConfig;
+import org.apache.iotdb.consensus.config.IoTConsensusConfig;
+import org.apache.iotdb.consensus.iot.logdispatcher.LogDispatcher;
+import org.apache.iotdb.consensus.iot.logdispatcher.LogDispatcher.LogDispatcherThread;
+import org.apache.iotdb.consensus.iot.util.TestEntry;
+import org.apache.iotdb.consensus.iot.util.TestStateMachine;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.TemporaryFolder;
+
+import java.io.File;
+import java.lang.reflect.Field;
+import java.net.ServerSocket;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.function.BooleanSupplier;
+import java.util.stream.Collectors;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotSame;
+import static org.junit.Assert.assertTrue;
+
+/** Diagnostic reproduction: assertions describe the current faulty behavior, not a fix. */
+public class StalePeerListRestartReproductionTest {
+
+ private static final DataRegionId GROUP = new DataRegionId(91);
+ private static final int INITIAL_WRITES = 8;
+ private static final int WRITES_AFTER_RESTART = 8;
+
+ @Rule public final TemporaryFolder temporaryFolder = new TemporaryFolder();
+
+ @Test
+ public void removedPeerIsRecreatedByStaleStartupListAndPinsRealSyncLag() throws Exception {
+ final List<Peer> peers = new ArrayList<>();
+ final List<File> storageDirectories = new ArrayList<>();
+ final List<TestStateMachine> stateMachines = new ArrayList<>();
+ final IoTConsensus[] servers = new IoTConsensus[3];
+ for (int index = 0; index < servers.length; index++) {
+ peers.add(new Peer(GROUP, index + 1, new TEndPoint("127.0.0.1", unusedPort())));
+ storageDirectories.add(temporaryFolder.newFolder("peer-" + index));
+ stateMachines.add(new TestStateMachine());
+ }
+ final Peer writer = peers.get(0);
+ final Peer removed = peers.get(1);
+ final Peer survivor = peers.get(2);
+ final List<Peer> finalPeers = Arrays.asList(writer, survivor);
+ try {
+ for (int index = 0; index < servers.length; index++) {
+ servers[index] =
+ newServer(peers.get(index), storageDirectories.get(index), stateMachines.get(index));
+ servers[index].start();
+ servers[index].createLocalPeer(GROUP, peers);
+ }
+ final IoTConsensus originalWriter = servers[0];
+ write(originalWriter, writer, INITIAL_WRITES);
+ await("initial replicas catch up", () -> originalWriter.getImpl(GROUP).getSyncLag() == 0);
+ final LogDispatcherThread originalRemovedThread =
+ dispatcher(originalWriter.getImpl(GROUP), removed);
+ assertEquals(INITIAL_WRITES, originalRemovedThread.getCurrentSyncIndex());
+ printState("BEFORE_REMOVE", originalWriter.getImpl(GROUP));
+
+ // This is the real remove RPC broadcast used by removeRemotePeer. A and C acknowledge it.
+ originalWriter.getImpl(GROUP).notifyPeersToRemoveSyncLogChannel(removed);
+ assertEquals(finalPeers, originalWriter.getImpl(GROUP).getConfiguration());
+ assertEquals(finalPeers, servers[2].getImpl(GROUP).getConfiguration());
+ assertEquals(
+ Collections.singletonList(survivor), dispatcherPeers(originalWriter.getImpl(GROUP)));
+ assertTrue(originalRemovedThread.isStopped());
+ assertEquals(0, runFinished(originalRemovedThread).getCount());
+ final File regionDirectory = new File(originalWriter.getRegionDirFromConsensusGroupId(GROUP));
+ assertEquals(0, regionDirectory.listFiles((dir, name) -> name.startsWith("2-")).length);
+ printState("REMOVE_ACKNOWLEDGED", originalWriter.getImpl(GROUP));
+
+ // Keep B's RPC endpoint alive: it now returns INTERNAL_SERVER_ERROR (region does not exist).
+ servers[1].deleteLocalPeer(GROUP);
+ originalWriter.stop();
+ servers[0] = null;
+
+ // Model a restart registration response obtained before ConfigNode commits removal.
+ final IoTConsensus restartedWriter =
+ newServer(writer, storageDirectories.get(0), stateMachines.get(0));
+ servers[0] = restartedWriter;
+ restartedWriter.recordCorrectPeerListBeforeStarting(Collections.singletonMap(GROUP, peers));
+ restartedWriter.start();
+ final IoTConsensusServerImpl recovered = restartedWriter.getImpl(GROUP);
+ assertEquals(peers, recovered.getConfiguration());
+ assertEquals(Arrays.asList(removed, survivor), dispatcherPeers(recovered));
+ final LogDispatcherThread resurrectedThread = dispatcher(recovered, removed);
+ assertNotSame(originalRemovedThread, resurrectedThread);
+ assertFalse(resurrectedThread.isStopped());
+ assertEquals(1, runFinished(resurrectedThread).getCount());
+ final long removedPeerIndex = resurrectedThread.getCurrentSyncIndex();
+ assertEquals(INITIAL_WRITES, removedPeerIndex);
+ assertEquals(1, regionDirectory.listFiles((dir, name) -> name.startsWith("2-")).length);
+ printState("STALE_STARTUP_RECREATED_REMOVED_PEER", recovered);
+
+ write(restartedWriter, writer, WRITES_AFTER_RESTART);
+ final long expectedSearchIndex = INITIAL_WRITES + WRITES_AFTER_RESTART;
+ final LogDispatcherThread healthyThread = dispatcher(recovered, survivor);
+ await(
+ "surviving replica catches up",
+ () -> healthyThread.getCurrentSyncIndex() == expectedSearchIndex);
+ assertEquals(expectedSearchIndex, recovered.getSearchIndex());
+ assertEquals(expectedSearchIndex, stateMachines.get(2).getRequestSet().size());
+ for (int sample = 0; sample < 5; sample++) {
+ assertEquals(removedPeerIndex, resurrectedThread.getCurrentSyncIndex());
+ assertEquals(WRITES_AFTER_RESTART, recovered.getSyncLag());
+ assertFalse(resurrectedThread.isStopped());
+ assertEquals(1, runFinished(resurrectedThread).getCount());
+ printState("LAG_PINNED_SAMPLE_" + sample, recovered);
+ Thread.sleep(200);
+ }
+
+ // Later process shutdown changes the transport error, but does not repair membership or lag.
+ servers[1].stop();
+ servers[1] = null;
+ Thread.sleep(300);
+ assertEquals(removedPeerIndex, resurrectedThread.getCurrentSyncIndex());
+ assertEquals(WRITES_AFTER_RESTART, recovered.getSyncLag());
+ printState("REMOVED_ENDPOINT_DOWN_LAG_STILL_PINNED", recovered);
+
+ // Control: explicit membership reconciliation removes the new thread and immediately unpins
+ // lag.
+ restartedWriter.resetPeerList(GROUP, finalPeers);
+ assertEquals(Collections.singletonList(survivor), dispatcherPeers(recovered));
+ assertTrue(resurrectedThread.isStopped());
+ assertEquals(0, recovered.getSyncLag());
+ printState("EXPLICIT_RECONCILIATION_UNPINS_LAG", recovered);
+
+ restartedWriter.stop();
+ servers[0] = null;
+ final IoTConsensus freshListWriter =
+ newServer(writer, storageDirectories.get(0), stateMachines.get(0));
+ servers[0] = freshListWriter;
+ freshListWriter.recordCorrectPeerListBeforeStarting(
+ Collections.singletonMap(GROUP, finalPeers));
+ freshListWriter.start();
+ assertEquals(finalPeers, freshListWriter.getImpl(GROUP).getConfiguration());
+ assertEquals(
+ Collections.singletonList(survivor), dispatcherPeers(freshListWriter.getImpl(GROUP)));
+ write(freshListWriter, writer, 1);
+ await("fresh-list restart converges", () -> freshListWriter.getImpl(GROUP).getSyncLag() == 0);
+ assertEquals(expectedSearchIndex + 1, freshListWriter.getImpl(GROUP).getSearchIndex());
+ printState("FRESH_STARTUP_LIST_CONVERGES", freshListWriter.getImpl(GROUP));
+ } finally {
+ Arrays.stream(servers)
+ .filter(server -> server != null)
+ .parallel()
+ .forEach(IoTConsensus::stop);
+ }
+ }
+
+ private static IoTConsensus newServer(Peer peer, File directory, TestStateMachine stateMachine) {
+ return new IoTConsensus(
+ ConsensusConfig.newBuilder()
+ .setThisNodeId(peer.getNodeId())
+ .setThisNode(peer.getEndpoint())
+ .setStorageDir(directory.getAbsolutePath())
+ .setConsensusGroupType(TConsensusGroupType.DataRegion)
+ .setIoTConsensusConfig(
+ IoTConsensusConfig.newBuilder()
+ .setRpc(
+ IoTConsensusConfig.RPC.newBuilder().setConnectionTimeoutInMs(500).build())
+ .setReplication(
+ IoTConsensusConfig.Replication.newBuilder()
+ .setCheckpointGap(1)
+ .setBasicRetryWaitTimeMs(20)
+ .setMaxRetryWaitTimeMs(100)
+ .build())
+ .build())
+ .build(),
+ ignored -> stateMachine);
+ }
+
+ private static int unusedPort() throws Exception {
+ try (ServerSocket socket = new ServerSocket(0)) {
+ return socket.getLocalPort();
+ }
+ }
+
+ private static void write(IoTConsensus server, Peer writer, int count) throws Exception {
+ final int previousCount = (int) server.getImpl(GROUP).getSearchIndex();
+ for (int index = 1; index <= count; index++) {
+ assertEquals(
+ TSStatusCode.SUCCESS_STATUS.getStatusCode(),
+ server.write(GROUP, new TestEntry(previousCount + index, writer)).getCode());
+ }
+ }
+
+ private static void await(String condition, BooleanSupplier satisfied) throws Exception {
+ final long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(20);
+ while (!satisfied.getAsBoolean() && System.nanoTime() < deadline) {
+ Thread.sleep(20);
+ }
+ assertTrue(condition, satisfied.getAsBoolean());
+ }
+
+ @SuppressWarnings("unchecked")
+ private static List<LogDispatcherThread> dispatchers(IoTConsensusServerImpl server)
+ throws Exception {
+ final LogDispatcher dispatcher = server.getLogDispatcher();
+ final Field field = LogDispatcher.class.getDeclaredField("threads");
+ field.setAccessible(true);
+ synchronized (dispatcher) {
+ return new ArrayList<>((List<LogDispatcherThread>) field.get(dispatcher));
+ }
+ }
+
+ private static List<Peer> dispatcherPeers(IoTConsensusServerImpl server) throws Exception {
+ return dispatchers(server).stream()
+ .map(LogDispatcherThread::getPeer)
+ .collect(Collectors.toList());
+ }
+
+ private static LogDispatcherThread dispatcher(IoTConsensusServerImpl server, Peer peer)
+ throws Exception {
+ return dispatchers(server).stream()
+ .filter(thread -> thread.getPeer().equals(peer))
+ .findFirst()
+ .get();
+ }
+
+ private static CountDownLatch runFinished(LogDispatcherThread thread) throws Exception {
+ final Field field = LogDispatcherThread.class.getDeclaredField("runFinished");
+ field.setAccessible(true);
+ return (CountDownLatch) field.get(thread);
+ }
+
+ private static void printState(String stage, IoTConsensusServerImpl server) throws Exception {
+ System.out.printf(
+ "REPRO %s searchIndex=%d minSyncIndex=%d syncLag=%d dispatchers=%s%n",
+ stage,
+ server.getSearchIndex(),
+ server.getMinSyncIndex(),
+ server.getSyncLag(),
+ dispatchers(server).stream()
+ .map(thread -> thread.getPeer().getNodeId() + ":" + thread.getCurrentSyncIndex())
+ .collect(Collectors.toList()));
+ }
+}