[ISSUE #10898] Wake NettyEventExecutor during shutdown

remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingAbstract.java

@@ -763,6 +763,8 @@ public HashMap<Integer, Pair<NettyRequestProcessor, ExecutorService>> 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<NettyEvent> 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());

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);
     }
-}
+
+    private static class TestNettyEventExecutor extends NettyRemotingAbstract.NettyEventExecutor {
+        TestNettyEventExecutor(NettyRemotingAbstract remotingAbstract) {
+            remotingAbstract.super();
+        }
+
+        Thread.State getThreadState() {
+            return thread.getState();
+        }
+    }
+}