diff --git a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServer.java b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServer.java index 4f16a530c76..5221615f961 100644 --- a/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServer.java +++ b/proxy/src/main/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServer.java @@ -260,6 +260,9 @@ public void shutdown() throws Exception { this.updateOffsetExecutor.shutdown(); this.topicRouteExecutor.shutdown(); this.defaultExecutor.shutdown(); + // The constructor schedules the expired-request cleanup on this executor; leaving it running + // keeps a non-daemon thread (and its 100ms task) alive after the server has shut down. + this.timerExecutor.shutdown(); } @Override diff --git a/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServerTest.java b/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServerTest.java index acd9c1c2d44..ea5197dc5b0 100644 --- a/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServerTest.java +++ b/proxy/src/test/java/org/apache/rocketmq/proxy/remoting/RemotingProtocolServerTest.java @@ -17,6 +17,11 @@ package org.apache.rocketmq.proxy.remoting; +import java.util.concurrent.ThreadPoolExecutor; +import java.util.concurrent.ScheduledExecutorService; +import org.apache.commons.lang3.reflect.FieldUtils; +import org.apache.rocketmq.proxy.remoting.channel.RemotingChannelManager; +import org.apache.rocketmq.proxy.service.cert.TlsCertificateManager; import org.apache.rocketmq.remoting.RemotingServer; import org.apache.rocketmq.remoting.protocol.RequestHeaderRegistry; import org.junit.Test; @@ -26,9 +31,30 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.mockStatic; import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.times; public class RemotingProtocolServerTest { + @Test + public void testShutdownStopsTheExpiredRequestTimer() throws Exception { + RemotingProtocolServer protocolServer = mock(RemotingProtocolServer.class, Mockito.CALLS_REAL_METHODS); + FieldUtils.writeDeclaredField(protocolServer, "tlsCertificateManager", mock(TlsCertificateManager.class), true); + FieldUtils.writeDeclaredField(protocolServer, "remotingChannelManager", mock(RemotingChannelManager.class), true); + FieldUtils.writeDeclaredField(protocolServer, "defaultRemotingServer", mock(RemotingServer.class), true); + for (String field : new String[] {"sendMessageExecutor", "pullMessageExecutor", "heartbeatExecutor", + "updateOffsetExecutor", "topicRouteExecutor", "defaultExecutor"}) { + FieldUtils.writeDeclaredField(protocolServer, field, mock(ThreadPoolExecutor.class), true); + } + ScheduledExecutorService timerExecutor = mock(ScheduledExecutorService.class); + FieldUtils.writeDeclaredField(protocolServer, "timerExecutor", timerExecutor, true); + + protocolServer.shutdown(); + + // The constructor schedules the expired-request cleanup on this executor; a shutdown that + // leaves it running keeps a non-daemon thread (and its 100ms task) alive after the server. + verify(timerExecutor, times(1)).shutdown(); + } + @Test public void shouldInitializeRequestHeaderRegistryWhenRegisteringProcessors() { RemotingProtocolServer protocolServer =