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 666ccd54dab [To dev/1.3] Pipe: Fix unnecessary client shutdown logic
and reduce handshake waiting time (#16341) (#16381)
666ccd54dab is described below
commit 666ccd54dabe4d86948e69a6ea976c216f41f7d9
Author: Zhenyu Luo <[email protected]>
AuthorDate: Thu Sep 11 09:51:46 2025 +0800
[To dev/1.3] Pipe: Fix unnecessary client shutdown logic and reduce
handshake waiting time (#16341) (#16381)
* Pipe: Fix unnecessary client shutdown logic and reduce handshake waiting
time
* spotless
* update
(cherry picked from commit d29e45cad2994d62d2dc84e347c81e9ba96bac85)
---
.../pipe/sink/client/IoTDBDataNodeAsyncClientManager.java | 15 ++++++++++-----
1 file changed, 10 insertions(+), 5 deletions(-)
diff --git
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
index 37480a844d0..465f1e1cb8e 100644
---
a/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
+++
b/iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java
@@ -252,6 +252,9 @@ public class IoTDBDataNodeAsyncClientManager extends
IoTDBClientManager
}
isHandshakeFinished.set(true);
+ synchronized (isHandshakeFinished) {
+ isHandshakeFinished.notifyAll();
+ }
}
@Override
@@ -265,6 +268,9 @@ public class IoTDBDataNodeAsyncClientManager extends
IoTDBClientManager
exception.set(e);
isHandshakeFinished.set(true);
+ synchronized (isHandshakeFinished) {
+ isHandshakeFinished.notifyAll();
+ }
}
};
@@ -350,14 +356,13 @@ public class IoTDBDataNodeAsyncClientManager extends
IoTDBClientManager
private void waitHandshakeFinished(final AtomicBoolean isHandshakeFinished) {
try {
- final long startTime = System.currentTimeMillis();
while (!isHandshakeFinished.get()) {
- if (isClosed
- || System.currentTimeMillis() - startTime
- >
PipeConfig.getInstance().getPipeConnectorHandshakeTimeoutMs() * 2L) {
+ if (isClosed) {
throw new PipeConnectionException("Timed out when waiting for client
handshake finish.");
}
- Thread.sleep(10);
+ synchronized (isHandshakeFinished) {
+ isHandshakeFinished.wait(1);
+ }
}
} catch (final InterruptedException e) {
Thread.currentThread().interrupt();