Support safe removal of multiple cluster nodes

integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/IoTDBRegionOperationReliabilityITFramework.java

@@ -620,6 +620,21 @@ public static Map<Integer, Set<Integer>> getDataRegionMap(Statement statement) t
     return regionMap;
   }
 
+  public static Map<Integer, Set<Integer>> getSchemaRegionMap(Statement statement)
+      throws Exception {
+    ResultSet showRegionsResult = statement.executeQuery(SHOW_REGIONS);
+    Map<Integer, Set<Integer>> regionMap = new HashMap<>();
+    while (showRegionsResult.next()) {
+      if (String.valueOf(TConsensusGroupType.SchemaRegion)
+          .equals(showRegionsResult.getString(ColumnHeaderConstant.TYPE))) {
+        int regionId = showRegionsResult.getInt(ColumnHeaderConstant.REGION_ID);
+        int dataNodeId = showRegionsResult.getInt(ColumnHeaderConstant.DATA_NODE_ID);
+        regionMap.computeIfAbsent(regionId, id -> new HashSet<>()).add(dataNodeId);
+      }
+    }
+    return regionMap;
+  }
+
   public static Map<Integer, Pair<Integer, Set<Integer>>> getDataRegionMapWithLeader(
       Statement statement) throws Exception {
     ResultSet showRegionsResult = statement.executeQuery(SHOW_REGIONS);

integration-test/src/test/java/org/apache/iotdb/confignode/it/removeconfignode/IoTDBRemoveConfigNodeITFramework.java

@@ -152,6 +152,35 @@ public void testRemoveConfigNode(
     }
   }
 
