Fix deadlock between DataNode createDataRegion and ConfigNode PipeTaskCoordinatorLock by delegating consensus pipe lifecycle to ConfigNode (#17233)

* Fix deadlock between DataNode createDataRegion and ConfigNode PipeTaskCoordinatorLock by delegating consensus pipe lifecycle to ConfigNode

* CN get IoTV2 replication mode

* remove duplicated code

* spotless

* fix not create consensus pipe when initially create data region

* Async create pipe to avoid audit log dead lock

* fix PipeTaskCoordinator can not be shared cross thread

* fix DN CN restart at the same time with data race and DN where region was removed failed to clear region file

* refine code

* remove useless comment

* unlock for PipeTaskCoordinator no longer return

Author: Pengzna

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeConfig.java

@@ -77,6 +77,9 @@ public class ConfigNodeConfig {
   /** Data region consensus protocol. */
   private String dataRegionConsensusProtocolClass = ConsensusFactory.IOT_CONSENSUS;
 
+  /** IoTConsensusV2 replicate mode: "batch" or "stream". */
+  private String iotConsensusV2Mode = "batch";
+
   /** Default number of DataRegion replicas. */
   private int dataReplicationFactor = 1;
 
@@ -530,6 +533,14 @@ public void setDataRegionConsensusProtocolClass(String dataRegionConsensusProtoc
     this.dataRegionConsensusProtocolClass = dataRegionConsensusProtocolClass;
   }
 
+  public String getIotConsensusV2Mode() {
+    return iotConsensusV2Mode;
+  }
+
+  public void setIotConsensusV2Mode(String iotConsensusV2Mode) {
+    this.iotConsensusV2Mode = iotConsensusV2Mode;
+  }
+
   public int getDataRegionPerDataNode() {
     return dataRegionPerDataNode;
   }

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/conf/ConfigNodeDescriptor.java

@@ -205,6 +205,9 @@ private void loadProperties(TrimProperties properties) throws BadNodeUrlExceptio
         properties.getProperty(
             "data_region_consensus_protocol_class", conf.getDataRegionConsensusProtocolClass()));
 
+    conf.setIotConsensusV2Mode(
+        properties.getProperty("iot_consensus_v2_mode", conf.getIotConsensusV2Mode()));
+
     conf.setDataReplicationFactor(
         Integer.parseInt(
             properties.getProperty(

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

@@ -1485,6 +1485,23 @@ public TSStatus createConsensusPipe(TCreatePipeReq req) {
     }
   }
 
+  /**
+   * Submit a consensus pipe creation procedure without blocking. The procedure will execute in the
+   * background. Failures are logged and can be repaired by the consensus pipe guardian.
+   */
+  public void createConsensusPipeAsync(TCreatePipeReq req) {
+    try {
+      CreatePipeProcedureV2 procedure = new CreatePipeProcedureV2(req);
+      executor.submitProcedure(procedure);
+      LOGGER.info("Submitted async consensus pipe creation: {}", req.getPipeName());
+    } catch (Exception e) {
+      LOGGER.warn(
+          "Failed to submit async consensus pipe creation for {}: {}",
+          req.getPipeName(),
+          e.getMessage());
+    }
+  }
+
   public TSStatus createPipe(TCreatePipeReq req) {
     try {
       CreatePipeProcedureV2 procedure = new CreatePipeProcedureV2(req);
@@ -1579,6 +1596,21 @@ public TSStatus dropConsensusPipe(String pipeName) {
     }
   }
 
+  /**
+   * Submit a consensus pipe drop procedure without blocking. The procedure will execute in the
+   * background. Failures are logged and can be repaired by the consensus pipe guardian.
+   */
+  public void dropConsensusPipeAsync(String pipeName) {
+    try {
+      DropPipeProcedureV2 procedure = new DropPipeProcedureV2(pipeName);
+      executor.submitProcedure(procedure);
+      LOGGER.info("Submitted async consensus pipe drop: {}", pipeName);
+    } catch (Exception e) {
+      LOGGER.warn(
+          "Failed to submit async consensus pipe drop for {}: {}", pipeName, e.getMessage());
+    }
+  }
+
   public TSStatus dropPipe(String pipeName) {
     try {
       DropPipeProcedureV2 procedure = new DropPipeProcedureV2(pipeName);

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinator.java

@@ -85,19 +85,9 @@ public AtomicReference<PipeTaskInfo> lock() {
   /**
    * Unlock the pipe task coordinator. Calling this method will clear the pipe task info holder,
    * which means that the holder will be null after calling this method.
-   *
-   * @return {@code true} if successfully unlocked, {@code false} if current thread is not holding
-   *     the lock.
    */
-  public boolean unlock() {
-    try {
-      pipeTaskCoordinatorLock.unlock();
-      return true;
-    } catch (IllegalMonitorStateException ignored) {
-      // This is thrown if unlock() is called without lock() called first.
-      LOGGER.warn("This thread is not holding the lock.");
-      return false;
-    }
+  public void unlock() {
+    pipeTaskCoordinatorLock.unlock();
   }
 
   public boolean isLocked() {

iotdb-core/confignode/src/main/java/org/apache/iotdb/confignode/manager/pipe/coordinator/task/PipeTaskCoordinatorLock.java

@@ -22,24 +22,29 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.concurrent.Semaphore;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.locks.ReentrantLock;
 
 /**
- * {@link PipeTaskCoordinatorLock} is a cross thread lock for pipe task coordinator. It is used to
+ * {@link PipeTaskCoordinatorLock} is a cross-thread lock for pipe task coordinator. It is used to
  * ensure that only one thread can execute the pipe task coordinator at the same time.
+ *
+ * <p>Uses {@link Semaphore} instead of {@link java.util.concurrent.locks.ReentrantLock} to support
+ * cross-thread acquire/release, which is required by the procedure recovery mechanism: locks may be
+ * acquired on the StateMachineUpdater thread during {@code restoreLock()} and released on a
+ * ProcedureCoreWorker thread after execution.
  */
 public class PipeTaskCoordinatorLock {
 
   private static final Logger LOGGER = LoggerFactory.getLogger(PipeTaskCoordinatorLock.class);
 
-  private final ReentrantLock lock = new ReentrantLock();
+  private final Semaphore semaphore = new Semaphore(1);
 
   public void lock() {
     LOGGER.debug(
         "PipeTaskCoordinator lock waiting for thread {}", Thread.currentThread().getName());
     try {
-      lock.lockInterruptibly();
+      semaphore.acquire();
       LOGGER.debug(
           "PipeTaskCoordinator lock acquired by thread {}", Thread.currentThread().getName());
     } catch (final InterruptedException e) {
@@ -54,7 +59,7 @@ public boolean tryLock() {
     try {
       LOGGER.debug(
           "PipeTaskCoordinator lock waiting for thread {}", Thread.currentThread().getName());
-      if (lock.tryLock(10, TimeUnit.SECONDS)) {
+      if (semaphore.tryAcquire(10, TimeUnit.SECONDS)) {
         LOGGER.debug(
             "PipeTaskCoordinator lock acquired by thread {}", Thread.currentThread().getName());
         return true;
@@ -74,12 +79,12 @@ public boolean tryLock() {
   }
 
   public void unlock() {
-    lock.unlock();
+    semaphore.release();
     LOGGER.debug(
         "PipeTaskCoordinator lock released by thread {}", Thread.currentThread().getName());
   }
 
   public boolean isLocked() {
-    return lock.isLocked();
+    return semaphore.availablePermits() == 0;
   }
 }

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

@@ -105,14 +105,8 @@ public boolean unlock() {
       subscriptionInfoHolder = null;
     }
 
-    try {
-      coordinatorLock.unlock();
-      return true;
-    } catch (IllegalMonitorStateException ignored) {
-      // This is thrown if unlock() is called without lock() called first.
-      LOGGER.warn("This thread is not holding the lock.");
-      return false;
-    }
+    coordinatorLock.unlock();
+    return true;
   }
 
   public boolean isLocked() {