optimize: Add `ChannelEventListener` support to prevent memory leaks #7337

YongGoose · 2025-05-11T12:09:28Z

I have registered the PR changes.

Ⅰ. Describe what this PR did

Ⅱ. Does this pull request fix one issue?

Ⅲ. Why don't you add test cases (unit test/integration test)?

After the implementation is complete, it will be added.

graph TD;
    A[RemotingClient Interface] -->|implements| B[AbstractNettyRemotingClient];
    C[ChannelEventListener Interface] -->|implements| D[Custom Listener];
    B -->|registerChannelEventListener| C;
    B -->|setChannelHandlers| E[ChannelEventHandler];

    E -->|detects| F1[channelInactive];
    E -->|detects| F2[onChannelException];
    E -->|detects| F3[onChannelActive];
    E -->|detects| F4[onChannelIdle];

    F1 -->|invokes| G[cleanupResourcesForChannel];
    F2 -->|invokes| G;
    G -->|cleans up| H[mergeMsgMap];
    G -->|cleans up| I[futures];

    F1 -->|notifies| C;
    F2 -->|notifies| C;
    F3 -->|notifies| C;
    F4 -->|notifies| C;

codecov · 2025-05-11T13:49:10Z

Codecov Report

Attention: Patch coverage is 78.90625% with 27 lines in your changes missing coverage. Please review.

Project coverage is 54.88%. Comparing base (3b40167) to head (02e4c90).
Report is 1 commits behind head on 2.x.

Files with missing lines	Patch %	Lines
...ta/core/rpc/netty/AbstractNettyRemotingClient.java	86.48%	3 Missing and 7 partials ⚠️
...he/seata/core/rpc/netty/RmNettyRemotingClient.java	10.00%	9 Missing ⚠️
...che/seata/core/rpc/netty/ChannelEventListener.java	0.00%	4 Missing ⚠️
...ache/seata/core/rpc/netty/ChannelEventHandler.java	88.00%	0 Missing and 3 partials ⚠️
...he/seata/core/rpc/netty/TmNettyRemotingClient.java	90.00%	0 Missing and 1 partial ⚠️

Additional details and impacted files

@@             Coverage Diff              @@
##                2.x    #7337      +/-   ##
============================================
- Coverage     55.79%   54.88%   -0.91%     
+ Complexity     7476     7398      -78     
============================================
  Files          1178     1181       +3     
  Lines         41962    42089     +127     
  Branches       4923     4934      +11     
============================================
- Hits          23413    23101     -312     
- Misses        16310    16818     +508     
+ Partials       2239     2170      -69

Files with missing lines	Coverage Δ
.../apache/seata/core/rpc/netty/ChannelEventType.java	`100.00% <100.00%> (ø)`
...he/seata/core/rpc/netty/TmNettyRemotingClient.java	`73.21% <90.00%> (+0.66%)`	⬆️
...ache/seata/core/rpc/netty/ChannelEventHandler.java	`88.00% <88.00%> (ø)`
...che/seata/core/rpc/netty/ChannelEventListener.java	`0.00% <0.00%> (ø)`
...he/seata/core/rpc/netty/RmNettyRemotingClient.java	`48.96% <10.00%> (-3.63%)`	⬇️
...ta/core/rpc/netty/AbstractNettyRemotingClient.java	`37.11% <86.48%> (+6.37%)`	⬆️

... and 55 files with indirect coverage changes

🚀 New features to boost your workflow:

❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

funky-eyes

This design is to callback the ChannelEventListener registered by TM and RM, rather than adding a listener after channel.write(), right?

YongGoose · 2025-05-12T02:20:39Z

This design is to callback the ChannelEventListener registered by TM and RM, rather than adding a listener after channel.write(), right?

Yes, that's correct.

It is implemented in a way that calls back the ChannelEventListener registered in advance by TM and RM, so that user-defined actions can be performed when the channel event is detected.

funky-eyes · 2025-05-12T06:19:47Z

This design is to callback the ChannelEventListener registered by TM and RM, rather than adding a listener after channel.write(), right?

Yes, that's correct.

It is implemented in a way that calls back the ChannelEventListener registered in advance by TM and RM, so that user-defined actions can be performed when the channel event is detected.

