Pipe: Avoid unnecessary close-client in async client (Follow up fix for #16008)

Author: Caideyipi

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/IoTDBDataRegionAsyncSink.java

@@ -756,7 +756,7 @@ public synchronized void close() {
           .forEach(
               handler -> {
                 handler.clearEventsReferenceCount();
-                eliminateHandler(handler);
+                eliminateHandler(handler, true);
               });
     }
 
@@ -813,7 +813,11 @@ public void trackHandler(final PipeTransferTrackableHandler handler) {
     pendingHandlers.put(handler, handler);
   }
 
-  public void eliminateHandler(final PipeTransferTrackableHandler handler) {
+  public void eliminateHandler(
+      final PipeTransferTrackableHandler handler, final boolean closeClient) {
+    if (closeClient) {
+      handler.closeClient();
+    }
     handler.close();
     pendingHandlers.remove(handler);
   }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTrackableHandler.java

@@ -46,7 +46,7 @@ public PipeTransferTrackableHandler(final IoTDBDataRegionAsyncSink connector) {
   public void onComplete(final TPipeTransferResp response) {
     if (connector.isClosed()) {
       clearEventsReferenceCount();
-      connector.eliminateHandler(this);
+      connector.eliminateHandler(this, true);
       return;
     }
 
@@ -55,20 +55,20 @@ public void onComplete(final TPipeTransferResp response) {
       // completed
       // NOTE: We should not clear the reference count of events, as this would cause the
       // `org.apache.iotdb.pipe.it.dual.tablemodel.manual.basic.IoTDBPipeDataSinkIT#testSinkTsFileFormat3` test to fail.
-      connector.eliminateHandler(this);
+      connector.eliminateHandler(this, false);
     }
   }
 
   @Override
   public void onError(final Exception exception) {
     if (connector.isClosed()) {
       clearEventsReferenceCount();
-      connector.eliminateHandler(this);
+      connector.eliminateHandler(this, true);
       return;
     }
 
     onErrorInternal(exception);
-    connector.eliminateHandler(this);
+    connector.eliminateHandler(this, false);
   }
 
   /**
@@ -90,7 +90,7 @@ protected boolean tryTransfer(
     connector.trackHandler(this);
     if (connector.isClosed()) {
       clearEventsReferenceCount();
-      connector.eliminateHandler(this);
+      connector.eliminateHandler(this, true);
       client.setShouldReturnSelf(true);
       try {
         client.returnSelf();
@@ -119,8 +119,7 @@ protected abstract void doTransfer(
 
   public abstract void clearEventsReferenceCount();
 
-  @Override
-  public void close() {
+  public void closeClient() {
     if (Objects.isNull(client)) {
       return;
     }
@@ -135,4 +134,9 @@ public void close() {
           e);
     }
   }
+
+  @Override
+  public void close() {
+    // Do nothing
+  }
 }

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/protocol/thrift/async/handler/PipeTransferTsFileHandler.java

@@ -418,7 +418,7 @@ private void returnClientIfNecessary() {
     }
 
     if (connector.isClosed()) {
-      close();
+      closeClient();
     }
 
     client.setShouldReturnSelf(true);