Pipe: Add Thrift callback processing logic for ClosedChannelException (apache#16421)

* Add Thrift callback processing logic for ClosedChannelException

* update

* fix

* fix

Author: luoluoyuyu

iotdb-core/datanode/src/main/java/org/apache/iotdb/db/pipe/sink/client/IoTDBDataNodeAsyncClientManager.java

@@ -22,6 +22,7 @@
 import org.apache.iotdb.common.rpc.thrift.TEndPoint;
 import org.apache.iotdb.commons.client.ClientPoolFactory;
 import org.apache.iotdb.commons.client.IClientManager;
+import org.apache.iotdb.commons.client.ThriftClient;
 import org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient;
 import org.apache.iotdb.commons.concurrent.IoTDBThreadPoolFactory;
 import org.apache.iotdb.commons.concurrent.ThreadName;
@@ -259,6 +260,7 @@ public void onComplete(final TPipeTransferResp response) {
 
           @Override
           public void onError(final Exception e) {
+            ThriftClient.resolveException(e, client);
             PipeLogger.log(
                 LOGGER::warn,
                 e,

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

@@ -19,6 +19,7 @@
 
 package org.apache.iotdb.db.pipe.sink.protocol.thrift.async.handler;
 
+import org.apache.iotdb.commons.client.ThriftClient;
 import org.apache.iotdb.commons.client.async.AsyncPipeDataTransferServiceClient;
 import org.apache.iotdb.db.pipe.sink.protocol.thrift.async.IoTDBDataRegionAsyncSink;
 import org.apache.iotdb.service.rpc.thrift.TPipeTransferReq;
@@ -61,6 +62,10 @@ public void onComplete(final TPipeTransferResp response) {
 
   @Override
   public void onError(final Exception exception) {
+    if (client != null) {
+      ThriftClient.resolveException(exception, client);
+    }
+
     if (connector.isClosed()) {
       clearEventsReferenceCount();
       connector.eliminateHandler(this, true);

iotdb-core/node-commons/src/main/java/org/apache/iotdb/commons/client/ThriftClient.java

@@ -29,6 +29,7 @@
 import java.lang.reflect.InvocationTargetException;
 import java.net.ConnectException;
 import java.net.SocketException;
+import java.nio.channels.ClosedChannelException;
 import java.util.Optional;
 
 /**
@@ -113,7 +114,8 @@ static boolean isConnectionBroken(Throwable cause) {
         || (cause instanceof IOException
             && (hasExpectedMessage(cause, "Connection reset by peer")
                 || hasExpectedMessage(cause, "Broken pipe")))
-        || (cause instanceof ConnectException && hasExpectedMessage(cause, "Connection refused"));
+        || (cause instanceof ConnectException && hasExpectedMessage(cause, "Connection refused")
+            || (cause instanceof ClosedChannelException));
   }
 
   static boolean hasExpectedMessage(Throwable cause, String expectedMessage) {