[To dev/1.3] Pipe: Fix the stuck state caused by unfair lock in Sink start phase (#16100)

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtask.java

@@ -228,11 +228,8 @@ public void discardEventsOfPipe(final String pipeNameToDrop, int regionId) {
     // Try to remove the events as much as possible
     inputPendingQueue.discardEventsOfPipe(pipeNameToDrop, regionId);
 
-    highPriorityLockTaskCount.incrementAndGet();
     try {
-      synchronized (highPriorityLockTaskCount) {
-        highPriorityLockTaskCount.notifyAll();
-      }
+      increaseHighPriorityTaskCount();
 
       // synchronized to use the lastEvent & lastExceptionEvent
       synchronized (this) {
@@ -271,7 +268,7 @@ public void discardEventsOfPipe(final String pipeNameToDrop, int regionId) {
         }
       }
     } finally {
-      highPriorityLockTaskCount.decrementAndGet();
+      decreaseHighPriorityTaskCount();
     }
 
     if (outputPipeConnector instanceof IoTDBSink) {

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/sink/PipeSinkSubtaskLifeCycle.java

@@ -123,7 +123,12 @@ public synchronized void start() {
     }
 
     if (runningTaskCount == 0) {
-      executor.start(subtask.getTaskID());
+      try {
+        subtask.increaseHighPriorityTaskCount();
+        executor.start(subtask.getTaskID());
+      } finally {
+        subtask.decreaseHighPriorityTaskCount();
+      }
     }
 
     runningTaskCount++;

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractSinkSubtask.java

@@ -170,14 +170,8 @@ private boolean onPipeConnectionException(final Throwable throwable) {
             MAX_RETRY_TIMES,
             e);
         try {
-          synchronized (highPriorityLockTaskCount) {
-            // The wait operation will release the highPriorityLockTaskCount lock, so there will be
-            // no deadlock.
-            if (highPriorityLockTaskCount.get() == 0) {
-              highPriorityLockTaskCount.wait(
-                  retry * PipeConfig.getInstance().getPipeConnectorRetryIntervalMs());
-            }
-          }
+          sleepIfNoHighPriorityTask(
+              retry * PipeConfig.getInstance().getPipeConnectorRetryIntervalMs());
         } catch (final InterruptedException interruptedException) {
           LOGGER.info(
               "Interrupted while sleeping, will retry to handshake with the target system.",
@@ -254,17 +248,4 @@ protected synchronized void clearReferenceCountAndReleaseLastExceptionEvent() {
       lastExceptionEvent = null;
     }
   }
-
-  private void preScheduleLowPriorityTask(int maxRetries) {
-    while (highPriorityLockTaskCount.get() != 0L && maxRetries-- > 0) {
-      try {
-        // Introduce a short delay to avoid CPU spinning
-        Thread.sleep(10);
-      } catch (InterruptedException e) {
-        Thread.currentThread().interrupt();
-        LOGGER.warn("Interrupted while waiting for the high priority lock task.", e);
-        break;
-      }
-    }
-  }
 }

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeReportableSubtask.java

@@ -90,14 +90,8 @@ private void onEnrichedEventFailure(final Throwable throwable) {
           throwable.getMessage(),
           throwable);
       try {
-        synchronized (highPriorityLockTaskCount) {
-          // The wait operation will release the highPriorityLockTaskCount lock, so there will be
-          // no deadlock.
-          if (highPriorityLockTaskCount.get() == 0) {
-            highPriorityLockTaskCount.wait(
-                retryCount.get() * PipeConfig.getInstance().getPipeConnectorRetryIntervalMs());
-          }
-        }
+        sleepIfNoHighPriorityTask(
+            retryCount.get() * PipeConfig.getInstance().getPipeConnectorRetryIntervalMs());
       } catch (final InterruptedException e) {
         LOGGER.warn(
             "Interrupted when retrying to execute subtask {} (creation time: {}, simple class: {})",
@@ -164,14 +158,8 @@ private void onNonEnrichedEventFailure(final Throwable throwable) {
         throwable.getMessage(),
         throwable);
     try {
-      synchronized (highPriorityLockTaskCount) {
-        // The wait operation will release the highPriorityLockTaskCount lock, so there will be
-        // no deadlock.
-        if (highPriorityLockTaskCount.get() == 0) {
-          highPriorityLockTaskCount.wait(
-              retryCount.get() * PipeConfig.getInstance().getPipeConnectorRetryIntervalMs());
-        }
-      }
+      sleepIfNoHighPriorityTask(
+          retryCount.get() * PipeConfig.getInstance().getPipeConnectorRetryIntervalMs());
     } catch (final InterruptedException e) {
       LOGGER.warn(
           "Interrupted when retrying to execute subtask {} (creation time: {}, simple class: {})",
@@ -183,4 +171,38 @@ private void onNonEnrichedEventFailure(final Throwable throwable) {
 
     submitSelf();
   }
+
+  protected void preScheduleLowPriorityTask(int maxRetries) {
+    while (highPriorityLockTaskCount.get() != 0L && maxRetries-- > 0) {
+      try {
+        // Introduce a short delay to avoid CPU spinning
+        Thread.sleep(10);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+        LOGGER.warn("Interrupted while waiting for the high priority lock task.", e);
+        break;
+      }
+    }
+  }
+
+  protected void sleepIfNoHighPriorityTask(long sleepMillis) throws InterruptedException {
+    synchronized (highPriorityLockTaskCount) {
+      // The wait operation will release the highPriorityLockTaskCount lock, so there will be
+      // no deadlock.
+      if (highPriorityLockTaskCount.get() > 0) {
+        highPriorityLockTaskCount.wait(sleepMillis);
+      }
+    }
+  }
+
+  public void increaseHighPriorityTaskCount() {
+    highPriorityLockTaskCount.incrementAndGet();
+    synchronized (highPriorityLockTaskCount) {
+      highPriorityLockTaskCount.notifyAll();
+    }
+  }
+
+  public void decreaseHighPriorityTaskCount() {
+    highPriorityLockTaskCount.decrementAndGet();
+  }
 }