I think this design is good, but since this functionality is not a user-facing API, it shouldn't be a feature-type PR. I believe it should be changed to an optimize type instead. Could you also fix the codecov/patch issue by adding some test cases to increase coverage? Of course, you could also improve unit test coverage in a subsequent PR.

YongGoose · 2025-05-12T06:41:43Z

I think this design is good, but since this functionality is not a user-facing API, it shouldn't be a feature-type PR. I believe it should be changed to an optimize type instead. Could you also fix the codecov/patch issue by adding some test cases to increase coverage? Of course, you could also improve unit test coverage in a subsequent PR.

Got it!
I'll update the PR type and add both unit and integration tests. 🚀

YongGoose · 2025-05-12T15:05:22Z

@funky-eyes

I believe the part I worked on this time is critical and could cause serious issues if a bug occurs. 🐞

That's why I plan to write more comprehensive unit and integration tests to cover a wide range of cases.
As a result, it might take some extra time.

I'll request a review once all the tests are complete!

funky-eyes · 2025-05-13T08:14:44Z

@funky-eyes

I believe the part I worked on this time is critical and could cause serious issues if a bug occurs. 🐞

That's why I plan to write more comprehensive unit and integration tests to cover a wide range of cases. As a result, it might take some extra time.

I'll request a review once all the tests are complete!

OK, I wish you success. If you have any questions, feel free to contact me.

YongGoose · 2025-05-15T14:25:27Z

core/src/test/java/org/apache/seata/core/rpc/netty/ChannelEventHandlerIntegrationTest.java

+    @Test
+    void testChannelInactiveByServer() throws Exception {
+        connectClient();
+
+        DefaultChannelGroup serverChannels = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);
+        serverChannels.addAll(collectServerChannels(workerGroup));
+        Channel serverSideClientChannel = serverChannels.stream()
+                .filter(ch -> ch.isActive() && ch.remoteAddress() != null)
+                .findFirst()
+                .orElseThrow(() -> new AssertionError("Failed to find client channel on server side"));
+
+        serverSideClientChannel.close().sync();
+        assertTrue(
+                channelInactiveLatch.await(TIMEOUT_SECONDS, TimeUnit.SECONDS),
+                "Channel inactive event was not detected on client side when server closed the connection");
+        verify(mockRemotingClient).onChannelInactive(any(Channel.class));
+    }


Please check whether this test verifies what we're trying to implement in this issue!

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

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

funky-eyes

LGTM

funky-eyes · 2025-05-20T07:37:37Z

I would like to know if a PR will be submitted soon to fix the issue mentioned in #5283?

YongGoose · 2025-05-20T07:47:03Z

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

