Fix stale IoTConsensus peers after restart during removal

integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/batch/IoTDBRegionMigrateConfigNodeCrashIoTV2BatchIT.java

@@ -173,6 +173,8 @@ public void cnCrashTest() throws Exception {
             .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);

integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/iotv2/stream/IoTDBRegionMigrateConfigNodeCrashIoTV2StreamIT.java

@@ -187,6 +187,8 @@ public void cnCrashTest() throws Exception {
             .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);

integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/daily/ratis/IoTDBRegionMigrateConfigNodeCrashForRatisIT.java

@@ -183,6 +183,8 @@ public void cnCrashTest() throws Exception {
             .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);

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";

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";

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/env/RegionMaintainHandler.java

@@ -360,6 +360,58 @@ public Map<Integer, TSStatus> resetPeerList(
     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);
@@ -449,7 +501,7 @@ public void forceUpdateRegionCache(
         .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);
@@ -460,6 +512,7 @@ public void removeRegionLocation(
         status);
     configManager.getLoadManager().removeRegionCache(regionId, deprecatedLocation.getDataNodeId());
     configManager.getLoadManager().getRouteBalancer().balanceRegionLeaderAndPriority();
+    return status;
   }
 
   /**

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/node/RemoveDataNodesProcedure.java

@@ -54,6 +54,7 @@
 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 @@ protected ProcedureLockState acquireLock(ConfigNodeProcedureEnv configNodeProced
     }
   }
 
+  @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 @@ private void checkRegionStatusAndStopDataNode(ConfigNodeProcedureEnv env) throws
 
   @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) {

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.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 @@ protected Flow executeFromState(ConfigNodeProcedureEnv env, RemoveRegionPeerStat
           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 {}

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,
 }