From ef3d889a58ae6acc21502139b68b9848e87106c1 Mon Sep 17 00:00:00 2001 From: qianye Date: Mon, 14 Sep 2026 15:37:55 +0800 Subject: [PATCH] [ISSUE #11161] Fix closeChannel table eviction and ChannelWrapper.close lock ordering Bug 1: closeChannel(addr, channel) took the "created again, nothing to do" branch on the normal close path because the condition introduced by #8366 (prevCW.isWrapperOf(channel)) was the un-negated form of the original prevCW.getChannel() != channel. Restore the negation so the table entry is evicted when the stored wrapper wraps the channel being closed, and kept only when the wrapper has already been recreated for a different channel. Bug 2: ChannelWrapper.close() held the wrapper write lock while calling closeChannel(...), which acquires lockChannelTables, inverting the lockChannelTables -> wrapper lock order used by createChannelAsync and closeChannel (via tryClose). Snapshot both channel futures under the read lock, release it, then close outside any wrapper lock. Both channelFuture and channelToClose are still closed to avoid leaking the channel left over from a reconnect. Add NettyRemotingClientCloseChannelTest covering both closeChannel branches. --- .../remoting/netty/NettyRemotingClient.java | 27 ++++-- .../NettyRemotingClientCloseChannelTest.java | 82 +++++++++++++++++++ 2 files changed, 100 insertions(+), 9 deletions(-) create mode 100644 remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingClientCloseChannelTest.java diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java index 627d6255f51..4ba18b53d9f 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingClient.java @@ -436,7 +436,7 @@ public void closeChannel(final String addr, final Channel channel) { if (null == prevCW) { LOGGER.info("closeChannel: the channel[addr={}, id={}] has been removed from the channel table before", addrRemote, channel.id()); removeItemFromTable = false; - } else if (prevCW.isWrapperOf(channel)) { + } else if (!prevCW.isWrapperOf(channel)) { LOGGER.info("closeChannel: the channel[addr={}, id={}] has been closed before, and has been created again, nothing to do.", addrRemote, channel.id()); removeItemFromTable = false; @@ -1087,16 +1087,25 @@ public boolean tryClose(Channel channel) { } public void close() { + // Snapshot the channels under the read lock, then close them without holding + // any wrapper lock. Holding the wrapper lock across closeChannel would take + // lockChannelTables while inside the wrapper lock, which inverts the + // lockChannelTables -> wrapper lock order used by createChannelAsync and + // closeChannel (via tryClose). + ChannelFuture current; + ChannelFuture toClose; + lock.readLock().lock(); try { - lock.writeLock().lock(); - if (channelFuture != null) { - closeChannel(channelFuture.channel()); - } - if (channelToClose != null) { - closeChannel(channelToClose.channel()); - } + current = this.channelFuture; + toClose = this.channelToClose; } finally { - lock.writeLock().unlock(); + lock.readLock().unlock(); + } + if (current != null) { + closeChannel(channelAddress, current.channel()); + } + if (toClose != null) { + closeChannel(channelAddress, toClose.channel()); } } } diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingClientCloseChannelTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingClientCloseChannelTest.java new file mode 100644 index 00000000000..dd51fde0791 --- /dev/null +++ b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingClientCloseChannelTest.java @@ -0,0 +1,82 @@ +/* + * 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.rocketmq.remoting.netty; + +import io.netty.channel.ChannelFuture; +import io.netty.channel.DefaultChannelPromise; +import io.netty.channel.embedded.EmbeddedChannel; +import io.netty.util.concurrent.ImmediateEventExecutor; +import java.util.Map; +import org.apache.commons.lang3.reflect.FieldUtils; +import org.junit.After; +import org.junit.Before; +import org.junit.Test; + +import static org.assertj.core.api.Assertions.assertThat; + +public class NettyRemotingClientCloseChannelTest { + private static final String ADDR = "127.0.0.1:9876"; + private NettyRemotingClient client; + + @Before + public void setUp() { + client = new NettyRemotingClient(new NettyClientConfig()); + } + + @After + public void tearDown() { + client.shutdown(); + } + + @SuppressWarnings("unchecked") + private Map channelTables() throws Exception { + return (Map) FieldUtils.readField(client, "channelTables", true); + } + + private ChannelFuture activeChannelFuture() { + EmbeddedChannel channel = new EmbeddedChannel(); + DefaultChannelPromise promise = new DefaultChannelPromise(channel, ImmediateEventExecutor.INSTANCE); + promise.setSuccess(); + return promise; + } + + @Test + public void testCloseChannelRemovesEntryWhenWrapperMatchesChannel() throws Exception { + ChannelFuture future = activeChannelFuture(); + NettyRemotingClient.ChannelWrapper wrapper = client.new ChannelWrapper(ADDR, future); + channelTables().put(ADDR, wrapper); + + client.closeChannel(ADDR, future.channel()); + + assertThat(channelTables()).doesNotContainKey(ADDR); + } + + @Test + public void testCloseChannelKeepsEntryWhenWrapperRecreatedForAnotherChannel() throws Exception { + // The table holds the wrapper for a freshly recreated channel. Closing an older, + // unrelated channel for the same address must not evict the current entry. + ChannelFuture current = activeChannelFuture(); + NettyRemotingClient.ChannelWrapper wrapper = client.new ChannelWrapper(ADDR, current); + channelTables().put(ADDR, wrapper); + + ChannelFuture stale = activeChannelFuture(); + client.closeChannel(ADDR, stale.channel()); + + assertThat(channelTables()).containsKey(ADDR); + assertThat(channelTables().get(ADDR)).isSameAs(wrapper); + } +}