Validate region operation requests before submitting procedures

integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBMigrateMultiRegionForIoTV1IT.java

@@ -36,6 +36,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.sql.Connection;
+import java.sql.SQLException;
 import java.sql.Statement;
 import java.util.ArrayList;
 import java.util.List;
@@ -125,14 +126,8 @@ public void multiRegionMigrateTest() throws Exception {
                 try {
                   statement.execute(command);
                   return true;
-                } catch (Exception e) {
+                } catch (SQLException e) {
                   String errorMessage = e.getMessage();
-                  if (errorMessage != null
-                      && errorMessage.contains("successfully submitted")
-                      && errorMessage.contains("failed to submit")) {
-                    LOGGER.warn("Multi-region migrate partially succeeded: {}", errorMessage);
-                    return true;
-                  }
                   LOGGER.warn("Multi-region migrate failed, retrying: {}", errorMessage);
                   return false;
                 }

integration-test/src/test/java/org/apache/iotdb/confignode/it/regionmigration/pass/commit/IoTDBRegionGroupExpandAndShrinkForIoTV1IT.java

@@ -26,6 +26,7 @@
 import org.apache.iotdb.it.env.EnvFactory;
 import org.apache.iotdb.it.framework.IoTDBTestRunner;
 import org.apache.iotdb.itbase.category.ClusterIT;
+import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.awaitility.Awaitility;
 import org.junit.Assert;
@@ -168,14 +169,13 @@ public void extendRegionToInvalidDataNodeTest() throws Exception {
         Assert.assertFalse(
             "ConfigNode should not throw NullPointerException, but got: " + message,
             message.contains("NullPointerException"));
-        // ... and the submission must be rejected cleanly. "extend region" wraps every region's
-        // result, so the top-level message only reports the aggregate counts; the concrete "does
-        // not
-        // exist in the cluster" reason is carried in the per-region sub-status.
+        // The operation-specific error and the concrete reason must reach the JDBC client.
+        Assert.assertEquals(TSStatusCode.EXTEND_REGION_ERROR.getStatusCode(), e.getErrorCode());
         Assert.assertTrue(
             "Expected the extend submission to be rejected but got: " + message,
-            message.contains("failed to submit: 1"));
+            message.contains("Target DataNode " + invalidDataNodeId + " does not exist"));
       }
+      assertRegionMapUnchanged(statement, regionMap);
     }
   }
 
@@ -314,7 +314,7 @@ public void multiRegionNormalTest() throws Exception {
     }
   }
 
-  /** Test multi-region expand with partial regions already in target DataNode */
+  /** Reject the entire expansion when a later region already exists on the target DataNode. */
   @Test
   public void multiRegionExpandPartialExistTest() throws Exception {
     EnvFactory.getEnv()
@@ -338,49 +338,42 @@ public void multiRegionExpandPartialExistTest() throws Exception {
       Map<Integer, Set<Integer>> regionMap = getAllRegionMap(statement);
       Set<Integer> allDataNodeId = getAllDataNodes(statement);
 
-      List<Integer> allRegions = new ArrayList<>(regionMap.keySet());
-      List<Integer> selectedRegions = allRegions.subList(0, Math.min(3, allRegions.size()));
+      Assert.assertEquals(2, regionMap.size());
+      List<Integer> selectedRegions = new ArrayList<>(regionMap.keySet());
 
       int targetDataNode =
           findDataNodeNotContainsAnyRegion(allDataNodeId, regionMap, selectedRegions);
 
-      // first expand some regions individually
-      List<Integer> preExpandRegions =
-          selectedRegions.subList(0, Math.min(2, selectedRegions.size()));
-      for (int regionId : preExpandRegions) {
-        regionGroupExpand(statement, client, regionId, targetDataNode);
-      }
-
-      // now try to expand all regions (including already expanded ones)
-      LOGGER.info(
-          "Testing multi-expand with regions {} to DataNode {}, where {} already exist",
-          selectedRegions,
-          targetDataNode,
-          preExpandRegions);
-
-      multiRegionGroupExpand(statement, client, selectedRegions, targetDataNode);
-
-      // verify all regions are in target DataNode
+      // Keep the first region valid so this also catches submission during validation.
+      regionGroupExpand(statement, client, selectedRegions.get(1), targetDataNode);
       regionMap = getAllRegionMap(statement);
-      for (int regionId : selectedRegions) {
-        Assert.assertTrue(
-            "Region " + regionId + " should contain target DataNode " + targetDataNode,
-            regionMap.get(regionId).contains(targetDataNode));
-      }
-      LOGGER.info("Multi-region expand partial exist test passed");
+      Assert.assertFalse(regionMap.get(selectedRegions.get(0)).contains(targetDataNode));
+      Assert.assertTrue(regionMap.get(selectedRegions.get(1)).contains(targetDataNode));
+
+      SQLException exception =
+          Assert.assertThrows(
+              SQLException.class,
+              () ->
+                  statement.execute(
+                      buildMultiRegionCommand(
+                          MULTI_EXPAND_FORMAT, selectedRegions, targetDataNode)));
+      Assert.assertEquals(
+          TSStatusCode.EXTEND_REGION_ERROR.getStatusCode(), exception.getErrorCode());
+      Assert.assertTrue(exception.getMessage().contains("already contains region"));
+      assertRegionMapUnchanged(statement, regionMap);
     }
   }
 
