[ConfigNode] Fence Region creation across database deletion

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

@@ -665,4 +665,16 @@ private ConfigNodeMessages() {}
   public static final String
       EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_MODE_CONSENSUS_DOES_NOT_SUPPORT_TOPIC_ATTRIBUTES_ARG_3C2D0BDA =
           "Failed to create or alter topic, mode=consensus does not support topic attributes %s";
+  public static final String
+      LOG_REJECT_CREATEREGIONGROUPSPLAN_BECAUSE_DATABASE_ARG_DOES_NOT_EXIST_616E0CDE =
+          "Reject CreateRegionGroupsPlan because database {} does not exist";
+  public static final String
+      LOG_REJECT_CREATEREGIONGROUPSPLAN_BECAUSE_DATABASE_ARG_IS_BEING_DELETED_C085AC01 =
+          "Reject CreateRegionGroupsPlan because database {} is being deleted";
+  public static final String
+      MESSAGE_CREATE_REGIONGROUPS_FAILED_BECAUSE_DATABASE_ARG_DOES_NOT_EXIST_AF0F2440 =
+          "Create RegionGroups failed because database %s does not exist";
+  public static final String
+      MESSAGE_CREATE_REGIONGROUPS_FAILED_BECAUSE_DATABASE_ARG_IS_BEING_DELETED_651DB780 =
+          "Create RegionGroups failed because database %s is being deleted";
 }

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

@@ -522,8 +522,6 @@ public final class ManagerMessages {
       "Unexpected interruption during waiting for configNode leader ready.";
   public static final String UNEXPECTED_INTERRUPTION_DURING_WAITING_FOR_GET_CLUSTER_ID =
       "Unexpected interruption during waiting for get cluster id.";
-  public static final String UNEXPECTED_NON_CREATE_REGION_MAINTAIN_TASK_SKIPPED =
-      "Unexpected non-create task in the RegionMaintainer queue; skipping it (the queue only recreates region replicas now, and region deletion is handled by RemoveRegionGroupProcedure).";
   public static final String UNEXPECTED_NULL_PROCEDURE_PARAMETERS_FOR_WAITINGPROCEDUREFINISHED =
       "Unexpected null procedure parameters for waitingProcedureFinished";
   public static final String UNKNOWN_DATAPARTITION_ALLOCATION_STRATEGY_USING_INHERIT_STRATEGY_BY_DEFAULT =
@@ -587,6 +585,11 @@ private ManagerMessages() {}
   public static final String MESSAGE_SCHEMA_ENGINE_MODE_E37ED98C = "schema_engine_mode";
   public static final String MESSAGE_TAG_ATTRIBUTE_TOTAL_SIZE_AF658CFE = "tag_attribute_total_size";
   public static final String MESSAGE_DATABASE_LIMIT_THRESHOLD_45C23274 = "database_limit_threshold";
+  public static final String
+      MESSAGE_DATABASE_ARG_STILL_HAS_UNFINISHED_LIFECYCLE_PROCEDURES_67573924 =
+          "Database %s still has unfinished lifecycle procedures";
+  public static final String MESSAGE_SOME_OTHER_TASK_IS_DELETING_DATABASE_ARG_7BDB2C0F =
+      "Some other task is deleting database %s";
   public static final String LOG_UNEXPECTED_ERROR_HAPPENED_SETTING_SPACE_QUOTA_DATABASE_ARG_F6ED7586 = "Unexpected error happened while setting space quota on database: %s ";
   public static final String LOG_UNEXPECTED_ERROR_HAPPENED_SETTING_THROTTLE_QUOTA_USER_ARG_C111BE81 = "Unexpected error happened while setting throttle quota on user: %s ";
   public static final String LOG_SCHEMA_TEMPLATE_NEED_TWO_FILES_1E57542A = "schema_template need two files";

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

@@ -710,4 +710,16 @@ private ConfigNodeMessages() {}
   public static final String
       EXCEPTION_FAILED_TO_CREATE_OR_ALTER_TOPIC_MODE_CONSENSUS_DOES_NOT_SUPPORT_TOPIC_ATTRIBUTES_ARG_3C2D0BDA =
           "创建或修改 topic 失败，mode=consensus 不支持 topic 属性 %s";
+  public static final String
+      LOG_REJECT_CREATEREGIONGROUPSPLAN_BECAUSE_DATABASE_ARG_DOES_NOT_EXIST_616E0CDE =
+          "拒绝 CreateRegionGroupsPlan，因为数据库 {} 不存在";
+  public static final String
+      LOG_REJECT_CREATEREGIONGROUPSPLAN_BECAUSE_DATABASE_ARG_IS_BEING_DELETED_C085AC01 =
+          "拒绝 CreateRegionGroupsPlan，因为数据库 {} 正在删除";
+  public static final String
+      MESSAGE_CREATE_REGIONGROUPS_FAILED_BECAUSE_DATABASE_ARG_DOES_NOT_EXIST_AF0F2440 =
+          "创建 RegionGroups 失败，因为数据库 %s 不存在";
+  public static final String
+      MESSAGE_CREATE_REGIONGROUPS_FAILED_BECAUSE_DATABASE_ARG_IS_BEING_DELETED_651DB780 =
+          "创建 RegionGroups 失败，因为数据库 %s 正在删除";
 }

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

