Pipe: Fixed the issue that the DropPipe Connector subtask process waits too long for the callback thread to end (apache#15605)

Author: luoluoyuyu

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

@@ -234,6 +234,10 @@ public void discardEventsOfPipe(final String pipeNameToDrop, int regionId) {
 
     highPriorityLockTaskCount.incrementAndGet();
     try {
+      synchronized (highPriorityLockTaskCount) {
+        highPriorityLockTaskCount.notifyAll();
+      }
+
       // synchronized to use the lastEvent & lastExceptionEvent
       synchronized (this) {
         // Here we discard the last event, and re-submit the pipe task to avoid that the pipe task

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

@@ -175,7 +175,14 @@ private boolean onPipeConnectionException(final Throwable throwable) {
             MAX_RETRY_TIMES,
             e);
         try {
-          Thread.sleep(retry * PipeConfig.getInstance().getPipeConnectorRetryIntervalMs());
+          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());
+            }
+          }
         } catch (final InterruptedException interruptedException) {
           LOGGER.info(
               "Interrupted while sleeping, will retry to handshake with the target system.",