-  /** Test multi-region shrink with partial regions not in target DataNode */
+  /** Reject the entire removal when a later region does not exist on the target DataNode. */
   @Test
   public void multiRegionShrinkPartialNotExistTest() throws Exception {
     EnvFactory.getEnv()
         .getConfig()
         .getCommonConfig()
         .setDataRegionConsensusProtocolClass(ConsensusFactory.IOT_CONSENSUS)
         .setSchemaRegionConsensusProtocolClass(ConsensusFactory.RATIS_CONSENSUS)
-        .setDataReplicationFactor(1)
-        .setSchemaReplicationFactor(1);
+        .setDataReplicationFactor(2)
+        .setSchemaReplicationFactor(2);
 
     EnvFactory.getEnv().initClusterEnvironment(1, 5);
 
@@ -395,42 +388,43 @@ public void multiRegionShrinkPartialNotExistTest() throws Exception {
       Map<Integer, Set<Integer>> regionMap = getAllRegionMap(statement);
       Set<Integer> allDataNodeId = getAllDataNodes(statement);
 
-      List<Integer> allRegions = new ArrayList<>(regionMap.keySet());
-      List<Integer> selectedRegions = allRegions.subList(0, Math.min(3, allRegions.size()));
+      Assert.assertEquals(2, regionMap.size());
+      List<Integer> selectedRegions = new ArrayList<>(regionMap.keySet());
 
       int targetDataNode =
           findDataNodeNotContainsAnyRegion(allDataNodeId, regionMap, selectedRegions);
 
       // first expand all regions to target DataNode
       multiRegionGroupExpand(statement, client, selectedRegions, targetDataNode);
 
-      // then shrink some regions individually
-      List<Integer> preShrinkRegions =
-          selectedRegions.subList(0, Math.min(2, selectedRegions.size()));
-      for (int regionId : preShrinkRegions) {
-        regionGroupShrink(statement, client, regionId, targetDataNode);
-      }
-
-      // now try to shrink all regions (including already shrunk ones)
-      LOGGER.info(
-          "Testing multi-shrink with regions {} from DataNode {}, where {} already removed",
-          selectedRegions,
-          targetDataNode,
-          preShrinkRegions);
-
-      multiRegionGroupShrink(statement, client, selectedRegions, targetDataNode);
-
-      // verify all regions are not in target DataNode
+      // Keep the first region valid and leave two replicas of the second region elsewhere.
+      regionGroupShrink(statement, client, selectedRegions.get(1), targetDataNode);
       regionMap = getAllRegionMap(statement);
-      for (int regionId : selectedRegions) {
-        Assert.assertFalse(
-            "Region " + regionId + " should not contain target DataNode " + targetDataNode,
-            regionMap.get(regionId).contains(targetDataNode));
-      }
-      LOGGER.info("Multi-region shrink partial not exist test passed");
+      Assert.assertTrue(regionMap.get(selectedRegions.get(0)).contains(targetDataNode));
+      Assert.assertFalse(regionMap.get(selectedRegions.get(1)).contains(targetDataNode));
+
+      SQLException exception =
+          Assert.assertThrows(
+              SQLException.class,
+              () ->
+                  statement.execute(
+                      buildMultiRegionCommand(
+                          MULTI_SHRINK_FORMAT, selectedRegions, targetDataNode)));
+      Assert.assertEquals(
+          TSStatusCode.REMOVE_REGION_PEER_ERROR.getStatusCode(), exception.getErrorCode());
+      Assert.assertTrue(exception.getMessage().contains("doesn't contain Region"));
+      assertRegionMapUnchanged(statement, regionMap);
     }
   }
 
+  private void assertRegionMapUnchanged(
+      Statement statement, Map<Integer, Set<Integer>> expectedRegionMap) {
+    Awaitility.await()
+        .during(3, TimeUnit.SECONDS)
+        .atMost(10, TimeUnit.SECONDS)
+        .untilAsserted(() -> Assert.assertEquals(expectedRegionMap, getAllRegionMap(statement)));
+  }
+
   private void multiRegionGroupExpand(
       Statement statement,
       SyncConfigNodeIServiceClient client,
@@ -519,17 +513,8 @@ private void executeMultiRegionOperation(
               try {
                 statement.execute(command);
                 return true;
-              } catch (Exception e) {
+              } catch (SQLException e) {
                 String errorMessage = e.getMessage();
-                // If error message contains both "successfully submitted" and "failed to submit",
-                // consider it as partial success and continue
-                if (errorMessage != null
-                    && errorMessage.contains("successfully submitted")
-                    && errorMessage.contains("failed to submit")) {
-                  LOGGER.warn(
-                      "Multi-region {} partially succeeded: {}", operationType, errorMessage);
-                  return true;
-                }
                 LOGGER.warn(
                     "Multi-region {} command execution failed, retrying: {}",
                     operationType,

iotdb-client/cli/src/test/java/org/apache/iotdb/cli/AbstractCliTest.java

@@ -22,9 +22,13 @@
 import org.apache.iotdb.cli.AbstractCli.OperationResult;
 import org.apache.iotdb.cli.type.ExitType;
 import org.apache.iotdb.cli.utils.CliContext;
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
 import org.apache.iotdb.exception.ArgsErrorException;
 import org.apache.iotdb.jdbc.IoTDBConnection;
+import org.apache.iotdb.jdbc.IoTDBConnectionParams;
 import org.apache.iotdb.jdbc.IoTDBDatabaseMetadata;
+import org.apache.iotdb.jdbc.IoTDBSQLException;
+import org.apache.iotdb.rpc.TSStatusCode;
 
 import org.apache.commons.cli.CommandLine;
 import org.apache.commons.cli.CommandLineParser;
@@ -44,6 +48,7 @@
 import java.lang.reflect.Field;
 import java.lang.reflect.Method;
 import java.nio.charset.StandardCharsets;
+import java.sql.Statement;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
@@ -52,6 +57,7 @@
 import static org.junit.Assert.assertFalse;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
 
 public class AbstractCliTest {
@@ -71,6 +77,44 @@ public void setUp() throws Exception {
   public void tearDown() throws Exception {
     setStaticField("lineCount", 0);
     setStaticField("isReachEnd", false);
+    AbstractCli.lastProcessStatus = AbstractCli.CODE_OK;
+  }
+
+  @Test
+  public void testRegionValidationErrorIsPrintedAndReturnsErrorStatus() throws Exception {
+    String[] statements = {
+      "MIGRATE REGION 12,99 FROM 6 TO 7",
+      "RECONSTRUCT REGION 12,99 ON 7",
+      "EXTEND REGION 12,99 TO 7",
+      "REMOVE REGION 12,99 FROM 7"
+    };
+    TSStatusCode[] codes = {
+      TSStatusCode.MIGRATE_REGION_ERROR,
+      TSStatusCode.RECONSTRUCT_REGION_ERROR,
+      TSStatusCode.EXTEND_REGION_ERROR,
+      TSStatusCode.REMOVE_REGION_PEER_ERROR
+    };
+    when(connection.getParams())
+        .thenReturn(new IoTDBConnectionParams("jdbc:iotdb://localhost:6667/"));
+    for (int i = 0; i < statements.length; i++) {
+      ByteArrayOutputStream out = new ByteArrayOutputStream();
+      try (PrintStream printer = new PrintStream(out, true, StandardCharsets.UTF_8.name())) {
+        CliContext ctx = new CliContext(System.in, printer, System.err, ExitType.EXCEPTION);
+        Statement statement = mock(Statement.class);
+        when(connection.createStatement()).thenReturn(statement);
+        TSStatus status =
+            new TSStatus(codes[i].getStatusCode()).setMessage("Region 99 does not exist");
+        when(statement.execute(statements[i]))
+            .thenThrow(new IoTDBSQLException(status.getMessage(), status));
+
+        AbstractCli.handleInputCmd(ctx, statements[i], connection);
+
+        assertEquals(AbstractCli.CODE_ERROR, AbstractCli.lastProcessStatus);
+        String output = new String(out.toByteArray(), StandardCharsets.UTF_8);
+        assertTrue(output.contains(status.getMessage()));
+        assertFalse(output.contains("The statement is executed successfully"));
+      }
+    }
   }
 
   @Test

iotdb-client/jdbc/src/test/java/org/apache/iotdb/jdbc/IoTDBStatementTest.java

@@ -19,8 +19,11 @@
 
 package org.apache.iotdb.jdbc;
 
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
 import org.apache.iotdb.rpc.RpcUtils;
+import org.apache.iotdb.rpc.TSStatusCode;
 import org.apache.iotdb.service.rpc.thrift.IClientRPCService.Iface;
+import org.apache.iotdb.service.rpc.thrift.TSExecuteStatementResp;
 import org.apache.iotdb.service.rpc.thrift.TSFetchMetadataReq;
 import org.apache.iotdb.service.rpc.thrift.TSFetchMetadataResp;
 
@@ -106,4 +109,40 @@ public void setTimeoutTest() throws SQLException {
     statement.setQueryTimeout(100);
     Assert.assertEquals(100, statement.getQueryTimeout());
   }
+
+  @SuppressWarnings("resource")
+  @Test
+  public void regionValidationErrorsArePropagatedByExecuteAndExecuteUpdate() throws Exception {
+    String[] statements = {
+      "MIGRATE REGION 1,1 FROM 2 TO 3",
+      "RECONSTRUCT REGION 1,1 ON 2",
+      "EXTEND REGION 1,1 TO 2",
+      "REMOVE REGION 1,1 FROM 2"
+    };
+    TSStatusCode[] codes = {
+      TSStatusCode.MIGRATE_REGION_ERROR,
+      TSStatusCode.RECONSTRUCT_REGION_ERROR,
+      TSStatusCode.EXTEND_REGION_ERROR,
+      TSStatusCode.REMOVE_REGION_PEER_ERROR
+    };
+    for (int i = 0; i < statements.length; i++) {
+      final String sql = statements[i];
+      TSStatus status =
+          new TSStatus(codes[i].getStatusCode()).setMessage("Duplicate Region ID 1 in the request");
+      TSExecuteStatementResp response = new TSExecuteStatementResp().setStatus(status);
+      when(client.executeStatementV2(any())).thenReturn(response);
+      when(client.executeUpdateStatement(any())).thenReturn(response);
+      IoTDBStatement statement = new IoTDBStatement(connection, client, sessionId, zoneID, 0, 1L);
+
+      SQLException executeError =
+          Assert.assertThrows(SQLException.class, () -> statement.execute(sql));
+      assertEquals(status.getCode(), executeError.getErrorCode());
+      Assert.assertTrue(executeError.getMessage().contains(status.getMessage()));
+      SQLException updateError =
+          Assert.assertThrows(SQLException.class, () -> statement.executeUpdate(sql));
+      assertEquals(status.getCode(), updateError.getErrorCode());
+      Assert.assertTrue(updateError.getMessage().contains(status.getMessage()));
+      Assert.assertNull(statement.getWarnings());
+    }
+  }
 }

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

@@ -295,6 +295,20 @@ public final class ManagerMessages {
   public static final String
       LOG_SKIP_NON_EXISTENT_REGION_ID_ARG_IN_RECONSTRUCTREGION_REQUEST_TO_DATANODE_ARG_7F76D789 =
           "Skip non-existent Region ID {} in ReconstructRegion request to DataNode {}.";
+  public static final String MESSAGE_DUPLICATE_REGION_ID_ARG_IN_THE_REQUEST_B6FFCCFC =
+      "Duplicate Region ID %d in the request";
+  public static final String MESSAGE_REGION_IDS_MUST_NOT_BE_EMPTY_B42DAAFD =
+      "Region IDs must not be empty";
+  public static final String MESSAGE_SOURCE_AND_TARGET_DATANODE_IDS_MUST_BE_DIFFERENT_ARG_286D3838 =
+      "Source and target DataNode IDs must be different: %d";
+  public static final String LOG_SUBMIT_REGION_OPERATION_PROCEDURE_SUCCESSFULLY_ARG_90468B38 =
+      "Submit region operation procedure successfully: {}";
+  public static final String MESSAGE_REGION_ARG_DOES_NOT_EXIST_3C8400C9 =
+      "Region %d does not exist";
+  public static final String MESSAGE_SOURCE_DATANODE_ARG_DOES_NOT_EXIST_IN_THE_CLUSTER_2255633C =
+      "Source DataNode %s does not exist in the cluster";
+  public static final String MESSAGE_TARGET_DATANODE_ARG_DOES_NOT_EXIST_IN_THE_CLUSTER_679D59AF =
+      "Target DataNode %s does not exist in the cluster";
   public static final String MIGRATEREGION_SUBMIT_REGIONMIGRATEPROCEDURE_SUCCESSFULLY_REGION_ORIGIN_DATANODE =
       "[MigrateRegion] Submit RegionMigrateProcedure successfully, Region: {}, Origin DataNode: {}, Dest DataNode: {}, Add Coordinator: {}, Remove Coordinator: {}";
   public static final String SUBMIT_REGIONMIGRATEPROCEDURE_FAILED_BECAUSE_REGIONGROUP_DOESN_T_EXIST =

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

@@ -293,6 +293,20 @@ public final class ManagerMessages {
   public static final String
       LOG_SKIP_NON_EXISTENT_REGION_ID_ARG_IN_RECONSTRUCTREGION_REQUEST_TO_DATANODE_ARG_7F76D789 =
           "跳过 ReconstructRegion 请求中不存在的 Region ID {}，目标 DataNode 为 {}。";
+  public static final String MESSAGE_DUPLICATE_REGION_ID_ARG_IN_THE_REQUEST_B6FFCCFC =
+      "请求中包含重复的 Region ID %d";
+  public static final String MESSAGE_REGION_IDS_MUST_NOT_BE_EMPTY_B42DAAFD =
+      "Region ID 列表不能为空";
+  public static final String MESSAGE_SOURCE_AND_TARGET_DATANODE_IDS_MUST_BE_DIFFERENT_ARG_286D3838 =
+      "源和目标 DataNode ID 不能相同：%d";
+  public static final String LOG_SUBMIT_REGION_OPERATION_PROCEDURE_SUCCESSFULLY_ARG_90468B38 =
+      "成功提交 Region 运维 procedure：{}";
+  public static final String MESSAGE_REGION_ARG_DOES_NOT_EXIST_3C8400C9 =
+      "Region %d 不存在";
+  public static final String MESSAGE_SOURCE_DATANODE_ARG_DOES_NOT_EXIST_IN_THE_CLUSTER_2255633C =
+      "源 DataNode %s 不存在于集群中";
+  public static final String MESSAGE_TARGET_DATANODE_ARG_DOES_NOT_EXIST_IN_THE_CLUSTER_679D59AF =
+      "目标 DataNode %s 不存在于集群中";
   public static final String MIGRATEREGION_SUBMIT_REGIONMIGRATEPROCEDURE_SUCCESSFULLY_REGION_ORIGIN_DATANODE =
       "[MigrateRegion] 成功提交 RegionMigrateProcedure，Region：{}，原 DataNode：{}，目标 DataNode：{}，新增 Coordinator：{}，移除 Coordinator：{}";
   public static final String SUBMIT_REGIONMIGRATEPROCEDURE_FAILED_BECAUSE_REGIONGROUP_DOESN_T_EXIST =