+    protected void cleanupResourcesForChannel(Channel channel) {
+        if (channel == null) {
+            return;
+        }
+        ChannelException cause = new ChannelException(
+            String.format("Channel disconnected: %s", channel.remoteAddress()));
+
+        Set<Integer> messageIds = collectMessageIdsForChannel(channel.id());
+        cleanupFuturesForMessageIds(messageIds, cause);
+
+        LOGGER.info("Cleaned up {} pending requests for disconnected channel: {}",
+            messageIds.size(), channel.remoteAddress());
+    }
+
+    /**
+     * Collects message IDs associated with a specific channel.
+     * This is used during channel cleanup to identify pending requests.
+     *
+     * @param channelId the ID of the channel
+     * @return a set of message IDs associated with the channel
+     */
+    private Set<Integer> collectMessageIdsForChannel(ChannelId channelId) {
+        Set<Integer> messageIds = new HashSet<>();
+
+        String serverAddress = null;
+        for (Map.Entry<String, Channel> entry : clientChannelManager.getChannels().entrySet()) {
+            Channel channel = entry.getValue();
+            if (channelId.equals(channel.id())) {
+                serverAddress = entry.getKey();
+                break;
+            }
+        }


@funky-eyes

I believe this method addresses issue #5283.
What are your thoughts?

YongGoose · 2025-05-20T07:47:35Z

core/src/test/java/org/apache/seata/core/rpc/netty/ResourceCleanupTest.java

+    @Test
+    void testCleanupMessageFuturesOnChannelDisconnection() {
+        when(channel.id()).thenReturn(channelId);
+        when(channel.remoteAddress()).thenReturn(new InetSocketAddress("127.0.0.1", 8091));
+
+        MessageFuture messageFuture1 = new MessageFuture();
+        RpcMessage rpcMessage1 = createRpcMessage(1);
+        messageFuture1.setRequestMessage(rpcMessage1);
+        futures.put(1, messageFuture1);
+
+        MessageFuture messageFuture2 = new MessageFuture();
+        RpcMessage rpcMessage2 = createRpcMessage(2);
+        messageFuture2.setRequestMessage(rpcMessage2);
+        futures.put(2, messageFuture2);
+
+        int parentId = 100;
+        MergedWarpMessage mergeMessage = new MergedWarpMessage();
+        mergeMessage.msgIds = new ArrayList<>();
+        mergeMessage.msgIds.add(1);
+        mergeMessage.msgIds.add(2);
+
+        mergeMsgMap.put(parentId, mergeMessage);
+
+        String serverAddress = "127.0.0.1:8091";
+        channels.put(serverAddress, channel);
+
+        BlockingQueue<RpcMessage> basket = new LinkedBlockingQueue<>();
+        basket.add(rpcMessage1);
+        basket.add(rpcMessage2);
+        basketMap.put(serverAddress, basket);
+
+        client.cleanupResourcesForChannel(channel);
+
+        assertFalse(futures.containsKey(1), "Future ID 1 has not been removed");
+        assertFalse(futures.containsKey(2), "Future ID 2 has not been removed");
+
+        assertThrows(RuntimeException.class, () -> messageFuture1.get(0, java.util.concurrent.TimeUnit.MILLISECONDS));
+        assertThrows(RuntimeException.class, () -> messageFuture2.get(0, java.util.concurrent.TimeUnit.MILLISECONDS));
+    }


@funky-eyes

Ditto

YongGoose · 2025-05-20T07:48:12Z

I would like to know if a PR will be submitted soon to fix the issue mentioned in #5283?

Please take a look

optimize: Add ChannelEventListener support to prevent memory leaks #7337 (comment)
optimize: Add ChannelEventListener support to prevent memory leaks #7337 (comment)

funky-eyes · 2025-05-20T09:06:05Z

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

+     * @param cause the exception to set as the result for each future
+     */
+    private void cleanupFuturesForMessageIds(Set<Integer> messageIds, Exception cause) {
+        for (Integer messageId : messageIds) {


@Override public void sendAsyncRequest(Channel channel, Object msg) { if (channel == null) { LOGGER.warn("sendAsyncRequest nothing, caused by null channel."); throw new FrameworkException(new Throwable("throw"), "frameworkException", FrameworkErrorCode.ChannelIsNotWritable); } RpcMessage rpcMessage = buildRequestMessage(msg, msg instanceof HeartbeatMessage ? ProtocolConstants.MSGTYPE_HEARTBEAT_REQUEST : ProtocolConstants.MSGTYPE_RESQUEST_ONEWAY); Object body = rpcMessage.getBody(); if (body instanceof MergeMessage) { Integer parentId = rpcMessage.getId(); mergeMsgMap.put(parentId, (MergeMessage)rpcMessage.getBody()); if (body instanceof MergedWarpMessage) { for (Integer msgId : ((MergedWarpMessage)rpcMessage.getBody()).msgIds) { childToParentMap.put(msgId, parentId); } } } super.sendAsync(channel, rpcMessage); }

It seems that childToParentMap is not being cleared.

I'll create a PR that also clears the childToParentMap as soon as possible—hopefully by tomorrow.

…pache#7337)

YongGoose added 2 commits May 11, 2025 21:06

feat: Add channel event listener support to prevent memory leaks

06066e4

Add changes

ed2052b

YongGoose changed the title ~~feature: Add channel event listener support to prevent memory leaks~~ feature: Add ChannelEventListener support to prevent memory leaks May 11, 2025

YongGoose mentioned this pull request May 11, 2025

RemotingClient does not support setting a listener #7058

Closed

1 task

Add @sharable annotation

063c078

YongGoose force-pushed the feature/7058 branch from 82b7a7e to 063c078 Compare May 11, 2025 13:34

Merge branch '2.x' into feature/7058

6866c77

funky-eyes reviewed May 12, 2025

View reviewed changes

YongGoose added 2 commits May 12, 2025 23:48

Add ChannelEventListenerTest

81f14ba

Modify changes

0aac7c6

funky-eyes changed the title ~~feature: Add ChannelEventListener support to prevent memory leaks~~ optimize: Add ChannelEventListener support to prevent memory leaks May 13, 2025

YongGoose added 5 commits May 14, 2025 21:38

Add ResourceCleanupTest

c0a1516

Add License header

be00c16

Merge branch '2.x' into feature/7058

b05a1f2

Add ChannelEventHandlerIntegrationTest

597b93b

Modify listener action

1c4d945

YongGoose requested a review from funky-eyes May 15, 2025 14:15

YongGoose commented May 15, 2025

View reviewed changes

funky-eyes reviewed May 18, 2025

View reviewed changes

core/src/main/java/org/apache/seata/core/rpc/netty/RmNettyRemotingClient.java Outdated Show resolved Hide resolved

core/src/main/java/org/apache/seata/core/rpc/netty/TmNettyRemotingClient.java Outdated Show resolved Hide resolved

Apply comment

02e4c90

funky-eyes added this to the 2.5.0 milestone May 19, 2025

funky-eyes added module/core core module type: optimize labels May 19, 2025

funky-eyes approved these changes May 19, 2025

View reviewed changes

YongGoose requested a review from slievrly May 19, 2025 05:19

funky-eyes merged commit c5efbd9 into apache:2.x May 19, 2025
9 checks passed

YongGoose commented May 20, 2025

View reviewed changes

funky-eyes reviewed May 20, 2025

View reviewed changes

YongGoose mentioned this pull request May 20, 2025

ClientOnResponseProcessor#mergeMsgMap&futures is a risk of memory leaks #5283

Closed

1 task

jsbxyyx pushed a commit to jsbxyyx/incubator-seata that referenced this pull request Jun 11, 2025

optimize: Add ChannelEventListener support to prevent memory leaks (a…

67c53fe

…pache#7337)

slievrly pushed a commit to slievrly/fescar that referenced this pull request Oct 21, 2025

optimize: Add ChannelEventListener support to prevent memory leaks (a…

ac7cd26

…pache#7337)

YvCeung pushed a commit to YvCeung/incubator-seata that referenced this pull request Dec 25, 2025

optimize: Add ChannelEventListener support to prevent memory leaks (a…

4ba44b7

…pache#7337)

YongGoose mentioned this pull request Dec 28, 2025

📢오픈소스 기여모임 - 운영진 소개 opensource-contributors-group/opensource-contributors#2

Open

optimize: Add ChannelEventListener support to prevent memory leaks #7337

optimize: Add ChannelEventListener support to prevent memory leaks #7337

Uh oh!

Conversation

YongGoose commented May 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Ⅰ. Describe what this PR did

Ⅱ. Does this pull request fix one issue?

Ⅲ. Why don't you add test cases (unit test/integration test)?

Uh oh!

codecov bot commented May 11, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Codecov Report

Uh oh!

funky-eyes left a comment

Choose a reason for hiding this comment

Uh oh!

YongGoose commented May 12, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

funky-eyes commented May 12, 2025

Uh oh!

YongGoose commented May 12, 2025

Uh oh!

YongGoose commented May 12, 2025

Uh oh!

funky-eyes commented May 13, 2025

Uh oh!

YongGoose May 15, 2025

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

funky-eyes left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

funky-eyes commented May 20, 2025

Uh oh!

YongGoose May 20, 2025

Choose a reason for hiding this comment

Uh oh!

YongGoose May 20, 2025

Choose a reason for hiding this comment

Uh oh!

YongGoose commented May 20, 2025

Uh oh!

funky-eyes May 20, 2025

Choose a reason for hiding this comment

Uh oh!

YongGoose May 20, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants

optimize: Add `ChannelEventListener` support to prevent memory leaks #7337

optimize: Add `ChannelEventListener` support to prevent memory leaks #7337

YongGoose commented May 11, 2025 •

edited

Loading

codecov bot commented May 11, 2025 •

edited

Loading

YongGoose commented May 12, 2025 •

edited

Loading