diff --git a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java index a735f8455d3..c91601e1326 100644 --- a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java +++ b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java @@ -763,6 +763,8 @@ public HashMap> getProcess } class NettyEventExecutor extends ServiceThread { + // This private identity marker is never dispatched to the listener. + private final NettyEvent wakeupEvent = new NettyEvent(NettyEventType.IDLE, null, null); private final LinkedBlockingQueue eventQueue = new LinkedBlockingQueue<>(); public void putNettyEvent(final NettyEvent event) { @@ -775,6 +777,12 @@ public void putNettyEvent(final NettyEvent event) { } } + @Override + public void wakeup() { + super.wakeup(); + this.eventQueue.offer(wakeupEvent); + } + @Override public void run() { log.info(this.getServiceName() + " service started"); @@ -784,7 +792,7 @@ public void run() { while (!this.isStopped()) { try { NettyEvent event = this.eventQueue.poll(3000, TimeUnit.MILLISECONDS); - if (event != null && listener != null) { + if (event != null && event != wakeupEvent && listener != null) { switch (event.getType()) { case IDLE: listener.onChannelIdle(event.getRemoteAddr(), event.getChannel()); diff --git a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstractTest.java b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstractTest.java index dbbea86ea2f..ad885f82324 100644 --- a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstractTest.java +++ b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstractTest.java @@ -16,7 +16,10 @@ */ package org.apache.rocketmq.remoting.netty; +import java.time.Duration; import java.util.concurrent.Semaphore; +import java.util.concurrent.TimeUnit; +import org.apache.rocketmq.remoting.ChannelEventListener; import org.apache.rocketmq.remoting.InvokeCallback; import org.apache.rocketmq.remoting.common.SemaphoreReleaseOnlyOnce; import org.apache.rocketmq.remoting.protocol.RemotingCommand; @@ -25,8 +28,12 @@ import org.mockito.Spy; import org.mockito.junit.MockitoJUnitRunner; +import static org.awaitility.Awaitility.await; import static org.assertj.core.api.Assertions.assertThat; import static org.junit.Assert.assertNull; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @RunWith(MockitoJUnitRunner.class) @@ -137,6 +144,42 @@ public void operationFail(Throwable throwable) { assertNull(remotingAbstract.responseTable.get(dummyId)); } + @Test + public void testNettyEventExecutorShutdownWithoutPollDelay() { + TestNettyEventExecutor executor = new TestNettyEventExecutor(remotingAbstract); + executor.start(); + try { + await().atMost(Duration.ofSeconds(3)) + .until(() -> executor.getThreadState() == Thread.State.TIMED_WAITING); + + long beginTime = System.nanoTime(); + executor.shutdown(); + long elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - beginTime); + + assertThat(elapsedMillis).isLessThan(1000L); + } finally { + executor.shutdown(true); + } + } + + @Test + public void testNettyEventExecutorDoesNotDispatchWakeupEvent() { + ChannelEventListener listener = mock(ChannelEventListener.class); + when(remotingAbstract.getChannelEventListener()).thenReturn(listener); + TestNettyEventExecutor executor = new TestNettyEventExecutor(remotingAbstract); + executor.start(); + try { + executor.wakeup(); + executor.putNettyEvent(new NettyEvent(NettyEventType.CONNECT, "remoteAddr", null)); + + await().atMost(Duration.ofSeconds(3)) + .untilAsserted(() -> verify(listener).onChannelConnect("remoteAddr", null)); + verify(listener, never()).onChannelIdle(null, null); + } finally { + executor.shutdown(true); + } + } + @Test public void testProcessRequestCommand() throws InterruptedException { final Semaphore semaphore = new Semaphore(0); @@ -168,4 +211,14 @@ public void operationFail(Throwable throwable) { semaphore.acquire(1); assertThat(semaphore.availablePermits()).isEqualTo(0); } -} \ No newline at end of file + + private static class TestNettyEventExecutor extends NettyRemotingAbstract.NettyEventExecutor { + TestNettyEventExecutor(NettyRemotingAbstract remotingAbstract) { + remotingAbstract.super(); + } + + Thread.State getThreadState() { + return thread.getState(); + } + } +}