You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@seata.apache.org by wa...@apache.org on 2026/10/11 09:39:58 UTC
(incubator-seata) branch 2.x updated: bugfix: fix TC memory leak from retained client channels
This is an automated email from the ASF dual-hosted git repository.
WangzJi pushed a commit to branch 2.x
in repository https://gitbox.apache.org/repos/asf/incubator-seata.git
The following commit(s) were added to refs/heads/2.x by this push:
new 8a31b1f41d bugfix: fix TC memory leak from retained client channels
8a31b1f41d is described below
commit 8a31b1f41da325f5b498d3fa2853221620388d35
Author: alex-ashuo <sa...@hotmail.com>
AuthorDate: Sun Oct 11 17:39:53 2026 +0800
bugfix: fix TC memory leak from retained client channels
---
changes/en-us/2.x.md | 3 +
changes/zh-cn/2.x.md | 3 +
.../rpc/netty/AbstractNettyRemotingServer.java | 40 ++-
.../seata/core/rpc/netty/ChannelManager.java | 24 ++
.../core/rpc/processor/server/RegRmProcessor.java | 10 +-
.../processor/server/ServerOnRequestProcessor.java | 9 +-
.../core/rpc/netty/ChannelManagerOOMTest.java | 341 +++++++++++++++++++++
.../server/ServerOnRequestProcessorTest.java | 31 ++
8 files changed, 455 insertions(+), 6 deletions(-)
diff --git a/changes/en-us/2.x.md b/changes/en-us/2.x.md
index 2dc7b9a38b..23668b3d5e 100644
--- a/changes/en-us/2.x.md
+++ b/changes/en-us/2.x.md
@@ -45,6 +45,7 @@ Add changes here for all PR submitted to the 2.x branch.
- [[#8182](https://github.com/apache/incubator-seata/issues/8182)] fix registry preferred networks and ignored interfaces from Spring Boot application.yml
- [[#8201](https://github.com/apache/incubator-seata/issues/8197)] fix residual file-mode global locks caused by an async commit cleanup race
- [[#8276](https://github.com/apache/incubator-seata/pull/8276)] fix Redis global session query returning a non-matching XID row when a status filter is empty
+- [[#8224](https://github.com/apache/incubator-seata/pull/8224)] remove closed client channels from the TC identified channels after a grace period to fix TC FullGC and OOM
@@ -108,6 +109,8 @@ Thanks to these contributors for their code commits. Please report an unintended
- [pyshiweijia](https://github.com/pyshiweijia)
- [biedongbin](https://github.com/biedongbin)
- [Zhengcy05](https://github.com/Zhengcy05)
+- [saul-c](https://github.com/saul-c)
+- [YongGoose](https://github.com/YongGoose)
diff --git a/changes/zh-cn/2.x.md b/changes/zh-cn/2.x.md
index ccd7dc4aaa..b2473c2061 100644
--- a/changes/zh-cn/2.x.md
+++ b/changes/zh-cn/2.x.md
@@ -45,6 +45,7 @@
- [[#8182](https://github.com/apache/incubator-seata/issues/8182)] 修复 Spring Boot application.yml 中 registry preferred networks 和 ignored interfaces 配置不生效的问题
- [[#8201](https://github.com/apache/incubator-seata/issues/8197)] 修复异步提交清理竞态导致 file 模式全局锁残留的问题
- [[#8276](https://github.com/apache/incubator-seata/pull/8276)] 修复 Redis 全局会话按 XID 加状态查询时，状态不匹配仍返回该会话的问题
+- [[#8224](https://github.com/apache/incubator-seata/pull/8224)] 客户端连接关闭后延迟清理 TC 中的已识别通道，修复其持续堆积导致的 FullGC 与 OOM 问题
### optimize:
@@ -108,6 +109,8 @@
- [pyshiweijia](https://github.com/pyshiweijia)
- [biedongbin](https://github.com/biedongbin)
- [Zhengcy05](https://github.com/Zhengcy05)
+- [saul-c](https://github.com/saul-c)
+- [YongGoose](https://github.com/YongGoose)
diff --git a/core/src/main/java/org/apache/seata/core/rpc/netty/AbstractNettyRemotingServer.java b/core/src/main/java/org/apache/seata/core/rpc/netty/AbstractNettyRemotingServer.java
index 1bcad8c6f7..8cdefebf37 100644
--- a/core/src/main/java/org/apache/seata/core/rpc/netty/AbstractNettyRemotingServer.java
+++ b/core/src/main/java/org/apache/seata/core/rpc/netty/AbstractNettyRemotingServer.java
@@ -24,6 +24,7 @@ import io.netty.channel.ChannelPromise;
import io.netty.handler.codec.DecoderException;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
+import org.apache.seata.common.DefaultValues;
import org.apache.seata.common.util.NetUtil;
import org.apache.seata.common.util.StringUtils;
import org.apache.seata.core.protocol.AbstractMessage;
@@ -41,7 +42,9 @@ import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
/**
@@ -53,6 +56,13 @@ public abstract class AbstractNettyRemotingServer extends AbstractNettyRemoting
private static final Logger LOGGER = LoggerFactory.getLogger(AbstractNettyRemotingServer.class);
+ /**
+ * How long a closed channel stays identified, so requests and responses still in flight for it keep working;
+ * a TM response, for one, can fall back to another channel of the same client. Matches the default TM request
+ * timeout, after which the client no longer waits.
+ */
+ private static final long CLOSED_CHANNEL_RETENTION_MILLS = DefaultValues.DEFAULT_RPC_TM_REQUEST_TIMEOUT;
+
private final NettyServerBootstrap serverBootstrap;
@Override
@@ -136,6 +146,26 @@ public abstract class AbstractNettyRemotingServer extends AbstractNettyRemoting
super.destroy();
}
+ /**
+ * Remove a closed channel from the identified channels once the retention time has passed.
+ * Until then, requests already received from it are still handled and a TM response can fall back to another
+ * channel of the same client.
+ *
+ * @param channel the closed channel
+ * @param rpcContext the context the channel was bound to
+ */
+ protected void removeIdentifiedChannelLater(Channel channel, RpcContext rpcContext) {
+ try {
+ timerExecutor.schedule(
+ () -> ChannelManager.removeIdentifiedChannel(channel, rpcContext),
+ CLOSED_CHANNEL_RETENTION_MILLS,
+ TimeUnit.MILLISECONDS);
+ } catch (RejectedExecutionException e) {
+ // the server is shutting down, nothing routes to this channel anymore
+ ChannelManager.removeIdentifiedChannel(channel, rpcContext);
+ }
+ }
+
/**
* Debug log.
*
@@ -203,7 +233,14 @@ public abstract class AbstractNettyRemotingServer extends AbstractNettyRemoting
if (messageExecutor.isShutdown()) {
return;
}
- handleDisconnect(ctx);
+ RpcContext rpcContext = ChannelManager.getContextFromIdentified(ctx.channel());
+ try {
+ handleDisconnect(ctx);
+ } finally {
+ if (rpcContext != null) {
+ removeIdentifiedChannelLater(ctx.channel(), rpcContext);
+ }
+ }
super.channelInactive(ctx);
}
@@ -288,6 +325,7 @@ public abstract class AbstractNettyRemotingServer extends AbstractNettyRemoting
// If the client is not version 2.3.0 or higher, splitting MergedWarpMessage will result in the client’s
// mergeMsgMap not being cleared
if (body instanceof MergedWarpMessage
+ && rpcContext != null
&& (StringUtils.isNotBlank(rpcContext.getVersion())
&& Version.isAboveOrEqualVersion230(rpcContext.getVersion()))) {
MergedWarpMessage mergedWarpMessage = (MergedWarpMessage) body;
diff --git a/core/src/main/java/org/apache/seata/core/rpc/netty/ChannelManager.java b/core/src/main/java/org/apache/seata/core/rpc/netty/ChannelManager.java
index 4fc6d04249..b66025ca93 100644
--- a/core/src/main/java/org/apache/seata/core/rpc/netty/ChannelManager.java
+++ b/core/src/main/java/org/apache/seata/core/rpc/netty/ChannelManager.java
@@ -290,6 +290,30 @@ public class ChannelManager {
}
}
+ /**
+ * Remove a closed channel from the identified channels, only if it is still bound to the given context.
+ *
+ * @param channel the closed channel
+ * @param rpcContext the context the channel was bound to when it was closed
+ */
+ static void removeIdentifiedChannel(Channel channel, RpcContext rpcContext) {
+ IDENTIFIED_CHANNELS.remove(channel, rpcContext);
+ }
+
+ /**
+ * Release a closed channel and remove it from the identified channels at once. Used when the channel was
+ * registered after it had closed, so the disconnect handling found nothing to clean up.
+ *
+ * @param channel the closed channel
+ */
+ public static void removeClosedChannel(Channel channel) {
+ RpcContext rpcContext = getContextFromIdentified(channel);
+ if (rpcContext != null) {
+ rpcContext.release();
+ removeIdentifiedChannel(channel, rpcContext);
+ }
+ }
+
/**
* Gets get same income client channel.
*
diff --git a/core/src/main/java/org/apache/seata/core/rpc/processor/server/RegRmProcessor.java b/core/src/main/java/org/apache/seata/core/rpc/processor/server/RegRmProcessor.java
index 88578f446a..e43b164cac 100644
--- a/core/src/main/java/org/apache/seata/core/rpc/processor/server/RegRmProcessor.java
+++ b/core/src/main/java/org/apache/seata/core/rpc/processor/server/RegRmProcessor.java
@@ -93,7 +93,15 @@ public class RegRmProcessor implements RemotingProcessor {
if (StringUtils.isNotEmpty(errorInfo)) {
response.setMsg(errorInfo);
}
- remotingServer.sendAsyncResponse(rpcMessage, ctx.channel(), response);
+ try {
+ remotingServer.sendAsyncResponse(rpcMessage, ctx.channel(), response);
+ } finally {
+ // the channel may have closed while this request waited in the executor, and then the disconnect
+ // handling had nothing to clean up yet
+ if (!ctx.channel().isActive()) {
+ ChannelManager.removeClosedChannel(ctx.channel());
+ }
+ }
if (isSuccess && LOGGER.isInfoEnabled()) {
LOGGER.info(
"RM register success,message:{},channel:{},client version:{}",
diff --git a/core/src/main/java/org/apache/seata/core/rpc/processor/server/ServerOnRequestProcessor.java b/core/src/main/java/org/apache/seata/core/rpc/processor/server/ServerOnRequestProcessor.java
index 367120ec1b..1696d665d9 100644
--- a/core/src/main/java/org/apache/seata/core/rpc/processor/server/ServerOnRequestProcessor.java
+++ b/core/src/main/java/org/apache/seata/core/rpc/processor/server/ServerOnRequestProcessor.java
@@ -120,8 +120,10 @@ public class ServerOnRequestProcessor implements RemotingProcessor, Disposable {
@Override
public void process(ChannelHandlerContext ctx, RpcMessage rpcMessage) throws Exception {
- if (ChannelManager.isRegistered(ctx.channel())) {
- onRequestMessage(ctx, rpcMessage);
+ // look the context up only once, the IO thread may release the channel at any time
+ RpcContext rpcContext = ChannelManager.getContextFromIdentified(ctx.channel());
+ if (rpcContext != null) {
+ onRequestMessage(ctx, rpcMessage, rpcContext);
} else {
try {
if (LOGGER.isInfoEnabled()) {
@@ -146,9 +148,8 @@ public class ServerOnRequestProcessor implements RemotingProcessor, Disposable {
}
}
- private void onRequestMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage) {
+ private void onRequestMessage(ChannelHandlerContext ctx, RpcMessage rpcMessage, RpcContext rpcContext) {
Object message = rpcMessage.getBody();
- RpcContext rpcContext = ChannelManager.getContextFromIdentified(ctx.channel());
if (!(message instanceof AbstractMessage)) {
LOGGER.error("unrecognized message:{}", message);
return;
diff --git a/core/src/test/java/org/apache/seata/core/rpc/netty/ChannelManagerOOMTest.java b/core/src/test/java/org/apache/seata/core/rpc/netty/ChannelManagerOOMTest.java
new file mode 100644
index 0000000000..ce140cd7a3
--- /dev/null
+++ b/core/src/test/java/org/apache/seata/core/rpc/netty/ChannelManagerOOMTest.java
@@ -0,0 +1,341 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.seata.core.rpc.netty;
+
+import io.netty.channel.Channel;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelPromise;
+import io.netty.handler.codec.DecoderException;
+import org.apache.seata.common.thread.NamedThreadFactory;
+import org.apache.seata.core.protocol.AbstractMessage;
+import org.apache.seata.core.protocol.MergedWarpMessage;
+import org.apache.seata.core.protocol.MessageType;
+import org.apache.seata.core.protocol.RegisterRMRequest;
+import org.apache.seata.core.protocol.RegisterTMRequest;
+import org.apache.seata.core.protocol.RpcMessage;
+import org.apache.seata.core.protocol.transaction.BranchRegisterRequest;
+import org.apache.seata.core.protocol.transaction.GlobalBeginRequest;
+import org.apache.seata.core.protocol.transaction.GlobalBeginResponse;
+import org.apache.seata.core.protocol.transaction.GlobalCommitResponse;
+import org.apache.seata.core.rpc.RemotingServer;
+import org.apache.seata.core.rpc.RpcContext;
+import org.apache.seata.core.rpc.TransactionMessageHandler;
+import org.apache.seata.core.rpc.processor.RemotingProcessor;
+import org.apache.seata.core.rpc.processor.server.RegRmProcessor;
+import org.apache.seata.core.rpc.processor.server.ServerOnRequestProcessor;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.net.InetSocketAddress;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.ArgumentMatchers.notNull;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+/**
+ * Closed client channels must leave the identified channels of {@link ChannelManager}, without breaking the
+ * requests and responses that are still in flight for them.
+ */
+class ChannelManagerOOMTest {
+
+ private static final String APP_ID = "business-app";
+ private static final String TX_GROUP = "default_tx_group";
+ private static final String RESOURCE_ID = "jdbc:mysql://127.0.0.1:3306/seata";
+
+ private ThreadPoolExecutor messageExecutor;
+ private TestServer server;
+ private AbstractNettyRemotingServer.ServerHandler handler;
+
+ @BeforeEach
+ void setUp() throws Exception {
+ clearChannelMaps();
+ messageExecutor = new ThreadPoolExecutor(
+ 1, 1, 60, TimeUnit.SECONDS, new LinkedBlockingQueue<>(), new NamedThreadFactory("oom-test", 1));
+ server = new TestServer(messageExecutor);
+ handler = server.new ServerHandler();
+ }
+
+ @AfterEach
+ void tearDown() throws Exception {
+ try {
+ server.destroy();
+ } catch (Exception e) {
+ // Ignore
+ }
+ messageExecutor.shutdown();
+ clearChannelMaps();
+ }
+
+ @Test
+ void repeatedReconnectsDoNotAccumulateClosedChannels() throws Exception {
+ for (int i = 0; i < 200; i++) {
+ Channel tmChannel = channel(20000 + i, false);
+ ChannelManager.registerTMChannel(new RegisterTMRequest(APP_ID, TX_GROUP), tmChannel);
+ handler.channelInactive(ctx(tmChannel));
+
+ Channel rmChannel = channel(30000 + i, false);
+ ChannelManager.registerRMChannel(rmRequest(), rmChannel);
+ handler.channelInactive(ctx(rmChannel));
+ }
+
+ server.elapseRetention();
+
+ assertTrue(channelMap("IDENTIFIED_CHANNELS").isEmpty());
+ assertEquals(0, contextCount(channelMap("TM_CHANNELS")));
+ assertEquals(0, contextCount(channelMap("RM_CHANNELS")));
+ }
+
+ @Test
+ void rmRegistrationHandledAfterCloseDoesNotLeak() throws Exception {
+ Channel channel = channel(40010, false);
+ ChannelHandlerContext ctx = ctx(channel);
+
+ // the channel closes while its RegisterRM request still waits in the executor
+ handler.channelInactive(ctx);
+ new RegRmProcessor(mock(RemotingServer.class)).process(ctx, registerRequest(rmRequest()));
+
+ assertFalse(ChannelManager.isRegistered(channel));
+ assertEquals(0, contextCount(channelMap("RM_CHANNELS")));
+ }
+
+ @Test
+ void rmRegistrationDuringRetentionDoesNotRestoreRouting() throws Exception {
+ Channel channel = channel(40011, false);
+ ChannelManager.registerRMChannel(rmRequest(), channel);
+ ChannelHandlerContext ctx = ctx(channel);
+ RegisterRMRequest anotherResource = new RegisterRMRequest(APP_ID, TX_GROUP);
+ anotherResource.setResourceIds("jdbc:mysql://127.0.0.1:3306/another");
+
+ handler.channelInactive(ctx);
+ new RegRmProcessor(mock(RemotingServer.class)).process(ctx, registerRequest(anotherResource));
+
+ assertFalse(ChannelManager.isRegistered(channel));
+ assertEquals(0, contextCount(channelMap("RM_CHANNELS")));
+ }
+
+ @Test
+ void closedChannelStaysIdentifiedUntilRetentionElapses() throws Exception {
+ Channel channel = channel(40001, false);
+ ChannelManager.registerTMChannel(new RegisterTMRequest(APP_ID, TX_GROUP), channel);
+
+ handler.channelInactive(ctx(channel));
+ assertTrue(ChannelManager.isRegistered(channel));
+
+ server.elapseRetention();
+ assertFalse(ChannelManager.isRegistered(channel));
+ }
+
+ @Test
+ void responseForClosedTmChannelFallsBackToAnotherChannelOfTheSameClient() throws Exception {
+ Channel live = channel(40002, true);
+ when(live.isWritable()).thenReturn(true);
+ when(live.writeAndFlush(any())).thenReturn(mock(ChannelPromise.class));
+ Channel closed = channel(40003, false);
+ ChannelManager.registerTMChannel(new RegisterTMRequest(APP_ID, TX_GROUP), live);
+ ChannelManager.registerTMChannel(new RegisterTMRequest(APP_ID, TX_GROUP), closed);
+
+ handler.channelInactive(ctx(closed));
+ server.sendAsyncResponse(new RpcMessage(), closed, new GlobalCommitResponse());
+
+ verify(live).writeAndFlush(any(RpcMessage.class));
+ verify(closed, never()).writeAndFlush(any());
+ }
+
+ @Test
+ void requestQueuedBeforeCloseIsStillHandled() throws Exception {
+ Channel channel = channel(40004, false);
+ ChannelManager.registerTMChannel(new RegisterTMRequest(APP_ID, TX_GROUP), channel);
+ ChannelHandlerContext ctx = ctx(channel);
+ TransactionMessageHandler transactionMessageHandler = mock(TransactionMessageHandler.class);
+ when(transactionMessageHandler.onRequest(any(), any())).thenReturn(new GlobalBeginResponse());
+ ServerOnRequestProcessor processor =
+ new ServerOnRequestProcessor(mock(RemotingServer.class), transactionMessageHandler);
+ RpcMessage request = new RpcMessage();
+ request.setBody(new GlobalBeginRequest());
+
+ handler.channelInactive(ctx);
+ processor.process(ctx, request);
+
+ verify(transactionMessageHandler).onRequest(any(GlobalBeginRequest.class), notNull());
+ verify(ctx, never()).close();
+ }
+
+ @Test
+ void exceptionOnOpenChannelKeepsItServable() throws Exception {
+ Channel channel = channel(40005, true);
+ ChannelManager.registerRMChannel(rmRequest(), channel);
+ ChannelHandlerContext ctx = ctx(channel);
+ RemotingProcessor processor = mock(RemotingProcessor.class);
+ server.registerProcessor(MessageType.TYPE_BRANCH_REGISTER, processor, null);
+
+ handler.exceptionCaught(ctx, new DecoderException("broken frame"));
+ assertTrue(ChannelManager.isRegistered(channel));
+
+ server.processMessage(ctx, mergedRequest(new BranchRegisterRequest(), 1));
+ verify(processor).process(eq(ctx), any(RpcMessage.class));
+ }
+
+ @Test
+ void mergedRequestFromUnidentifiedChannelIsDispatchedWhole() throws Exception {
+ ChannelHandlerContext ctx = ctx(channel(40006, true));
+ RemotingProcessor processor = mock(RemotingProcessor.class);
+ server.registerProcessor(MessageType.TYPE_SEATA_MERGE, processor, null);
+ RpcMessage merged = mergedRequest(new BranchRegisterRequest(), 1);
+
+ server.processMessage(ctx, merged);
+
+ verify(processor).process(ctx, merged);
+ }
+
+ @Test
+ void removalKeepsTheContextTheChannelIsBoundTo() throws Exception {
+ Channel channel = channel(40007, true);
+ ChannelManager.registerTMChannel(new RegisterTMRequest(APP_ID, TX_GROUP), channel);
+
+ ChannelManager.removeIdentifiedChannel(channel, new RpcContext());
+
+ assertTrue(ChannelManager.isRegistered(channel));
+ }
+
+ @Test
+ void removalIsScheduledOnTheTimer() throws Exception {
+ Channel channel = channel(40008, false);
+ ChannelManager.registerTMChannel(new RegisterTMRequest(APP_ID, TX_GROUP), channel);
+ server.captureRemovals = false;
+
+ handler.channelInactive(ctx(channel));
+
+ assertTrue(ChannelManager.isRegistered(channel));
+ ScheduledThreadPoolExecutor timer = (ScheduledThreadPoolExecutor) server.timerExecutor;
+ assertEquals(1, timer.getQueue().size());
+ assertTrue(((ScheduledFuture<?>) timer.getQueue().peek()).getDelay(TimeUnit.MILLISECONDS) > 0);
+ }
+
+ @Test
+ void removalRunsAtOnceWhenTheServerIsShuttingDown() throws Exception {
+ Channel channel = channel(40009, false);
+ ChannelManager.registerTMChannel(new RegisterTMRequest(APP_ID, TX_GROUP), channel);
+ server.captureRemovals = false;
+ server.timerExecutor.shutdown();
+
+ server.removeIdentifiedChannelLater(channel, ChannelManager.getContextFromIdentified(channel));
+
+ assertFalse(ChannelManager.isRegistered(channel));
+ }
+
+ private static RegisterRMRequest rmRequest() {
+ RegisterRMRequest request = new RegisterRMRequest(APP_ID, TX_GROUP);
+ request.setResourceIds(RESOURCE_ID);
+ return request;
+ }
+
+ private static RpcMessage registerRequest(RegisterRMRequest request) {
+ RpcMessage rpcMessage = new RpcMessage();
+ rpcMessage.setBody(request);
+ return rpcMessage;
+ }
+
+ private static RpcMessage mergedRequest(AbstractMessage message, int msgId) {
+ MergedWarpMessage mergedWarpMessage = new MergedWarpMessage();
+ mergedWarpMessage.msgs.add(message);
+ mergedWarpMessage.msgIds.add(msgId);
+ RpcMessage rpcMessage = new RpcMessage();
+ rpcMessage.setBody(mergedWarpMessage);
+ return rpcMessage;
+ }
+
+ private static Channel channel(int remotePort, boolean active) {
+ Channel channel = mock(Channel.class);
+ when(channel.remoteAddress()).thenReturn(new InetSocketAddress("127.0.0.1", remotePort));
+ when(channel.isActive()).thenReturn(active);
+ return channel;
+ }
+
+ private static ChannelHandlerContext ctx(Channel channel) {
+ ChannelHandlerContext ctx = mock(ChannelHandlerContext.class);
+ when(ctx.channel()).thenReturn(channel);
+ return ctx;
+ }
+
+ private static Map<?, ?> channelMap(String fieldName) throws Exception {
+ Field field = ChannelManager.class.getDeclaredField(fieldName);
+ field.setAccessible(true);
+ return (Map<?, ?>) field.get(null);
+ }
+
+ private static int contextCount(Map<?, ?> map) {
+ int count = 0;
+ for (Object value : map.values()) {
+ count += value instanceof Map ? contextCount((Map<?, ?>) value) : 1;
+ }
+ return count;
+ }
+
+ private static void clearChannelMaps() throws Exception {
+ channelMap("IDENTIFIED_CHANNELS").clear();
+ channelMap("TM_CHANNELS").clear();
+ channelMap("RM_CHANNELS").clear();
+ }
+
+ /**
+ * Holds the delayed removals back until the test lets the retention time elapse.
+ */
+ static class TestServer extends AbstractNettyRemotingServer {
+
+ private final List<Runnable> pendingRemovals = new ArrayList<>();
+
+ private boolean captureRemovals = true;
+
+ TestServer(ThreadPoolExecutor messageExecutor) {
+ super(messageExecutor, new NettyServerConfig());
+ }
+
+ @Override
+ public void destroyChannel(String serverAddress, Channel channel) {}
+
+ @Override
+ protected void removeIdentifiedChannelLater(Channel channel, RpcContext rpcContext) {
+ if (captureRemovals) {
+ pendingRemovals.add(() -> ChannelManager.removeIdentifiedChannel(channel, rpcContext));
+ } else {
+ super.removeIdentifiedChannelLater(channel, rpcContext);
+ }
+ }
+
+ void elapseRetention() {
+ pendingRemovals.forEach(Runnable::run);
+ pendingRemovals.clear();
+ }
+ }
+}
diff --git a/core/src/test/java/org/apache/seata/core/rpc/processor/server/ServerOnRequestProcessorTest.java b/core/src/test/java/org/apache/seata/core/rpc/processor/server/ServerOnRequestProcessorTest.java
index 6c978b3f2a..c3b554f30c 100644
--- a/core/src/test/java/org/apache/seata/core/rpc/processor/server/ServerOnRequestProcessorTest.java
+++ b/core/src/test/java/org/apache/seata/core/rpc/processor/server/ServerOnRequestProcessorTest.java
@@ -207,6 +207,37 @@ public class ServerOnRequestProcessorTest {
verify(unregisteredCtx).close();
}
+ @Test
+ public void testProcessKeepsContextWhenChannelIsReleasedConcurrently() throws Exception {
+ GlobalBeginRequest request = new GlobalBeginRequest();
+ when(transactionMessageHandler.onRequest(any(GlobalBeginRequest.class), any(RpcContext.class)))
+ .thenReturn(new GlobalBeginResponse());
+ RpcMessage rpcMessage = new RpcMessage() {
+ @Override
+ public Object getBody() {
+ // the IO thread drops the channel while the request is being handled
+ removeIdentifiedChannel(channel);
+ return super.getBody();
+ }
+ };
+ rpcMessage.setId(1);
+ rpcMessage.setBody(request);
+
+ processor.process(ctx, rpcMessage);
+
+ verify(transactionMessageHandler).onRequest(eq(request), eq(rpcContext));
+ }
+
+ private static void removeIdentifiedChannel(Channel channel) {
+ try {
+ Field field = ChannelManager.class.getDeclaredField("IDENTIFIED_CHANNELS");
+ field.setAccessible(true);
+ ((Map<?, ?>) field.get(null)).remove(channel);
+ } catch (ReflectiveOperationException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+
@Test
public void testProcessNonAbstractMessage() throws Exception {
RpcMessage rpcMessage = new RpcMessage();
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@seata.apache.org
For additional commands, e-mail: notifications-help@seata.apache.org