[ISSUE #11161] Fix closeChannel table eviction and ChannelWrapper.close lock ordering (#11162)

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());
             }
         }
     }

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);
+    }
+}