This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new aea87c4e42 [ISSUE #11091] Fix remoting sub-server timeout scanning and
scheduler shutdown (#11092)
aea87c4e42 is described below
commit aea87c4e42dbce2ce9653cb55a088031ade239f1
Author: qianye <[email protected]>
AuthorDate: Thu Sep 10 13:44:33 2026 +0800
[ISSUE #11091] Fix remoting sub-server timeout scanning and scheduler
shutdown (#11092)
---
.../remoting/netty/NettyRemotingServer.java | 5 +-
.../netty/NettyRemotingServerLifecycleTest.java | 111 +++++++++++++++++++++
2 files changed, 115 insertions(+), 1 deletion(-)
diff --git
a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java
b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java
index ab2e208f48..64a1eb42d0 100644
---
a/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java
+++
b/remoting/src/main/java/org/apache/rocketmq/remoting/netty/NettyRemotingServer.java
@@ -265,7 +265,9 @@ public class NettyRemotingServer extends
NettyRemotingAbstract implements Remoti
@Override
public void run(Timeout timeout) {
try {
- NettyRemotingServer.this.scanResponseTable();
+ for (NettyRemotingAbstract server :
remotingServerTable.values()) {
+ server.scanResponseTable();
+ }
} catch (Throwable e) {
log.error("scanResponseTable exception", e);
} finally {
@@ -334,6 +336,7 @@ public class NettyRemotingServer extends
NettyRemotingAbstract implements Remoti
}
this.timer.stop();
+ this.scheduledExecutorService.shutdown();
this.eventLoopGroupBoss.shutdownGracefully();
diff --git
a/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingServerLifecycleTest.java
b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingServerLifecycleTest.java
new file mode 100644
index 0000000000..472ed77c30
--- /dev/null
+++
b/remoting/src/test/java/org/apache/rocketmq/remoting/netty/NettyRemotingServerLifecycleTest.java
@@ -0,0 +1,111 @@
+/*
+ * 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.Channel;
+import io.netty.channel.DefaultChannelPromise;
+import io.netty.util.concurrent.ImmediateEventExecutor;
+import java.net.InetSocketAddress;
+import java.util.Arrays;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.rocketmq.remoting.RemotingServer;
+import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
+import org.apache.rocketmq.remoting.protocol.RemotingCommand;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+public class NettyRemotingServerLifecycleTest {
+ private NettyRemotingServer server;
+ private ScheduledExecutorService scheduler;
+
+ @Before
+ public void setUp() throws Exception {
+ NettyServerConfig config = new NettyServerConfig();
+ config.setBindAddress("127.0.0.1");
+ config.setListenPort(0);
+ config.setServerSelectorThreads(1);
+ config.setServerWorkerThreads(1);
+ config.setServerCallbackExecutorThreads(1);
+ config.setServerAsyncSemaphoreValue(2);
+ server = new NettyRemotingServer(config);
+ scheduler = (ScheduledExecutorService)
FieldUtils.readDeclaredField(server, "scheduledExecutorService", true);
+ server.start();
+ }
+
+ @After
+ public void tearDown() throws Exception {
+ // Also clean up the scheduler when the shutdown regression test fails
on unfixed code.
+ if (scheduler != null) {
+ scheduler.shutdownNow();
+ }
+ if (server != null) {
+ server.shutdown();
+ server.eventLoopGroupBoss.terminationFuture().await(5,
TimeUnit.SECONDS);
+ server.eventLoopGroupSelector.terminationFuture().await(5,
TimeUnit.SECONDS);
+ if (server.getDefaultEventExecutorGroup() != null) {
+
server.getDefaultEventExecutorGroup().terminationFuture().await(5,
TimeUnit.SECONDS);
+ }
+ server.getCallbackExecutor().awaitTermination(5, TimeUnit.SECONDS);
+ }
+ }
+
+ @Test
+ public void testScanTimeoutsForParentAndSubServers() throws Exception {
+ // No extra listeners are needed: server-initiated requests use an
existing channel.
+ RemotingServer firstSubServer = server.newRemotingServer(1234);
+ RemotingServer secondSubServer = server.newRemotingServer(1235);
+ Channel channel = mock(Channel.class);
+ when(channel.remoteAddress()).thenReturn(new
InetSocketAddress("127.0.0.1", 4321));
+ when(channel.writeAndFlush(any())).thenAnswer(invocation ->
+ new DefaultChannelPromise(channel,
ImmediateEventExecutor.INSTANCE).setSuccess());
+
+ for (RemotingServer target : Arrays.asList(server, firstSubServer,
secondSubServer)) {
+ NettyRemotingAbstract remoting = (NettyRemotingAbstract) target;
+ CompletableFuture<ResponseFuture> expired = new
CompletableFuture<>();
+ RemotingCommand request = RemotingCommand.createRequestCommand(0,
null);
+ target.invokeAsync(channel, request, 100, expired::complete);
+ CompletableFuture<ResponseFuture> pending =
remoting.invokeImpl(channel,
+ RemotingCommand.createRequestCommand(0, null),
TimeUnit.MINUTES.toMillis(1));
+ assertThat(remoting.responseTable).hasSize(2);
+ assertThat(remoting.semaphoreAsync.availablePermits()).isZero();
+
+ ResponseFuture response = expired.get(10, TimeUnit.SECONDS);
+
assertThat(response.getCause()).isInstanceOf(RemotingTimeoutException.class);
+
assertThat(remoting.responseTable).hasSize(1).doesNotContainKey(request.getOpaque());
+
assertThat(remoting.semaphoreAsync.availablePermits()).isEqualTo(1);
+ assertThat(pending.isDone()).isFalse();
+ }
+ }
+
+ @Test
+ public void testShutdownStopsScheduler() throws Exception {
+ // Ensure its worker has started before checking termination.
+ scheduler.submit(() -> { }).get(5, TimeUnit.SECONDS);
+ server.shutdown();
+ assertThat(scheduler.isShutdown()).isTrue();
+ assertThat(scheduler.awaitTermination(5, TimeUnit.SECONDS)).isTrue();
+ }
+}
