This is an automated email from the ASF dual-hosted git repository.
Aias00 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/shenyu.git
The following commit(s) were added to refs/heads/master by this push:
new bb076aedc0 fix(admin): stop the master selector scheduler on cluster
shutdown (#7774) (#7782)
bb076aedc0 is described below
commit bb076aedc04e704dcd54493b45aeefbb30d9b0a7
Author: konlue <[email protected]>
AuthorDate: Mon Oct 19 09:33:16 2026 +0800
fix(admin): stop the master selector scheduler on cluster shutdown (#7774)
(#7782)
---
.../mode/cluster/service/ShenyuClusterService.java | 16 ++-
.../cluster/service/ShenyuClusterServiceTest.java | 127 +++++++++++++++++++++
2 files changed, 140 insertions(+), 3 deletions(-)
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/mode/cluster/service/ShenyuClusterService.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/mode/cluster/service/ShenyuClusterService.java
index d80de5d5db..6d89c8c5e6 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/mode/cluster/service/ShenyuClusterService.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/mode/cluster/service/ShenyuClusterService.java
@@ -31,9 +31,11 @@ import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
public class ShenyuClusterService implements ShenyuRunningModeService {
-
+
private static final Logger LOG =
LoggerFactory.getLogger(ShenyuClusterService.class);
-
+
+ private static final long SHUTDOWN_WAIT_SECONDS = 5L;
+
private final ClusterSelectMasterService shenyuClusterSelectMasterService;
private final UpstreamCheckService upstreamCheckService;
@@ -125,6 +127,14 @@ public class ShenyuClusterService implements
ShenyuRunningModeService {
@Override
public void shutdown() {
-
+ executorService.shutdownNow();
+ try {
+ if (!executorService.awaitTermination(SHUTDOWN_WAIT_SECONDS,
TimeUnit.SECONDS)) {
+ LOG.warn("master selector scheduler did not terminate within
{}s", SHUTDOWN_WAIT_SECONDS);
+ }
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ LOG.warn("interrupted while waiting for the master selector
scheduler to terminate");
+ }
}
}
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/mode/cluster/service/ShenyuClusterServiceTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/mode/cluster/service/ShenyuClusterServiceTest.java
new file mode 100644
index 0000000000..25874909b5
--- /dev/null
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/mode/cluster/service/ShenyuClusterServiceTest.java
@@ -0,0 +1,127 @@
+/*
+ * 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.shenyu.admin.mode.cluster.service;
+
+import org.apache.shenyu.admin.config.properties.ClusterProperties;
+import org.apache.shenyu.admin.service.impl.InstanceCheckService;
+import org.apache.shenyu.admin.service.impl.UpstreamCheckService;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+import org.springframework.test.util.ReflectionTestUtils;
+
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.Mockito.timeout;
+import static org.mockito.Mockito.when;
+
+/**
+ * Test case for {@link ShenyuClusterService} shutdown lifecycle.
+ */
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+public final class ShenyuClusterServiceTest {
+
+ private static final long AWAIT_MILLIS = 5000L;
+
+ @Mock
+ private ClusterSelectMasterService clusterSelectMasterService;
+
+ @Mock
+ private UpstreamCheckService upstreamCheckService;
+
+ @Mock
+ private InstanceCheckService instanceCheckService;
+
+ private ShenyuClusterService clusterService;
+
+ @BeforeEach
+ public void setUp() {
+ final ClusterProperties properties = new ClusterProperties();
+ properties.setSelectPeriod(1L);
+ clusterService = new ShenyuClusterService(clusterSelectMasterService,
upstreamCheckService, instanceCheckService, properties);
+ }
+
+ @Test
+ public void testShutdownBeforeStartDoesNotThrow() {
+ assertDoesNotThrow(clusterService::shutdown);
+ assertTrue(scheduler().isShutdown());
+ }
+
+ @Test
+ public void testShutdownStopsElectionScheduler() throws Exception {
+ final AtomicInteger attempts = new AtomicInteger();
+ when(clusterSelectMasterService.selectMaster(anyString(), anyString(),
anyString()))
+ .thenAnswer(invocation -> {
+ attempts.incrementAndGet();
+ return false;
+ });
+
+ clusterService.start("127.0.0.1", 9095, "/");
+ awaitAttempts(attempts, 1);
+
+ clusterService.shutdown();
+
+ assertTrue(scheduler().isShutdown());
+ final int afterShutdown = attempts.get();
+ Thread.sleep(2500L);
+ assertEquals(afterShutdown, attempts.get(), "no election attempt may
happen after completed shutdown");
+
+ assertDoesNotThrow(clusterService::shutdown);
+ }
+
+ @Test
+ public void testShutdownInterruptsRenewalAndReleasesLeadership() throws
Exception {
+ when(clusterSelectMasterService.selectMaster(anyString(), anyString(),
anyString())).thenReturn(true);
+ when(clusterSelectMasterService.checkMasterStatus()).thenReturn(true);
+
+ clusterService.start("127.0.0.1", 9095, "/");
+ // the first election succeeded and the renewal loop is running
+ org.mockito.Mockito.verify(clusterSelectMasterService,
timeout(AWAIT_MILLIS).atLeastOnce())
+ .selectMaster(anyString(), anyString(), anyString());
+
+ clusterService.shutdown();
+
+ // the interrupted renewal loop releases leadership through its own
finally block
+ org.mockito.Mockito.verify(clusterSelectMasterService,
timeout(3000L)).releaseMaster();
+ // and the existing error path closes the owned check services
+ org.mockito.Mockito.verify(upstreamCheckService,
timeout(3000L)).close();
+ org.mockito.Mockito.verify(instanceCheckService,
timeout(3000L)).close();
+ }
+
+ private ScheduledExecutorService scheduler() {
+ return (ScheduledExecutorService)
ReflectionTestUtils.getField(clusterService, "executorService");
+ }
+
+ private void awaitAttempts(final AtomicInteger attempts, final int
expected) throws Exception {
+ final long deadline = System.currentTimeMillis() + AWAIT_MILLIS;
+ while (attempts.get() < expected && System.currentTimeMillis() <
deadline) {
+ Thread.sleep(50L);
+ }
+ assertTrue(attempts.get() >= expected, "election attempt did not
happen in time");
+ }
+}
