Pipe: Fix asynchronous connector manager leak when the receiverAttrib…

…utes exists on constructing (apache#13501)
YC27 · Sep 13, 2024 · bc2cc04 · bc2cc04
1 parent 7217575
commit bc2cc04
Showing 1 changed file with 7 additions and 5 deletions.
diff --git a/.../main/java/org/apache/iotdb/db/pipe/connector/client/IoTDBDataNodeAsyncClientManager.java b/.../main/java/org/apache/iotdb/db/pipe/connector/client/IoTDBDataNodeAsyncClientManager.java
@@ -89,11 +89,13 @@ public IoTDBDataNodeAsyncClientManager(
     receiverAttributes =
         String.format("%s-%s", shouldReceiverConvertOnTypeMismatch, loadTsFileStrategy);
     synchronized (IoTDBDataNodeAsyncClientManager.class) {
-      ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER_HOLDER.putIfAbsent(
-          receiverAttributes,
-          new IClientManager.Factory<TEndPoint, AsyncPipeDataTransferServiceClient>()
-              .createClientManager(
-                  new ClientPoolFactory.AsyncPipeDataTransferServiceClientPoolFactory()));
+      if (!ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER_HOLDER.containsKey(receiverAttributes)) {
+        ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER_HOLDER.putIfAbsent(
+            receiverAttributes,
+            new IClientManager.Factory<TEndPoint, AsyncPipeDataTransferServiceClient>()
+                .createClientManager(
+                    new ClientPoolFactory.AsyncPipeDataTransferServiceClientPoolFactory()));
+      }
       endPoint2Client = ASYNC_PIPE_DATA_TRANSFER_CLIENT_MANAGER_HOLDER.get(receiverAttributes);
 
       RECEIVER_ATTRIBUTES_REF_COUNT.compute(