(iotdb) branch dev/1.3 updated: [To dev/1.3] Pipe: Fixed the issue that the DropPipe Connector subtask process waits too long for the callback thread to end (#15605) (#15638)

[justinchen]

This is an automated email from the ASF dual-hosted git repository.

justinchen pushed a commit to branch dev/1.3
in repository https://gitbox.apache.org/repos/asf/iotdb.git


The following commit(s) were added to refs/heads/dev/1.3 by this push:
     new 695130dd9aa [To dev/1.3] Pipe: Fixed the issue that the DropPipe 
Connector subtask process waits too long for the callback thread to end 
(#15605) (#15638)
695130dd9aa is described below

commit 695130dd9aa528b7b916566bd6fd71c6cefcfba5
Author: Zhenyu Luo <luoluoyuyu2...@gmail.com>
AuthorDate: Thu Jun 5 10:09:28 2025 +0800

    [To dev/1.3] Pipe: Fixed the issue that the DropPipe Connector subtask 
process waits too long for the callback thread to end (#15605) (#15638)
    
    (cherry picked from commit 764912c59ce5d41d7a74823dca04972f6380903b)
---
 .../pipe/agent/task/subtask/connector/PipeConnectorSubtask.java  | 4 ++++
 .../pipe/agent/task/subtask/PipeAbstractConnectorSubtask.java    | 9 ++++++++-
 2 files changed, 12 insertions(+), 1 deletion(-)

diff --git 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/connector/PipeConnectorSubtask.java
 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/connector/PipeConnectorSubtask.java
index f316e2cf462..5e49b76b240 100644
--- 
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/connector/PipeConnectorSubtask.java
+++ 
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/agent/task/subtask/connector/PipeConnectorSubtask.java
@@ -234,6 +234,10 @@ public class PipeConnectorSubtask extends 
PipeAbstractConnectorSubtask {
 
     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
diff --git 
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractConnectorSubtask.java
 
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractConnectorSubtask.java
index 055a546e7e3..0ca1f1be8d9 100644
--- 
a/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractConnectorSubtask.java
+++ 
b/iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/pipe/agent/task/subtask/PipeAbstractConnectorSubtask.java
@@ -175,7 +175,14 @@ public abstract class PipeAbstractConnectorSubtask extends 
PipeReportableSubtask
             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.",