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