+  public void testRejectRemoveConfigNodeFromTwoReplicaGroup(final SQLModel model) throws Exception {
+    EnvFactory.getEnv().initClusterEnvironment(2, 1);
+
+    try (final Connection connection = makeItCloseQuietly(getConnectionWithSQLType(model));
+        final Statement statement = makeItCloseQuietly(connection.createStatement())) {
+      final ResultSet result = statement.executeQuery(SHOW_CONFIGNODES);
+      final Set<Integer> allConfigNodeIds = new HashSet<>();
+      while (result.next()) {
+        allConfigNodeIds.add(result.getInt(ColumnHeaderConstant.NODE_ID));
+      }
+      Assert.assertEquals(2, allConfigNodeIds.size());
+
+      final int removeConfigNodeId = allConfigNodeIds.iterator().next();
+      try {
+        statement.execute(generateRemoveString(removeConfigNodeId));
+        Assert.fail();
+      } catch (IoTDBSQLException expected) {
+        // Removing one peer from a two-replica Ratis group would lose the majority.
+      }
+
+      final ResultSet remainingResult = statement.executeQuery(SHOW_CONFIGNODES);
+      int remainingConfigNodeCount = 0;
+      while (remainingResult.next()) {
+        remainingConfigNodeCount++;
+      }
+      Assert.assertEquals(2, remainingConfigNodeCount);
+    }
+  }
+
   private static void awaitUntilSuccess(Statement statement, int removeConfigNodeId) {
     AtomicReference<Set<Integer>> lastTimeConfigNodes = new AtomicReference<>();
     AtomicReference<Exception> lastException = new AtomicReference<>();

integration-test/src/test/java/org/apache/iotdb/confignode/it/removeconfignode/IoTDBRemoveConfigNodeNormalIT.java

@@ -39,4 +39,9 @@ public void test3C1DUseTreeSQL() throws Exception {
   public void test3C1DUseTableSQL() throws Exception {
     testRemoveConfigNode(1, 1, 3, 1, 2, SQLModel.TABLE_MODEL_SQL);
   }
+
+  @Test
+  public void testRejectRemovingOneOfTwoConfigNodes() throws Exception {
+    testRejectRemoveConfigNodeFromTwoReplicaGroup(SQLModel.TREE_MODEL_SQL);
+  }
 }

integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeNormalIT.java

@@ -58,11 +58,13 @@
 
 import static org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework.getAllRegionMap;
 import static org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework.getDataRegionMap;
+import static org.apache.iotdb.confignode.it.regionmigration.IoTDBRegionOperationReliabilityITFramework.getSchemaRegionMap;
 import static org.apache.iotdb.confignode.it.removedatanode.IoTDBRemoveDataNodeUtils.awaitUntilSuccess;
 import static org.apache.iotdb.confignode.it.removedatanode.IoTDBRemoveDataNodeUtils.generateRemoveString;
 import static org.apache.iotdb.confignode.it.removedatanode.IoTDBRemoveDataNodeUtils.getConnectionWithSQLType;
 import static org.apache.iotdb.confignode.it.removedatanode.IoTDBRemoveDataNodeUtils.restartDataNodes;
 import static org.apache.iotdb.confignode.it.removedatanode.IoTDBRemoveDataNodeUtils.selectRemoveDataNodes;
+import static org.apache.iotdb.confignode.it.removedatanode.IoTDBRemoveDataNodeUtils.selectRemoveDataNodesSharingWeakRegionSafely;
 import static org.apache.iotdb.confignode.it.removedatanode.IoTDBRemoveDataNodeUtils.selectRemoveDataNodesWithoutRegionConflict;
 import static org.apache.iotdb.util.MagicUtils.makeItCloseQuietly;
 
@@ -158,10 +160,11 @@ public void success1C4DIoTTestUseTableSQL() throws Exception {
   }
 
   @Test
-  public void success1C5DRemoveTwoDataNodesUseSQL() throws Exception {
-    // Setup 1C5D, and remove 2D in a single "remove datanode a, b" statement; 3 DataNodes remain
-    // which is enough to keep both the data (factor 2) and schema (factor 3) replicas.
-    successTest(2, 3, 1, 5, 2, 2, true, SQLModel.TREE_MODEL_SQL, ConsensusFactory.IOT_CONSENSUS);
+  public void success1C5DRemoveTwoDataNodesSharingRegionUseTableSQL() throws Exception {
+    // IoTConsensus is weakly consistent, so two replicas of a three-replica DataRegion can be
+    // removed by one statement. The procedure migrates the two replicas in separate rounds.
+    successTestWithDataRegionConflict(
+        3, 3, 1, 5, 2, 2, true, SQLModel.TABLE_MODEL_SQL, ConsensusFactory.IOT_CONSENSUS);
   }
 
   //  @Test
@@ -209,7 +212,33 @@ private void successTest(
         true,
         rejoinRemovedDataNode,
         model,
-        dataRegionGroupConsensusProtocol);
+        dataRegionGroupConsensusProtocol,
+        false);
+  }
+
+  private void successTestWithDataRegionConflict(
+      final int dataReplicateFactor,
+      final int schemaReplicationFactor,
+      final int configNodeNum,
+      final int dataNodeNum,
+      final int removeDataNodeNum,
+      final int dataRegionPerDataNode,
+      final boolean rejoinRemovedDataNode,
+      final SQLModel model,
+      final String dataRegionGroupConsensusProtocol)
+      throws Exception {
+    testRemoveDataNode(
+        dataReplicateFactor,
+        schemaReplicationFactor,
+        configNodeNum,
+        dataNodeNum,
+        removeDataNodeNum,
+        dataRegionPerDataNode,
+        true,
+        rejoinRemovedDataNode,
+        model,
+        dataRegionGroupConsensusProtocol,
+        true);
   }
 
   private void failTest(
@@ -233,7 +262,8 @@ private void failTest(
         false,
         rejoinRemovedDataNode,
         model,
-        dataRegionGroupConsensusProtocol);
+        dataRegionGroupConsensusProtocol,
+        false);
   }
 
   public void testRemoveDataNode(
@@ -246,7 +276,8 @@ public void testRemoveDataNode(
       final boolean expectRemoveSuccess,
       final boolean rejoinRemovedDataNode,
       final SQLModel model,
-      final String dataRegionGroupConsensusProtocol)
+      final String dataRegionGroupConsensusProtocol,
+      final boolean selectWithDataRegionConflict)
       throws Exception {
     // Set up specific environment
     EnvFactory.getEnv()
@@ -288,17 +319,17 @@ public void testRemoveDataNode(
         allDataNodeId.add(result.getInt(ColumnHeaderConstant.NODE_ID));
       }
 
-      // Select data nodes to remove. When removing more than one DataNode we must avoid picking two
-      // DataNodes that host replicas of the same consensus group, otherwise the
-      // RemoveDataNodesProcedure would try to migrate two replicas of one group at once, which the
-      // ConfigNode rejects ("Only one replica of the same consensus group is allowed to be migrated
-      // at the same time."). Selecting purely at random therefore makes this test flaky, so use a
-      // conflict-free selection.
+      // Most multi-node cases use a conflict-free selection so strong-consistency RegionGroups do
+      // not exceed their removal threshold. The targeted weak-consistency case deliberately picks
+      // replicas from the same DataRegion to verify round-based migration orchestration.
       final Set<Integer> removeDataNodes =
-          removeDataNodeNum > 1
-              ? selectRemoveDataNodesWithoutRegionConflict(
-                  allDataNodeId, removeDataNodeNum, getAllRegionMap(statement))
-              : selectRemoveDataNodes(allDataNodeId, removeDataNodeNum);
+          selectWithDataRegionConflict
+              ? selectRemoveDataNodesSharingWeakRegionSafely(
+                  removeDataNodeNum, regionMap, getSchemaRegionMap(statement))
+              : removeDataNodeNum > 1
+                  ? selectRemoveDataNodesWithoutRegionConflict(
+                      allDataNodeId, removeDataNodeNum, getAllRegionMap(statement))
+                  : selectRemoveDataNodes(allDataNodeId, removeDataNodeNum);
 
       List<DataNodeWrapper> removeDataNodeWrappers =
           removeDataNodes.stream()

integration-test/src/test/java/org/apache/iotdb/confignode/it/removedatanode/IoTDBRemoveDataNodeUtils.java

@@ -76,15 +76,68 @@ public static Set<Integer> selectRemoveDataNodes(
     return new HashSet<>(shuffledDataNodeIds.subList(0, removeDataNodeNum));
   }
 
+  /**
+   * Select DataNodes that share a weak-consistency RegionGroup while keeping every affected weak
+   * and strong RegionGroup within its removal-safety threshold.
+   */
+  public static Set<Integer> selectRemoveDataNodesSharingWeakRegionSafely(
+      int removeDataNodeNum,
+      Map<Integer, Set<Integer>> weakRegionMap,
+      Map<Integer, Set<Integer>> strongRegionMap) {
+    for (Set<Integer> weakRegionDataNodes : weakRegionMap.values()) {
+      Set<Integer> selected = new LinkedHashSet<>();
+      if (searchSafeSharedRegionSelection(
+          new ArrayList<>(weakRegionDataNodes),
+          0,
+          removeDataNodeNum,
+          selected,
+          weakRegionMap,
+          strongRegionMap)) {
+        return selected;
+      }
+    }
+    throw new IllegalStateException(
+        String.format(
+            "Cannot select %d DataNodes that safely share a weak-consistency RegionGroup. "
+                + "weakRegionMap=%s, strongRegionMap=%s",
+            removeDataNodeNum, weakRegionMap, strongRegionMap));
+  }
+
+  private static boolean searchSafeSharedRegionSelection(
+      List<Integer> candidates,
+      int start,
+      int need,
+      Set<Integer> selected,
+      Map<Integer, Set<Integer>> weakRegionMap,
+      Map<Integer, Set<Integer>> strongRegionMap) {
+    if (selected.size() == need) {
+      return weakRegionMap.values().stream()
+              .allMatch(region -> countSelectedReplicas(selected, region) < region.size())
+          && strongRegionMap.values().stream()
+              .allMatch(region -> 2L * countSelectedReplicas(selected, region) < region.size());
+    }
+    for (int i = start; i < candidates.size(); i++) {
+      selected.add(candidates.get(i));
+      if (searchSafeSharedRegionSelection(
+          candidates, i + 1, need, selected, weakRegionMap, strongRegionMap)) {
+        return true;
+      }
+      selected.remove(candidates.get(i));
+    }
+    return false;
+  }
+
+  private static long countSelectedReplicas(Set<Integer> selected, Set<Integer> regionDataNodes) {
+    return selected.stream().filter(regionDataNodes::contains).count();
+  }
+
   /**
    * Select {@code removeDataNodeNum} DataNodes to remove such that no two of them host a replica of
-   * the same consensus group. Removing two DataNodes that share a region group would make the
-   * generated {@code RemoveDataNodesProcedure} try to migrate two replicas of the same group at
-   * once, which the ConfigNode rejects ("Only one replica of the same consensus group is allowed to
-   * be migrated at the same time."). Randomly picking DataNodes (see {@link
-   * #selectRemoveDataNodes}) therefore makes multi-DataNode-remove tests flaky; this method instead
-   * searches exhaustively (with backtracking) for a conflict-free selection, so it fails only when
-   * no such selection exists rather than when an unlucky pick order dead-ends.
+   * the same consensus group. Most removal tests are not intended to exercise the per-RegionGroup
+   * safety threshold, so randomly picking DataNodes (see {@link #selectRemoveDataNodes}) can make
+   * them fail when the selected nodes contain too many replicas of a strong-consistency group. This
+   * method instead searches exhaustively (with backtracking) for a conflict-free selection, so it
+   * fails only when no such selection exists rather than when an unlucky pick order dead-ends.
    *
    * @param allDataNodeId all registered DataNode ids
    * @param removeDataNodeNum how many DataNodes to remove

iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ManagerMessages.java

@@ -660,6 +660,12 @@ private ManagerMessages() {}
       "Remove ConfigNode failed because there is no other ConfigNode in Running status in current"
       + " Cluster.";
   public static final String MESSAGE_REMOVE_CONFIGNODE_FAILED_BECAUSE_CONFIGNODEGROUP_LEADER_ELECTION_PLEASE_RETRY_3EE602F6 = "Remove ConfigNode failed because the ConfigNodeGroup is on leader election, please retry.";
+  public static final String
+      MESSAGE_CANNOT_REMOVE_CONFIGNODE_ARG_BECAUSE_IT_IS_ALREADY_BEING_REMOVED_EEE0128D =
+          "Cannot remove ConfigNode %d because it is already being removed.";
+  public static final String
+      MESSAGE_CANNOT_REMOVE_CONFIGNODE_ARG_REMOVING_ARG_CONFIGNODES_AT_THE_SAME_TIME_FROM_A_ARG_REPLICA_CONFIGNODEGROUP_USING_ARG_WOULD_VIOLATE_THE_SAFETY_CONDITION_ARG_REMOVINGCONFIGNODECOUNT_REPLICACOUNT_A64BF23B =
+          "Cannot remove ConfigNode %d: removing %d ConfigNodes at the same time from a %d-replica ConfigNodeGroup using %s would violate the safety condition %d * removingConfigNodeCount < replicaCount.";
   public static final String MESSAGE_TRANSFER_CONFIGNODE_LEADER_FAILED_BECAUSE_CAN_NOT_FIND_ANY_RUNNING_1FE4F96D = "Transfer ConfigNode leader failed because can not find any running ConfigNode.";
   public static final String MESSAGE_CONFIGNODE_REMOVED_LEADER_ALREADY_TRANSFER_LEADER_FA6D1603 = "The ConfigNode to be removed is leader, already transfer Leader to ";
   public static final String MESSAGE_TARGET_DATANODE_NOT_EXISTED_PLEASE_ENSURE_YOUR_INPUT_QUERYID_CORRECT_AB84CCDF = "The target DataNode is not existed, please ensure your input <queryId> is correct";

iotdb-core/confignode/src/main/i18n/en/org/apache/iotdb/confignode/i18n/ProcedureMessages.java

@@ -460,6 +460,9 @@ public final class ProcedureMessages {
       "Cannot remove %d DataNode(s): the cluster has %d available DataNode(s) and must retain at least %d of them (max(schema_replication_factor=%d, data_replication_factor=%d)) so that every region keeps enough replicas, but this request would leave only %d.";
   public static final String FAILED_TO_REMOVE_DATA_NODE_SINGLE_REPLICA_HINT =
       " With a single replica there is nowhere to migrate regions to, so at least one DataNode must always remain.";
+  public static final String
+      MESSAGE_CANNOT_REMOVE_ARG_REPLICAS_FROM_REGION_ARG_WITH_ARG_REPLICAS_USING_ARG_BECAUSE_THE_REMOVAL_SAFETY_CONDITION_ARG_REMOVEDREPLICACOUNT_REPLICACOUNT_IS_NOT_SATISFIED_4CF41C73 =
+          "Cannot remove %d replicas from region %s with %d replicas using %s because the removal safety condition %d * removedReplicaCount < replicaCount is not satisfied.";
   public static final String FAILED_TO_ROLLBACK_ALTER_PIPE_DETAILS_METADATA_WILL_BE_SYNCHRONIZED =
       "Failed to rollback alter pipe {}, details: {}, metadata will be synchronized later.";
   public static final String FAILED_TO_ROLLBACK_COMMIT_SET_TEMPLATE_ON_PATH_DUE_TO =

iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ManagerMessages.java

@@ -640,6 +640,12 @@ private ManagerMessages() {}
   public static final String MESSAGE_REMOVE_CONFIGNODE_FAILED_BECAUSE_THERE_ONLY_ONE_CONFIGNODE_CURRENT_CLUSTER_D1273758 = "移除 ConfigNode 失败，原因：当前集群中只有一个 ConfigNode。";
   public static final String MESSAGE_REMOVE_CONFIGNODE_FAILED_BECAUSE_THERE_NO_OTHER_CONFIGNODE_RUNNING_STATUS_C9C43315 = "移除 ConfigNode 失败，原因：当前集群中没有其他处于 Running 状态的 ConfigNode。";
   public static final String MESSAGE_REMOVE_CONFIGNODE_FAILED_BECAUSE_CONFIGNODEGROUP_LEADER_ELECTION_PLEASE_RETRY_3EE602F6 = "移除 ConfigNode 失败，原因：ConfigNodeGroup 正在进行 leader 选举，请重试。";
+  public static final String
+      MESSAGE_CANNOT_REMOVE_CONFIGNODE_ARG_BECAUSE_IT_IS_ALREADY_BEING_REMOVED_EEE0128D =
+          "无法移除 ConfigNode %d，因为该节点已处于移除流程中。";
+  public static final String
+      MESSAGE_CANNOT_REMOVE_CONFIGNODE_ARG_REMOVING_ARG_CONFIGNODES_AT_THE_SAME_TIME_FROM_A_ARG_REPLICA_CONFIGNODEGROUP_USING_ARG_WOULD_VIOLATE_THE_SAFETY_CONDITION_ARG_REMOVINGCONFIGNODECOUNT_REPLICACOUNT_A64BF23B =
+          "无法移除 ConfigNode %d：同时移除 %d 个 ConfigNode（ConfigNodeGroup 当前为 %d 副本并使用 %s）将违反安全条件：%d * removingConfigNodeCount < replicaCount。";
   public static final String MESSAGE_TRANSFER_CONFIGNODE_LEADER_FAILED_BECAUSE_CAN_NOT_FIND_ANY_RUNNING_1FE4F96D = "转移 ConfigNode leader 失败，原因：找不到任何正在运行的 ConfigNode。";
   public static final String MESSAGE_CONFIGNODE_REMOVED_LEADER_ALREADY_TRANSFER_LEADER_FA6D1603 = "待移除的 ConfigNode 是 leader，已将 Leader 转移到 ";
   public static final String MESSAGE_TARGET_DATANODE_NOT_EXISTED_PLEASE_ENSURE_YOUR_INPUT_QUERYID_CORRECT_AB84CCDF = "目标 DataNode 不存在，请确保输入的 <queryId> 正确";

iotdb-core/confignode/src/main/i18n/zh/org/apache/iotdb/confignode/i18n/ProcedureMessages.java

@@ -446,6 +446,9 @@ public final class ProcedureMessages {
       "无法移除 %d 个 DataNode：集群当前有 %d 个可用 DataNode，且至少需保留 %d 个（max(schema_replication_factor=%d, data_replication_factor=%d)），以保证每个 Region 仍有足够的副本；但本次请求执行后将只剩 %d 个。";
   public static final String FAILED_TO_REMOVE_DATA_NODE_SINGLE_REPLICA_HINT =
       " 单副本下没有其它节点可供迁移 Region，因此必须始终保留至少一个 DataNode。";
+  public static final String
+      MESSAGE_CANNOT_REMOVE_ARG_REPLICAS_FROM_REGION_ARG_WITH_ARG_REPLICAS_USING_ARG_BECAUSE_THE_REMOVAL_SAFETY_CONDITION_ARG_REMOVEDREPLICACOUNT_REPLICACOUNT_IS_NOT_SATISFIED_4CF41C73 =
+          "无法移除 %d 个副本：Region %s 当前有 %d 个副本并使用 %s，未满足移除安全条件：%d * removedReplicaCount < replicaCount。";
   public static final String FAILED_TO_ROLLBACK_ALTER_PIPE_DETAILS_METADATA_WILL_BE_SYNCHRONIZED =
       "回滚修改 pipe {} 失败，详情：{}，元数据将稍后同步。";
   public static final String FAILED_TO_ROLLBACK_COMMIT_SET_TEMPLATE_ON_PATH_DUE_TO =

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/ConfigManager.java

@@ -1606,7 +1606,7 @@ public TSStatus createPeerForConsensusGroup(List<TConfigNodeLocation> configNode
   }
 
   @Override
-  public TSStatus removeConfigNode(RemoveConfigNodePlan removeConfigNodePlan) {
+  public synchronized TSStatus removeConfigNode(RemoveConfigNodePlan removeConfigNodePlan) {
     TSStatus status = confirmLeader();
 
     if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {