[To dev/1.3] Harden RegionMaintainer retries and region operation atomicity

iotdb-client/service-rpc/src/main/java/org/apache/iotdb/rpc/TSStatusCode.java

@@ -166,6 +166,8 @@ public enum TSStatusCode {
   RECONSTRUCT_REGION_ERROR(908),
   EXTEND_REGION_ERROR(909),
   REMOVE_REGION_PEER_ERROR(910),
+  REGION_ALREADY_EXISTS(911),
+  REGION_NOT_EXIST(912),
 
   // Cluster Manager
   ADD_CONFIGNODE_ERROR(1000),

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeTSStatusRPCHandler.java

@@ -52,7 +52,7 @@ public void onComplete(TSStatus response) {
     // Put response
     responseMap.put(requestId, response);
 
-    if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+    if (isRequestCompleted(requestType, response)) {
       // Remove only if success
       nodeLocationMap.remove(requestId);
       LOGGER.info("Successfully {} on DataNode: {}", requestType, formattedTargetLocation);
@@ -68,6 +68,18 @@ public void onComplete(TSStatus response) {
     countDownLatch.countDown();
   }
 
+  static boolean isRequestCompleted(CnToDnAsyncRequestType requestType, TSStatus response) {
+    if (response.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+      return true;
+    }
+    if (requestType == CnToDnAsyncRequestType.DELETE_REGION) {
+      return response.getCode() == TSStatusCode.REGION_NOT_EXIST.getStatusCode();
+    }
+    return (requestType == CnToDnAsyncRequestType.CREATE_DATA_REGION
+            || requestType == CnToDnAsyncRequestType.CREATE_SCHEMA_REGION)
+        && response.getCode() == TSStatusCode.REGION_ALREADY_EXISTS.getStatusCode();
+  }
+
   @Override
   public void onError(Exception e) {
     String errorMsg =

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/procedure/impl/schema/DeleteDatabaseProcedure.java

@@ -112,38 +112,17 @@ protected Flow executeFromState(
               deleteDatabaseSchema.getName());
 
           // Submit RegionDeleteTasks
-          OfferRegionMaintainTasksPlan dataRegionDeleteTaskOfferPlan =
-              new OfferRegionMaintainTasksPlan();
           List<TRegionReplicaSet> regionReplicaSets =
               env.getAllReplicaSets(deleteDatabaseSchema.getName());
-          List<TRegionReplicaSet> schemaRegionReplicaSets = new ArrayList<>();
+          OfferRegionMaintainTasksPlan regionDeleteTaskOfferPlan =
+              buildDataRegionDeleteTaskOfferPlan(regionReplicaSets);
+          List<TRegionReplicaSet> schemaRegionReplicaSets =
+              getSchemaRegionReplicaSets(regionReplicaSets);
           regionReplicaSets.forEach(
-              regionReplicaSet -> {
-                // Clear heartbeat cache along the way
-                env.getConfigManager()
-                    .getLoadManager()
-                    .removeRegionGroupRelatedCache(regionReplicaSet.getRegionId());
-
-                if (regionReplicaSet
-                    .getRegionId()
-                    .getType()
-                    .equals(TConsensusGroupType.SchemaRegion)) {
-                  schemaRegionReplicaSets.add(regionReplicaSet);
-                } else {
-                  regionReplicaSet
-                      .getDataNodeLocations()
-                      .forEach(
-                          targetDataNode ->
-                              dataRegionDeleteTaskOfferPlan.appendRegionMaintainTask(
-                                  new RegionDeleteTask(
-                                      targetDataNode, regionReplicaSet.getRegionId())));
-                }
-              });
-
-          if (!dataRegionDeleteTaskOfferPlan.getRegionMaintainTaskList().isEmpty()) {
-            // submit async data region delete task
-            env.getConfigManager().getConsensusManager().write(dataRegionDeleteTaskOfferPlan);
-          }
+              regionReplicaSet ->
+                  env.getConfigManager()
+                      .getLoadManager()
+                      .removeRegionGroupRelatedCache(regionReplicaSet.getRegionId()));
 
           // try sync delete schemaengine region
           DataNodeAsyncRequestContext<TConsensusGroupId, TSStatus> asyncClientHandler =
@@ -166,7 +145,7 @@ protected Flow executeFromState(
                 .sendAsyncRequestWithRetry(asyncClientHandler);
             for (Map.Entry<Integer, TSStatus> entry :
                 asyncClientHandler.getResponseMap().entrySet()) {
-              if (entry.getValue().getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
+              if (isRegionDeleteCompleted(entry.getValue())) {
                 LOG.info(
                     "[DeleteDatabaseProcedure] Successfully delete SchemaRegion[{}] on {}",
                     asyncClientHandler.getRequest(entry.getKey()),
@@ -181,16 +160,17 @@ protected Flow executeFromState(
             }
 
             if (!schemaRegionDeleteTaskMap.isEmpty()) {
-              // submit async schemaengine region delete task for failed sync execution
-              OfferRegionMaintainTasksPlan schemaRegionDeleteTaskOfferPlan =
-                  new OfferRegionMaintainTasksPlan();
-              schemaRegionDeleteTaskMap
-                  .values()
-                  .forEach(schemaRegionDeleteTaskOfferPlan::appendRegionMaintainTask);
-              env.getConfigManager().getConsensusManager().write(schemaRegionDeleteTaskOfferPlan);
+              // submit async schemaengine region delete tasks for failed sync executions
+              appendFailedSchemaRegionDeleteTasks(
+                  regionDeleteTaskOfferPlan, schemaRegionDeleteTaskMap);
             }
           }
 
+          if (!offerRegionDeleteTasks(env, regionDeleteTaskOfferPlan)) {
+            setNextState(DeleteStorageGroupState.DELETE_DATABASE_SCHEMA);
+            return Flow.HAS_MORE_STATE;
+          }
+
           env.getConfigManager()
               .getLoadManager()
               .clearDataPartitionPolicyTable(deleteDatabaseSchema.getName());
@@ -212,6 +192,8 @@ protected Flow executeFromState(
           } else if (getCycles() > RETRY_THRESHOLD) {
             setFailure(
                 new ProcedureException("[DeleteDatabaseProcedure] Delete DatabaseSchema failed"));
+          } else {
+            setNextState(DeleteStorageGroupState.DELETE_DATABASE_SCHEMA);
           }
       }
     } catch (ConsensusException | TException | IOException e) {
@@ -230,12 +212,67 @@ protected Flow executeFromState(
             e);
         if (getCycles() > RETRY_THRESHOLD) {
           setFailure(new ProcedureException("[DeleteDatabaseProcedure] State stuck at " + state));
+        } else {
+          setNextState(state);
         }
       }
     }
     return Flow.HAS_MORE_STATE;
   }
 
+  static boolean isRegionDeleteCompleted(TSStatus status) {
+    return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()
+        || status.getCode() == TSStatusCode.REGION_NOT_EXIST.getStatusCode();
+  }
+
+  static OfferRegionMaintainTasksPlan buildDataRegionDeleteTaskOfferPlan(
+      List<TRegionReplicaSet> regionReplicaSets) {
+    OfferRegionMaintainTasksPlan offerPlan = new OfferRegionMaintainTasksPlan();
+    regionReplicaSets.stream()
+        .filter(
+            regionReplicaSet ->
+                regionReplicaSet.getRegionId().getType() == TConsensusGroupType.DataRegion)
+        .forEach(
+            regionReplicaSet ->
+                regionReplicaSet
+                    .getDataNodeLocations()
+                    .forEach(
+                        targetDataNode ->
+                            offerPlan.appendRegionMaintainTask(
+                                new RegionDeleteTask(
+                                    targetDataNode, regionReplicaSet.getRegionId()))));
+    return offerPlan;
+  }
+
+  private static List<TRegionReplicaSet> getSchemaRegionReplicaSets(
+      List<TRegionReplicaSet> regionReplicaSets) {
+    List<TRegionReplicaSet> schemaRegionReplicaSets = new ArrayList<>();
+    regionReplicaSets.stream()
+        .filter(
+            regionReplicaSet ->
+                regionReplicaSet.getRegionId().getType() == TConsensusGroupType.SchemaRegion)
+        .forEach(schemaRegionReplicaSets::add);
+    return schemaRegionReplicaSets;
+  }
+
+  static void appendFailedSchemaRegionDeleteTasks(
+      OfferRegionMaintainTasksPlan offerPlan,
+      Map<Integer, RegionDeleteTask> failedSchemaRegionDeleteTasks) {
+    failedSchemaRegionDeleteTasks.values().forEach(offerPlan::appendRegionMaintainTask);
+  }
+
+  private boolean offerRegionDeleteTasks(
+      ConfigNodeProcedureEnv env, OfferRegionMaintainTasksPlan offerPlan)
+      throws ConsensusException {
+    return offerPlan.getRegionMaintainTaskList().isEmpty()
+        || isRegionDeleteTaskOfferSuccessful(
+            env.getConfigManager().getConsensusManager().write(offerPlan));
+  }
+
+  static boolean isRegionDeleteTaskOfferSuccessful(TSStatus status) {
+    return status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode();
+  }
+
   @Override
   protected void rollbackState(ConfigNodeProcedureEnv env, DeleteStorageGroupState state)
       throws IOException, InterruptedException {

iotdb-core/confignode/src/test/java/org/apache/iotdb/confignode/client/async/handlers/rpc/DataNodeTSStatusRPCHandlerTest.java

@@ -0,0 +1,63 @@
+/*
+ * 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.client.async.handlers.rpc;
+
+import org.apache.iotdb.common.rpc.thrift.TSStatus;
+import org.apache.iotdb.confignode.client.async.CnToDnAsyncRequestType;
+import org.apache.iotdb.rpc.TSStatusCode;
+
+import org.junit.Test;
+
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertTrue;
+
+public class DataNodeTSStatusRPCHandlerTest {
+
+  @Test
+  public void testRegionOperationTerminalStatuses() {
+    assertTrue(
+        DataNodeTSStatusRPCHandler.isRequestCompleted(
+            CnToDnAsyncRequestType.CREATE_DATA_REGION, status(TSStatusCode.SUCCESS_STATUS)));
+    assertTrue(
+        DataNodeTSStatusRPCHandler.isRequestCompleted(
+            CnToDnAsyncRequestType.CREATE_DATA_REGION, status(TSStatusCode.REGION_ALREADY_EXISTS)));
+    assertTrue(
+        DataNodeTSStatusRPCHandler.isRequestCompleted(
+            CnToDnAsyncRequestType.CREATE_SCHEMA_REGION,
+            status(TSStatusCode.REGION_ALREADY_EXISTS)));
+    assertTrue(
+        DataNodeTSStatusRPCHandler.isRequestCompleted(
+            CnToDnAsyncRequestType.DELETE_REGION, status(TSStatusCode.REGION_NOT_EXIST)));
+
+    assertFalse(
+        DataNodeTSStatusRPCHandler.isRequestCompleted(
+            CnToDnAsyncRequestType.CREATE_SCHEMA_REGION, status(TSStatusCode.CREATE_REGION_ERROR)));
+    assertFalse(
+        DataNodeTSStatusRPCHandler.isRequestCompleted(
+            CnToDnAsyncRequestType.DELETE_REGION, status(TSStatusCode.DELETE_REGION_ERROR)));
+    assertFalse(
+        DataNodeTSStatusRPCHandler.isRequestCompleted(
+            CnToDnAsyncRequestType.SET_TTL, status(TSStatusCode.REGION_ALREADY_EXISTS)));
+  }
+
+  private static TSStatus status(TSStatusCode statusCode) {
+    return new TSStatus(statusCode.getStatusCode());
+  }
+}