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 8c21742a45 fix(admin): activate future time-window scaling policies
(#7706) (#7886)
8c21742a45 is described below
commit 8c21742a45f03f3b63148167f4050a1e803a77de
Author: imDwAaY <[email protected]>
AuthorDate: Wed Oct 21 10:53:33 2026 +0800
fix(admin): activate future time-window scaling policies (#7706) (#7886)
ScaleService.executeScaling() evaluated the time window only at call time,
and it was only called at startup and after a committed policy update, so an
enabled policy whose beginTime was still in the future never activated when
the window opened. The window end was unhandled as well, and an enabled
time-window policy shadowed lower-priority policies even outside its window.
Reconcile the scheduling state on every evaluation:
- TaskSchedulerManager.registerOnceTask() schedules a one-shot callback that
removes its own registry entry before running, so a callback that triggers
reconciliation cannot cancel or interrupt itself.
- A still-future window registers an activation callback at beginTime; an
open window applies the target and registers a deactivation callback at
endTime.
- Time-window policies are eligible only inside their window, so lower
priority policies take over outside it.
- The applied (policyId, replicaCount) target is cached and only updated
after
Kubernetes reports success, so repeated reconciliation does not rewrite
the
same replica count and failed requests are retried.
Fixes #7706
---
.../admin/scale/scaler/KubernetesScaler.java | 6 +-
.../shenyu/admin/scale/scaler/ScaleService.java | 136 +++++++++++++++++---
.../scale/scaler/dynamic/TaskSchedulerManager.java | 33 ++++-
.../admin/scale/scaler/KubernetesScalerTest.java | 3 +-
.../admin/scale/scaler/ScaleServiceTest.java | 137 ++++++++++++++++++++-
.../scaler/dynamic/TaskSchedulerManagerTest.java | 20 +++
6 files changed, 308 insertions(+), 27 deletions(-)
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/scale/scaler/KubernetesScaler.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/scale/scaler/KubernetesScaler.java
index 27afa35ccf..8af9344499 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/scale/scaler/KubernetesScaler.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/scale/scaler/KubernetesScaler.java
@@ -67,8 +67,9 @@ public class KubernetesScaler {
* scale by number.
*
* @param replicaCount replicaCount
+ * @return true when the scale request succeeds
*/
- public void scaleByNum(final int replicaCount) {
+ public boolean scaleByNum(final int replicaCount) {
try {
V1Scale scale = new V1Scale();
scale.setSpec(new V1ScaleSpec().replicas(replicaCount));
@@ -86,9 +87,10 @@ public class KubernetesScaler {
// fieldValidation
null
);
-
+ return true;
} catch (ApiException e) {
LOG.error("Failed to scale deployment. Cause: {}", e.getMessage());
+ return false;
}
}
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/scale/scaler/ScaleService.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/scale/scaler/ScaleService.java
index f2377f26ae..75b8d53ae0 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/scale/scaler/ScaleService.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/scale/scaler/ScaleService.java
@@ -24,18 +24,32 @@ import
org.apache.shenyu.admin.scale.scaler.cache.ScalePolicyCache;
import org.apache.shenyu.admin.scale.scaler.dynamic.TaskSchedulerManager;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
-import java.util.Date;
+import java.time.Clock;
+import java.time.Instant;
import java.util.Comparator;
+import java.util.Date;
import java.util.Objects;
+/**
+ * Reconciles Kubernetes scaling state with the active scaling policy.
+ *
+ * <p>Policy {@code 1} sets a fixed replica count, policy {@code 2} applies
during a time window,
+ * and policy {@code 3} is metric-driven. Reconciliation runs at startup,
after policy updates,
+ * and from one-shot window-boundary tasks.</p>
+ */
@Component
public class ScaleService {
private static final String DYNAMIC_TASK_NAME = "DynamicScalingTask";
+ private static final String SCALE_WINDOW_ACTIVATION_TASK =
"ScaleWindowActivation";
+
+ private static final String SCALE_WINDOW_DEACTIVATION_TASK =
"ScaleWindowDeactivation";
+
private static final Logger LOG =
LoggerFactory.getLogger(ScaleService.class);
private final TaskSchedulerManager schedulerManager;
@@ -48,36 +62,60 @@ public class ScaleService {
private final KubernetesScaler kubernetesScaler;
+ private final Clock clock;
+
+ private AppliedTarget appliedTarget;
+
+ @Autowired
public ScaleService(final MetricsMonitor metricsMonitor,
final TaskSchedulerManager schedulerManager,
final ScalePolicyCache scalePolicyCache,
final ScaleProperties scaleProperties,
final KubernetesScaler kubernetesScaler) {
+ this(metricsMonitor, schedulerManager, scalePolicyCache,
scaleProperties,
+ kubernetesScaler, Clock.systemDefaultZone());
+ }
+
+ ScaleService(final MetricsMonitor metricsMonitor,
+ final TaskSchedulerManager schedulerManager,
+ final ScalePolicyCache scalePolicyCache,
+ final ScaleProperties scaleProperties,
+ final KubernetesScaler kubernetesScaler,
+ final Clock clock) {
this.schedulerManager = schedulerManager;
this.metricsMonitor = metricsMonitor;
this.scalePolicyCache = scalePolicyCache;
this.scaleProperties = scaleProperties;
this.kubernetesScaler = kubernetesScaler;
+ this.clock = clock;
}
/**
- * execute scaling.
+ * Reconcile the currently effective scaling policy.
+ *
+ * <p>This method is synchronized because startup, HTTP request, and
scheduler threads may invoke it.</p>
*/
- public void executeScaling() {
- ScalePolicyDO activePolicy = getActivePolicy();
+ public synchronized void executeScaling() {
+ cancelWindowTasks();
+ Instant now = clock.instant();
+ scheduleFutureWindowActivation(now);
+ ScalePolicyDO activePolicy = getActivePolicy(now);
if (Objects.nonNull(activePolicy)) {
switch (activePolicy.getId()) {
case "1":
stopDynamicTask();
- kubernetesScaler.scaleByNum(activePolicy.getNum());
+ applyTarget(activePolicy);
break;
case "2":
stopDynamicTask();
- if (isWithinTimeRange(activePolicy)) {
- kubernetesScaler.scaleByNum(activePolicy.getNum());
+ Instant end = toInstant(activePolicy.getEndTime());
+ applyTarget(activePolicy);
+ if (Objects.nonNull(end)) {
+
schedulerManager.registerOnceTask(SCALE_WINDOW_DEACTIVATION_TASK,
this::executeScaling, end);
}
break;
case "3":
+ appliedTarget = null;
startDynamicTask(scaleProperties.getMonitorInterval());
break;
default:
@@ -85,18 +123,22 @@ public class ScaleService {
throw new IllegalStateException("Unknown scaling policy: "
+ activePolicy.getId());
}
} else {
+ appliedTarget = null;
stopDynamicTask();
}
}
/**
- * get active policy.
+ * Select the enabled policy that is eligible at the supplied instant and
has the smallest sort value.
+ * Time-window policies participate only while their window is open.
*
- * @return ScalePolicyDO
+ * @param now reconciliation instant
+ * @return the active policy, or {@code null} when none is eligible
*/
- private ScalePolicyDO getActivePolicy() {
+ private ScalePolicyDO getActivePolicy(final Instant now) {
return scalePolicyCache.getAllPolicies().stream()
- .filter(policy -> policy.getStatus() == 1)
+ .filter(policy ->
Integer.valueOf(1).equals(policy.getStatus()))
+ .filter(policy -> isEligible(policy, now))
.min(Comparator.comparingInt(ScalePolicyDO::getSort))
.orElseGet(() -> {
LOG.warn("No active scaling policy found.");
@@ -105,14 +147,71 @@ public class ScaleService {
}
/**
- * isWithinTimeRange.
+ * Determine whether a policy can participate at the supplied instant.
+ * Non-window policies are always eligible; a null window boundary is
unbounded on that side.
*
- * @param policy policy
- * @return boolean
+ * @param policy scaling policy
+ * @param now reconciliation instant
+ * @return whether the policy is eligible
*/
- private boolean isWithinTimeRange(final ScalePolicyDO policy) {
- Date now = new Date();
- return now.after(policy.getBeginTime()) &&
now.before(policy.getEndTime());
+ private boolean isEligible(final ScalePolicyDO policy, final Instant now) {
+ if (!"2".equals(policy.getId())) {
+ return true;
+ }
+ Instant begin = toInstant(policy.getBeginTime());
+ Instant end = toInstant(policy.getEndTime());
+ return (Objects.isNull(begin) || !now.isBefore(begin))
+ && (Objects.isNull(end) || now.isBefore(end));
+ }
+
+ /**
+ * Register a future window boundary even when another policy currently
wins, so priority is
+ * re-evaluated when that window opens.
+ *
+ * @param now reconciliation instant
+ */
+ private void scheduleFutureWindowActivation(final Instant now) {
+ scalePolicyCache.getAllPolicies().stream()
+ .filter(policy -> "2".equals(policy.getId()))
+ .filter(policy ->
Integer.valueOf(1).equals(policy.getStatus()))
+ .filter(policy -> Objects.nonNull(policy.getBeginTime()))
+ .filter(policy ->
now.isBefore(policy.getBeginTime().toInstant()))
+ .filter(policy -> Objects.isNull(policy.getEndTime())
+ ||
policy.getBeginTime().toInstant().isBefore(policy.getEndTime().toInstant()))
+ .min(Comparator.comparingInt(ScalePolicyDO::getSort))
+ .ifPresent(policy -> schedulerManager.registerOnceTask(
+ SCALE_WINDOW_ACTIVATION_TASK, this::executeScaling,
policy.getBeginTime().toInstant()));
+ }
+
+ /**
+ * Apply a target idempotently and cache it only after Kubernetes reports
success.
+ * The Kubernetes call is skipped when the same policy and replica count
were already applied.
+ *
+ * @param policy active scaling policy
+ */
+ private void applyTarget(final ScalePolicyDO policy) {
+ AppliedTarget target = new AppliedTarget(policy.getId(),
policy.getNum());
+ if (!target.equals(appliedTarget) &&
kubernetesScaler.scaleByNum(policy.getNum())) {
+ appliedTarget = target;
+ }
+ }
+
+ /**
+ * Cancel both window-boundary tasks before recomputing their schedule.
+ */
+ private void cancelWindowTasks() {
+ schedulerManager.cancelMonitorTask(SCALE_WINDOW_ACTIVATION_TASK);
+ schedulerManager.cancelMonitorTask(SCALE_WINDOW_DEACTIVATION_TASK);
+ }
+
+ /**
+ * Convert a nullable date to an instant.
+ *
+ * @param date date to convert
+ * @return the instant, or {@code null} when the date is null
+ */
+ private Instant toInstant(final Date date) {
+ return Objects.isNull(date) ? null : date.toInstant();
}
/**
@@ -141,4 +240,7 @@ public class ScaleService {
public void stopDynamicTask() {
schedulerManager.cancelMonitorTask(DYNAMIC_TASK_NAME);
}
+
+ private record AppliedTarget(String policyId, int num) {
+ }
}
diff --git
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/scale/scaler/dynamic/TaskSchedulerManager.java
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/scale/scaler/dynamic/TaskSchedulerManager.java
index 1beca803fd..1fc74d7016 100644
---
a/shenyu-admin/src/main/java/org/apache/shenyu/admin/scale/scaler/dynamic/TaskSchedulerManager.java
+++
b/shenyu-admin/src/main/java/org/apache/shenyu/admin/scale/scaler/dynamic/TaskSchedulerManager.java
@@ -22,11 +22,13 @@ import org.apache.shenyu.admin.scale.config.ScaleProperties;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.stereotype.Component;
-import java.time.Instant;
import java.time.Duration;
+import java.time.Instant;
import java.util.Objects;
import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.atomic.AtomicReference;
@Component
public class TaskSchedulerManager {
@@ -62,6 +64,35 @@ public class TaskSchedulerManager {
taskMap.put(taskName, future);
}
+ /**
+ * Register a task to run once at the specified instant.
+ *
+ * @param taskName taskName
+ * @param task task
+ * @param fireAt fireAt
+ */
+ public void registerOnceTask(final String taskName, final Runnable task,
final Instant fireAt) {
+ cancelMonitorTask(taskName);
+ AtomicReference<ScheduledFuture<?>> futureReference = new
AtomicReference<>();
+ CountDownLatch registered = new CountDownLatch(1);
+ Runnable oneShotTask = () -> {
+ try {
+ registered.await();
+ } catch (InterruptedException e) {
+ Thread.currentThread().interrupt();
+ return;
+ }
+ ScheduledFuture<?> future = futureReference.get();
+ if (taskMap.remove(taskName, future)) {
+ task.run();
+ }
+ };
+ ScheduledFuture<?> future = scheduler.schedule(oneShotTask, fireAt);
+ futureReference.set(future);
+ taskMap.put(taskName, future);
+ registered.countDown();
+ }
+
/**
* cancelMonitorTask.
*
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/scale/scaler/KubernetesScalerTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/scale/scaler/KubernetesScalerTest.java
index a4f6c16dae..ed30e0a97b 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/scale/scaler/KubernetesScalerTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/scale/scaler/KubernetesScalerTest.java
@@ -35,6 +35,7 @@ import java.util.Optional;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.ArgumentMatchers.isNull;
@@ -68,7 +69,7 @@ public final class KubernetesScalerTest {
*/
@Test
public void testScaleByNumSetsSpecBeforeReplace() throws Exception {
- kubernetesScaler.scaleByNum(3);
+ assertTrue(kubernetesScaler.scaleByNum(3));
final ArgumentCaptor<V1Scale> captor =
ArgumentCaptor.forClass(V1Scale.class);
verify(appsV1Api).replaceNamespacedDeploymentScale(
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/scale/scaler/ScaleServiceTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/scale/scaler/ScaleServiceTest.java
index 76d28fff3b..c29db70d86 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/scale/scaler/ScaleServiceTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/scale/scaler/ScaleServiceTest.java
@@ -24,18 +24,25 @@ import
org.apache.shenyu.admin.scale.scaler.cache.ScalePolicyCache;
import org.apache.shenyu.admin.scale.scaler.dynamic.TaskSchedulerManager;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
+import java.time.Clock;
+import java.time.Instant;
+import java.util.Date;
import java.util.List;
import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.ArgumentMatchers.eq;
import static org.mockito.Mockito.inOrder;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.verifyNoInteractions;
import static org.mockito.Mockito.verifyNoMoreInteractions;
+import static org.mockito.Mockito.when;
/**
* Tests for scaling policy transitions and monitoring task cancellation.
@@ -48,20 +55,28 @@ public final class ScaleServiceTest {
private ScalePolicyCache policyCache;
+ private Clock clock;
+
private ScaleService service;
+ private Instant now;
+
@BeforeEach
public void setUp() {
schedulerManager = mock(TaskSchedulerManager.class);
kubernetesScaler = mock(KubernetesScaler.class);
policyCache = new ScalePolicyCache();
+ clock = mock(Clock.class);
+ now = Instant.parse("2026-01-15T12:00:00Z");
+ when(clock.instant()).thenAnswer(invocation -> now);
+ when(kubernetesScaler.scaleByNum(anyInt())).thenReturn(true);
final ScaleProperties properties = new ScaleProperties();
properties.setMonitorInterval(1000L);
service = new ScaleService(
mock(MetricsMonitor.class), schedulerManager,
- policyCache, properties, kubernetesScaler);
+ policyCache, properties, kubernetesScaler, clock);
}
@Test
@@ -74,11 +89,11 @@ public final class ScaleServiceTest {
service.executeScaling();
service.executeScaling();
- final InOrder order = inOrder(schedulerManager);
- order.verify(schedulerManager).registerMonitorTask(
+ verify(schedulerManager).registerMonitorTask(
eq("DynamicScalingTask"), any(Runnable.class), eq(1000L));
- order.verify(schedulerManager,
times(2)).cancelMonitorTask("DynamicScalingTask");
- order.verifyNoMoreInteractions();
+ verify(schedulerManager,
times(2)).cancelMonitorTask("DynamicScalingTask");
+ verifyWindowTaskCancellations(3);
+ verifyNoMoreInteractions(schedulerManager);
verifyNoInteractions(kubernetesScaler);
}
@@ -88,6 +103,7 @@ public final class ScaleServiceTest {
service.executeScaling();
verify(schedulerManager).cancelMonitorTask("DynamicScalingTask");
+ verifyWindowTaskCancellations(1);
verifyNoMoreInteractions(schedulerManager);
verifyNoInteractions(kubernetesScaler);
}
@@ -102,6 +118,7 @@ public final class ScaleServiceTest {
verify(schedulerManager).registerMonitorTask(
eq("DynamicScalingTask"), any(Runnable.class), eq(1000L));
+ verifyWindowTaskCancellations(1);
verifyNoMoreInteractions(schedulerManager);
verifyNoInteractions(kubernetesScaler);
}
@@ -121,7 +138,115 @@ public final class ScaleServiceTest {
eq("DynamicScalingTask"), any(Runnable.class), eq(1000L));
order.verify(schedulerManager).cancelMonitorTask("DynamicScalingTask");
order.verify(kubernetesScaler).scaleByNum(2);
- order.verifyNoMoreInteractions();
+ verifyWindowTaskCancellations(2);
+ }
+
+ @Test
+ public void testFutureWindowSchedulesActivation() {
+ final Instant begin = now.plusSeconds(60);
+ policyCache.initialize(List.of(windowPolicy(3, begin,
begin.plusSeconds(60))));
+
+ service.executeScaling();
+
+ verify(kubernetesScaler, never()).scaleByNum(anyInt());
+ verify(schedulerManager).registerOnceTask(
+ eq("ScaleWindowActivation"), any(Runnable.class), eq(begin));
+ }
+
+ @Test
+ public void testActiveWindowAppliesTargetAndSchedulesDeactivation() {
+ final Instant end = now.plusSeconds(60);
+ policyCache.initialize(List.of(windowPolicy(3, now.minusSeconds(60),
end)));
+
+ service.executeScaling();
+
+ verify(kubernetesScaler).scaleByNum(3);
+ verify(schedulerManager).registerOnceTask(
+ eq("ScaleWindowDeactivation"), any(Runnable.class), eq(end));
+ }
+
+ @Test
+ public void testRepeatedReconciliationDoesNotReapplyTarget() {
+ final ScalePolicyDO fixedPolicy = policy("1", 1, 1);
+ fixedPolicy.setNum(3);
+ policyCache.initialize(List.of(fixedPolicy));
+
+ service.executeScaling();
+ service.executeScaling();
+
+ verify(kubernetesScaler).scaleByNum(3);
+ }
+
+ @Test
+ public void testNewWindowReappliesTargetAfterPeriodWithNoActivePolicy() {
+ final Instant firstEnd = now.plusSeconds(60);
+ policyCache.initialize(List.of(windowPolicy(5, now.minusSeconds(60),
firstEnd)));
+
+ service.executeScaling();
+ verify(kubernetesScaler).scaleByNum(5);
+
+ now = firstEnd.plusSeconds(1);
+ service.executeScaling();
+ verify(kubernetesScaler).scaleByNum(5);
+
+ policyCache.updatePolicy(windowPolicy(5, now.minusSeconds(1),
now.plusSeconds(60)));
+ service.executeScaling();
+
+ verify(kubernetesScaler, times(2)).scaleByNum(5);
+ }
+
+ @Test
+ public void testExpiredWindowFallsBackToFixedPolicy() {
+ final ScalePolicyDO windowPolicy = windowPolicy(5,
now.minusSeconds(120), now.minusSeconds(60));
+ windowPolicy.setSort(2);
+ final ScalePolicyDO fixedPolicy = policy("1", 1, 3);
+ fixedPolicy.setNum(2);
+ policyCache.initialize(List.of(windowPolicy, fixedPolicy));
+
+ service.executeScaling();
+
+ verify(kubernetesScaler).scaleByNum(2);
+ verify(schedulerManager, never()).registerOnceTask(any(), any(),
any());
+ }
+
+ @Test
+ public void testUnboundedWindowAppliesTarget() {
+ final ScalePolicyDO windowPolicy = policy("2", 1, 2);
+ windowPolicy.setNum(4);
+ policyCache.initialize(List.of(windowPolicy));
+
+ service.executeScaling();
+
+ verify(kubernetesScaler).scaleByNum(4);
+ verify(schedulerManager, never()).registerOnceTask(any(), any(),
any());
+ }
+
+ @Test
+ public void testActivationCallbackAppliesTargetAtWindowStart() {
+ final Instant begin = now.plusSeconds(60);
+ policyCache.initialize(List.of(windowPolicy(3, begin,
begin.plusSeconds(60))));
+ final ArgumentCaptor<Runnable> callback =
ArgumentCaptor.forClass(Runnable.class);
+
+ service.executeScaling();
+ verify(schedulerManager).registerOnceTask(eq("ScaleWindowActivation"),
callback.capture(), eq(begin));
+
+ now = begin;
+ callback.getValue().run();
+
+ verify(kubernetesScaler).scaleByNum(3);
+ }
+
+ private ScalePolicyDO windowPolicy(final int num, final Instant begin,
final Instant end) {
+ final ScalePolicyDO policy = policy("2", 1, 2);
+ policy.setNum(num);
+ policy.setBeginTime(Date.from(begin));
+ policy.setEndTime(Date.from(end));
+ return policy;
+ }
+
+ private void verifyWindowTaskCancellations(final int times) {
+ verify(schedulerManager,
times(times)).cancelMonitorTask("ScaleWindowActivation");
+ verify(schedulerManager,
times(times)).cancelMonitorTask("ScaleWindowDeactivation");
}
private ScalePolicyDO policy(final String id, final int status, final int
sort) {
diff --git
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/scale/scaler/dynamic/TaskSchedulerManagerTest.java
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/scale/scaler/dynamic/TaskSchedulerManagerTest.java
index 73800d4c60..f618f18116 100644
---
a/shenyu-admin/src/test/java/org/apache/shenyu/admin/scale/scaler/dynamic/TaskSchedulerManagerTest.java
+++
b/shenyu-admin/src/test/java/org/apache/shenyu/admin/scale/scaler/dynamic/TaskSchedulerManagerTest.java
@@ -20,16 +20,20 @@ package org.apache.shenyu.admin.scale.scaler.dynamic;
import org.apache.shenyu.admin.scale.config.ScaleProperties;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
import
org.springframework.context.annotation.AnnotationConfigApplicationContext;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.test.util.ReflectionTestUtils;
+import java.time.Instant;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ScheduledThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicInteger;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -74,4 +78,20 @@ class TaskSchedulerManagerTest {
assertTrue(executor.isShutdown(), "Closing the context should shut
down the monitor scheduler");
assertTrue(executor.awaitTermination(5, TimeUnit.SECONDS), "Monitor
threads should terminate after context close");
}
+
+ @Test
+ void testRegisterOnceTaskFiresOnceAndRemovesEntry() throws
InterruptedException {
+ final CountDownLatch taskFired = new CountDownLatch(1);
+ final AtomicInteger invocationCount = new AtomicInteger();
+
+ manager.registerOnceTask("once", () -> {
+ invocationCount.incrementAndGet();
+ taskFired.countDown();
+ }, Instant.now().plusMillis(20));
+
+ assertTrue(taskFired.await(5, TimeUnit.SECONDS), "The one-shot task
should fire");
+ assertEquals(1, invocationCount.get());
+ final Object taskMap = ReflectionTestUtils.getField(manager,
"taskMap");
+ assertFalse(((java.util.Map<?, ?>) taskMap).containsKey("once"));
+ }
}
