optimize: support fast fail when channel disconnect after message sent for RemotingClient

changes/en-us/2.x.md

@@ -18,6 +18,7 @@ Add changes here for all PR submitted to the 2.x branch.
 - [[#7075](https://github.com/apache/incubator-seata/pull/7075)] fast fail when channel is null
 - [[#7089](https://github.com/apache/incubator-seata/pull/7089)] support instance registration to the registry center
 - [[#7093](https://github.com/apache/incubator-seata/pull/7093)] add a test workflow for JDK 21
+- [[#7095](https://github.com/apache/incubator-seata/pull/7095)] support fast fail when channel disconnect after message sent for RemotingClient
 
 ### security:
 

changes/zh-cn/2.x.md

@@ -18,6 +18,7 @@
 - [[#7075](https://github.com/apache/incubator-seata/pull/7075)] 当channel为空时，快速失败，以便于减少不必要的等待
 - [[#7089](https://github.com/apache/incubator-seata/pull/7089)] 新增instance注册到注册中心的接口
 - [[#7093](https://github.com/apache/incubator-seata/pull/7093)] 增加jdk21的工作流测试
+- [[#7095](https://github.com/apache/incubator-seata/pull/7095)] 当消息发送后通道中断，快速失败，以便于减少不必要的等待
 
 ### security:
 

core/src/main/java/org/apache/seata/core/rpc/netty/AbstractNettyRemotingClient.java

@@ -20,6 +20,7 @@
 import java.net.InetSocketAddress;
 import java.util.List;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ExecutorService;
@@ -91,6 +92,9 @@ public abstract class AbstractNettyRemotingClient extends AbstractNettyRemoting
 
     protected final Map<Integer, Integer> childToParentMap = new ConcurrentHashMap<>();
 
+    // Channel -> Set<msgId>
+    protected final ConcurrentHashMap<Channel, Set<Integer>> channelToRequestIds = new ConcurrentHashMap<>();
+
     /**
      * When batch sending is enabled, the message will be stored to basketMap
      * Send via asynchronous thread {@link AbstractNettyRemotingClient.MergedSendRunnable}
@@ -181,6 +185,7 @@ public Object sendSyncRequest(Object msg) throws TimeoutException {
             }
         } else {
             Channel channel = clientChannelManager.acquireChannel(serverAddress);
+            channelToRequestIds.computeIfAbsent(channel, ch -> ConcurrentHashMap.newKeySet()).add(rpcMessage.getId());
             return super.sendSync(channel, rpcMessage, timeoutMillis);
         }
 
@@ -193,6 +198,7 @@ public Object sendSyncRequest(Channel channel, Object msg) throws TimeoutExcepti
             return null;
         }
         RpcMessage rpcMessage = buildRequestMessage(msg, ProtocolConstants.MSGTYPE_RESQUEST_SYNC);
+        channelToRequestIds.computeIfAbsent(channel, ch -> ConcurrentHashMap.newKeySet()).add(rpcMessage.getId());
         return super.sendSync(channel, rpcMessage, this.getRpcRequestTimeout());
     }
 
@@ -214,6 +220,7 @@ public void sendAsyncRequest(Channel channel, Object msg) {
                     childToParentMap.put(msgId, parentId);
                 }
             }
+            channelToRequestIds.computeIfAbsent(channel, ch -> ConcurrentHashMap.newKeySet()).add(rpcMessage.getId());
         }
         super.sendAsync(channel, rpcMessage);
     }
@@ -222,6 +229,7 @@ public void sendAsyncRequest(Channel channel, Object msg) {
     public void sendAsyncResponse(String serverAddress, RpcMessage rpcMessage, Object msg) {
         RpcMessage rpcMsg = buildResponseMessage(rpcMessage, msg, ProtocolConstants.MSGTYPE_RESPONSE);
         Channel channel = clientChannelManager.acquireChannel(serverAddress);
+        channelToRequestIds.computeIfAbsent(channel, ch -> ConcurrentHashMap.newKeySet()).add(rpcMessage.getId());
         super.sendAsync(channel, rpcMsg);
     }
 
@@ -423,7 +431,9 @@ class ClientHandler extends ChannelDuplexHandler {
         @Override
         public void channelRead(final ChannelHandlerContext ctx, Object msg) throws Exception {
             if (msg instanceof RpcMessage) {
-                processMessage(ctx, (RpcMessage)msg);
+                RpcMessage rpcMessage = (RpcMessage)msg;
+                processMessage(ctx, rpcMessage);
+                removeRequestIdFromChannel(ctx.channel(), rpcMessage.getId());
             } else {
                 LOGGER.error("rpcMessage type error");
             }
@@ -448,6 +458,7 @@ public void channelInactive(ChannelHandlerContext ctx) throws Exception {
                 LOGGER.info("channel inactive: {}", ctx.channel());
             }
             clientChannelManager.releaseChannel(ctx.channel(), NetUtil.toStringAddress(ctx.channel().remoteAddress()));
+            failFuturesForChannel(ctx.channel(), new FrameworkException(FrameworkErrorCode.ChannelIsNotWritable));
             super.channelInactive(ctx);
         }
 
@@ -489,6 +500,7 @@ public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws E
             if (LOGGER.isInfoEnabled()) {
                 LOGGER.info("remove exception rm channel:{}", ctx.channel());
             }
+            failFuturesForChannel(ctx.channel(), cause);
             super.exceptionCaught(ctx, cause);
         }
 
@@ -501,4 +513,59 @@ public void close(ChannelHandlerContext ctx, ChannelPromise future) throws Excep
         }
     }
 
+    /**
+     * Iterates over futures and marks all MessageFutures associated with the specified Channel as failed.
+     *
+     * @param channel The Channel that has been disconnected or encountered an exception.
+     * @param cause   The reason for the failure.
+     */
+    private void failFuturesForChannel(Channel channel, Throwable cause) {
+        Set<Integer> requestIds = channelToRequestIds.remove(channel);
+        if (requestIds != null) {
+            for (Integer requestId : requestIds) {
+                MessageFuture future = futures.remove(requestId);
+                if (future != null) {
+                    future.setResultMessage(cause);
+                }
+            }
+        }
+    }
+
+    /**
+     * Removes the association between the specified Channel and requestId from the channelToRequestIds mapping.
+     * If the Channel no longer has any associated requestId, the Channel will be removed from the mapping.
+     *
+     * @param channel   The Channel that has been disconnected or encountered an exception.
+     * @param requestId The requestId to be removed.
+     */
+    private void removeRequestIdFromChannel(Channel channel, Integer requestId) {
+        if (channel == null) {
+            if (requestId != null) {
+                LOGGER.warn("Attempted to remove requestId {} from a null channel.", requestId);
+            } else {
+                LOGGER.warn("Attempted to remove a null requestId from a null channel.");
+            }
+            return;
+        }
+
+        if (requestId == null) {
+            LOGGER.warn("Attempted to remove a null requestId from channel {}.", channel);
+            return;
+        }
+
+        channelToRequestIds.computeIfPresent(channel, (ch, requestIds) -> {
+            boolean removed = requestIds.remove(requestId);
+            if (removed) {
+                LOGGER.debug("Removed requestId {} from channel {}.", requestId, ch);
+            } else {
+                LOGGER.warn("Attempted to remove non-existing requestId {} from channel {}.", requestId, ch);
+            }
+
+            if (requestIds.isEmpty()) {
+                LOGGER.debug("No more requestIds associated with channel {}. Channel removed from mapping.", ch);
+                return null;
+            }
+            return requestIds;
+        });
+    }
 }