@@ -517,8 +517,6 @@ public final class ManagerMessages {
       "等待 configNode leader 就绪过程中发生意外中断。";
   public static final String UNEXPECTED_INTERRUPTION_DURING_WAITING_FOR_GET_CLUSTER_ID =
       "等待获取 cluster id 过程中发生意外中断。";
-  public static final String UNEXPECTED_NON_CREATE_REGION_MAINTAIN_TASK_SKIPPED =
-      "RegionMaintainer 队列中出现意外的非 create 任务；跳过处理（该队列目前仅用于重建 region 副本，region 删除由 RemoveRegionGroupProcedure 处理）。";
   public static final String UNEXPECTED_NULL_PROCEDURE_PARAMETERS_FOR_WAITINGPROCEDUREFINISHED =
       "waitingProcedureFinished 的 procedure 参数为空";
   public static final String UNKNOWN_DATAPARTITION_ALLOCATION_STRATEGY_USING_INHERIT_STRATEGY_BY_DEFAULT =
@@ -579,6 +577,11 @@ private ManagerMessages() {}
   public static final String MESSAGE_SCHEMA_ENGINE_MODE_E37ED98C = "schema_engine_mode";
   public static final String MESSAGE_TAG_ATTRIBUTE_TOTAL_SIZE_AF658CFE = "tag_attribute_total_size";
   public static final String MESSAGE_DATABASE_LIMIT_THRESHOLD_45C23274 = "database_limit_threshold";
+  public static final String
+      MESSAGE_DATABASE_ARG_STILL_HAS_UNFINISHED_LIFECYCLE_PROCEDURES_67573924 =
+          "数据库 %s 仍有未完成的生命周期流程";
+  public static final String MESSAGE_SOME_OTHER_TASK_IS_DELETING_DATABASE_ARG_7BDB2C0F =
+      "其他任务正在删除数据库 %s";
   public static final String LOG_UNEXPECTED_ERROR_HAPPENED_SETTING_SPACE_QUOTA_DATABASE_ARG_F6ED7586 = "设置数据库 %s 的空间配额时发生意外错误 ";
   public static final String LOG_UNEXPECTED_ERROR_HAPPENED_SETTING_THROTTLE_QUOTA_USER_ARG_C111BE81 = "设置用户 %s 的限流配额时发生意外错误 ";
   public static final String LOG_SCHEMA_TEMPLATE_NEED_TWO_FILES_1E57542A = "schema_template 需要两个文件";

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlan.java

@@ -85,6 +85,7 @@
 import org.apache.iotdb.confignode.consensus.request.write.procedure.UpdateProcedurePlan;
 import org.apache.iotdb.confignode.consensus.request.write.quota.SetSpaceQuotaPlan;
 import org.apache.iotdb.confignode.consensus.request.write.quota.SetThrottleQuotaPlan;
+import org.apache.iotdb.confignode.consensus.request.write.region.BatchRemoveRegionCreateTasksPlan;
 import org.apache.iotdb.confignode.consensus.request.write.region.CreateRegionGroupsPlan;
 import org.apache.iotdb.confignode.consensus.request.write.region.OfferRegionMaintainTasksPlan;
 import org.apache.iotdb.confignode.consensus.request.write.region.PollRegionMaintainTaskPlan;
@@ -266,6 +267,9 @@ public static ConfigPhysicalPlan create(final ByteBuffer buffer) throws IOExcept
         case PollSpecificRegionMaintainTask:
           plan = new PollSpecificRegionMaintainTaskPlan();
           break;
+        case BatchRemoveRegionCreateTasks:
+          plan = new BatchRemoveRegionCreateTasksPlan();
+          break;
         case CreateSchemaPartition:
           plan = new CreateSchemaPartitionPlan();
           break;

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/ConfigPhysicalPlanType.java

@@ -72,6 +72,7 @@
   AddRegionLocation((short) 311),
   RemoveRegionLocation((short) 312),
   GetRegionGroupsByTime((short) 313),
+  BatchRemoveRegionCreateTasks((short) 314),
 
   /** Partition. */
   GetSchemaPartition((short) 400),

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/consensus/request/write/region/BatchRemoveRegionCreateTasksPlan.java

@@ -0,0 +1,82 @@
+/*
+ * 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.consensus.request.write.region;
+
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlan;
+import org.apache.iotdb.confignode.consensus.request.ConfigPhysicalPlanType;
+
+import org.apache.tsfile.utils.ReadWriteIOUtils;
+
+import java.io.DataOutputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Objects;
+
+/**
+ * Removes every queued RegionCreateTask that belongs to the specified pre-deleted database.
+ *
+ * <p>The state machine ignores this plan when the database is missing or active, so replaying a
+ * cancellation cannot affect a later database incarnation that reuses the same name.
+ */
+public class BatchRemoveRegionCreateTasksPlan extends ConfigPhysicalPlan {
+
+  private String database;
+
+  public BatchRemoveRegionCreateTasksPlan() {
+    super(ConfigPhysicalPlanType.BatchRemoveRegionCreateTasks);
+  }
+
+  public BatchRemoveRegionCreateTasksPlan(final String database) {
+    super(ConfigPhysicalPlanType.BatchRemoveRegionCreateTasks);
+    this.database = database;
+  }
+
+  public String getDatabase() {
+    return database;
+  }
+
+  @Override
+  protected void serializeImpl(DataOutputStream stream) throws IOException {
+    stream.writeShort(getType().getPlanType());
+    ReadWriteIOUtils.write(database, stream);
+  }
+
+  @Override
+  protected void deserializeImpl(ByteBuffer buffer) throws IOException {
+    database = ReadWriteIOUtils.readString(buffer);
+  }
+
+  @Override
+  public boolean equals(Object o) {
+    if (this == o) {
+      return true;
+    }
+    if (!(o instanceof BatchRemoveRegionCreateTasksPlan)) {
+      return false;
+    }
+    final BatchRemoveRegionCreateTasksPlan that = (BatchRemoveRegionCreateTasksPlan) o;
+    return Objects.equals(database, that.database);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(database);
+  }
+}

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

@@ -134,6 +134,7 @@
 import org.apache.iotdb.confignode.procedure.impl.testonly.CreateManyDatabasesProcedure;
 import org.apache.iotdb.confignode.procedure.impl.trigger.CreateTriggerProcedure;
 import org.apache.iotdb.confignode.procedure.impl.trigger.DropTriggerProcedure;
+import org.apache.iotdb.confignode.procedure.scheduler.DatabaseLockQueue.DatabaseLock;
 import org.apache.iotdb.confignode.procedure.scheduler.ProcedureScheduler;
 import org.apache.iotdb.confignode.procedure.scheduler.SimpleProcedureScheduler;
 import org.apache.iotdb.confignode.procedure.store.ConfigProcedureStore;
@@ -183,6 +184,7 @@
 import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.Arrays;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.LinkedHashSet;
 import java.util.List;
@@ -191,6 +193,7 @@
 import java.util.Optional;
 import java.util.Set;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.BiFunction;
 import java.util.stream.Collectors;
@@ -301,50 +304,96 @@
   public TSStatus deleteDatabases(
       final List<TDatabaseSchema> deleteSgSchemaList, final boolean isGeneratedByPipe) {
     final List<DeleteDatabaseProcedure> procedures = new ArrayList<>();
-    final long startCheckTimeForProcedures = System.currentTimeMillis();
+    final long deadlineNanos =
+        System.nanoTime() + TimeUnit.MILLISECONDS.toNanos(PROCEDURE_WAIT_TIME_OUT);
     for (final TDatabaseSchema databaseSchema : deleteSgSchemaList) {
-      final String database = databaseSchema.getName();
-      boolean hasOverlappedTask = false;
-      synchronized (this) {
-        while (executor.isRunning()
-            && System.currentTimeMillis() - startCheckTimeForProcedures < PROCEDURE_WAIT_TIME_OUT) {
+      if (!submitDeleteDatabaseProcedure(
+          databaseSchema, isGeneratedByPipe, procedures, deadlineNanos)) {
+        return getDeleteDatabaseOverlapStatus(databaseSchema.getName());
+      }
+    }
+
+    final List<TSStatus> results = new ArrayList<>(procedures.size());
+    procedures.forEach(procedure -> results.add(waitingProcedureFinished(procedure)));
+    if (results.stream()
+        .allMatch(result -> result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode())) {
+      return StatusUtils.OK;
+    }
+    return RpcUtils.getStatus(results);
+  }
+
+  private boolean submitDeleteDatabaseProcedure(
+      final TDatabaseSchema databaseSchema,
+      final boolean isGeneratedByPipe,
+      final List<DeleteDatabaseProcedure> procedures,
+      final long deadlineNanos) {
+    while (executor.isRunning()) {
+      final long remainingNanos = deadlineNanos - System.nanoTime();
+      if (remainingNanos <= 0) {
+        return false;
+      }
+
+      final DatabaseLock databaseLock;
+      try {
+        databaseLock =
+            env.getDatabaseLockQueue()
+                .tryAcquireLocks(
+                    Collections.singleton(databaseSchema.getName()),
+                    remainingNanos,
+                    TimeUnit.NANOSECONDS);
+      } catch (final InterruptedException e) {
+        Thread.currentThread().interrupt();
+        return false;
+      }
+      if (databaseLock == null) {
+        return false;
+      }
+
+      try (databaseLock) {
+        synchronized (this) {
           final Pair<Long, Boolean> procedureIdDuplicatePair =
               checkDuplicateTableTask(
-                  database, null, null, null, null, ProcedureType.DELETE_DATABASE_PROCEDURE);
-          hasOverlappedTask = procedureIdDuplicatePair.getRight();
-
+                  databaseSchema.getName(),
+                  null,
+                  null,
+                  null,
+                  null,
+                  ProcedureType.DELETE_DATABASE_PROCEDURE);
           if (Boolean.FALSE.equals(procedureIdDuplicatePair.getRight())) {
-            DeleteDatabaseProcedure procedure =
+            final DeleteDatabaseProcedure procedure =
                 new DeleteDatabaseProcedure(databaseSchema, isGeneratedByPipe);
-            this.executor.submitProcedure(procedure);
+            executor.submitProcedure(procedure);
             procedures.add(procedure);
-            break;
-          }
-          try {
-            wait(PROCEDURE_WAIT_RETRY_TIMEOUT);
-          } catch (final InterruptedException e) {
-            Thread.currentThread().interrupt();
+            return true;
           }
         }
-        if (hasOverlappedTask) {
-          return RpcUtils.getStatus(
-              TSStatusCode.OVERLAP_WITH_EXISTING_TASK,
-              String.format(
-                  "Some other task is operating table under the database %s, please retry after the procedure finishes.",
-                  database));
-        }
+      }
+
+      if (!waitForDeleteDatabaseRetry()) {
+        return false;
       }
     }
-    List<TSStatus> results = new ArrayList<>(procedures.size());
-    procedures.forEach(procedure -> results.add(waitingProcedureFinished(procedure)));
-    if (results.stream()
-        .allMatch(result -> result.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode())) {
-      return StatusUtils.OK;
-    } else {
-      return RpcUtils.getStatus(results);
+    return false;
+  }
+
+  private boolean waitForDeleteDatabaseRetry() {
+    try {
+      TimeUnit.MILLISECONDS.sleep(PROCEDURE_WAIT_RETRY_TIMEOUT);
+      return true;
+    } catch (final InterruptedException e) {
+      Thread.currentThread().interrupt();
+      return false;
     }
   }
 
+  private TSStatus getDeleteDatabaseOverlapStatus(final String database) {
+    return RpcUtils.getStatus(
+        TSStatusCode.OVERLAP_WITH_EXISTING_TASK,
+        String.format(
+            "Some other task is operating table under the database %s, please retry after the procedure finishes.",
+            database));
+  }
+
   public TSStatus alterEncodingCompressor(
       final String queryId,
       final PathPatternTree patternTree,
@@ -1121,7 +1170,7 @@
 
   private String checkRegionOperationModelCorrectness(TConsensusGroupId regionId, Model model) {
     String databaseName = configManager.getPartitionManager().getRegionDatabase(regionId);
     boolean isTreeModelDatabase = databaseName.startsWith(SqlConstant.TREE_MODEL_DATABASE_PREFIX);
     if (Model.TREE == model && isTreeModelDatabase
         || Model.TABLE == model && !isTreeModelDatabase) {
       return null;
@@ -1501,18 +1550,28 @@
 
   // endregion
 
+  public CreateRegionGroupsProcedure submitCreateRegionGroups(
+      final TConsensusGroupType consensusGroupType,
+      final CreateRegionGroupsPlan createRegionGroupsPlan) {
+    final CreateRegionGroupsProcedure procedure =
+        new CreateRegionGroupsProcedure(consensusGroupType, createRegionGroupsPlan);
+    // Reentrant for PartitionManager, which holds the same database locks while allocating the
+    // plan. Keeping this guard here also makes a future direct submission visible atomically to
+    // same-name database creation and deletion.
+    try (final DatabaseLock ignored =
+        acquireDatabaseLifecycleLocks(createRegionGroupsPlan.getRegionGroupMap().keySet())) {
+      executor.submitProcedure(procedure);
+    }
+    return procedure;
+  }
+
   /**
-   * Generate {@link CreateRegionGroupsProcedure} and wait until it finished.
+   * Wait until a {@link CreateRegionGroupsProcedure} finishes.
    *
    * @return {@link TSStatusCode#SUCCESS_STATUS} if all RegionGroups have been created successfully,
    *     {@link TSStatusCode#CREATE_REGION_ERROR} otherwise
    */
-  public TSStatus createRegionGroups(
-      final TConsensusGroupType consensusGroupType,
-      final CreateRegionGroupsPlan createRegionGroupsPlan) {
-    final CreateRegionGroupsProcedure procedure =
-        new CreateRegionGroupsProcedure(consensusGroupType, createRegionGroupsPlan);
-    executor.submitProcedure(procedure);
+  public TSStatus waitCreateRegionGroups(final CreateRegionGroupsProcedure procedure) {
     final TSStatus status = waitingProcedureFinished(procedure);
     if (status.getCode() == TSStatusCode.SUCCESS_STATUS.getStatusCode()) {
       return status;
@@ -1522,6 +1581,25 @@
     }
   }
 
+  public DatabaseLock acquireDatabaseLifecycleLock(final String database) {
+    return acquireDatabaseLifecycleLocks(Collections.singleton(database));
+  }
+
+  public DatabaseLock acquireDatabaseLifecycleLocks(final Set<String> databases) {
+    return env.getDatabaseLockQueue().acquireLocks(databases);
+  }
+
+  public boolean hasUnfinishedDatabaseLifecycleProcedure(final String database) {
+    return executor.getProcedures().values().stream()
+        .filter(procedure -> !procedure.isFinished())
+        .anyMatch(
+            procedure ->
+                (procedure instanceof DeleteDatabaseProcedure deleteProcedure
+                        && database.equals(deleteProcedure.getDatabase()))
+                    || (procedure instanceof CreateRegionGroupsProcedure createProcedure
+                        && createProcedure.containsDatabase(database)));
+  }
+
   /**
    * Generate {@link CreateTriggerProcedure} and wait until it finished.
    *