This is an automated email from the ASF dual-hosted git repository.
lollipopjin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new 923a80e9f4 [ISSUE #11161] Fix closeChannel table eviction and
ChannelWrapper.close lock ordering (#11162)
923a80e9f4 is described below
commit 923a80e9f49cf79b3424efb3228c602afb46c35b
Author: qianye <[email protected]>
AuthorDate: Mon Sep 14 17:19:37 2026 +0800
[ISSUE #11161] Fix closeChannel table eviction and ChannelWrapper.close
lock ordering (#11162)
---
.../remoting/netty/NettyRemotingClient.java | 27 ++++---
.../netty/NettyRemotingClientCloseChannelTest.java | 82 ++++++++++++++++++++++
2 files changed, 100 insertions(+), 9 deletions(-)
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 627d6255f5..4ba18b53d9 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 class NettyRemotingClient extends
NettyRemotingAbstract implements Remoti
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 class NettyRemotingClient extends
NettyRemotingAbstract implements Remoti
}
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 0000000000..dd51fde079
--- /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<String, NettyRemotingClient.ChannelWrapper> channelTables()
throws Exception {
+ return (Map<String, NettyRemotingClient.ChannelWrapper>)
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);
+ }
+}
