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/hertzbeat.git
The following commit(s) were added to refs/heads/master by this push:
new 038ed5dea8 [fix] jitter first cyclic schedule to avoid restart
thundering herd (#4328)
038ed5dea8 is described below
commit 038ed5dea82d78def331e010ce51f25fd8b203c6
Author: NekoPunch <[email protected]>
AuthorDate: Mon Aug 17 21:55:02 2026 -0700
[fix] jitter first cyclic schedule to avoid restart thundering herd (#4328)
Co-authored-by: aias00 <[email protected]>
---
.../hertzbeat/collector/timer/TimerDispatcher.java | 18 +++++++++-
.../collector/timer/TimerDispatcherTest.java | 39 ++++++++++++++++++++++
2 files changed, 56 insertions(+), 1 deletion(-)
diff --git
a/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/timer/TimerDispatcher.java
b/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/timer/TimerDispatcher.java
index 797da5254f..ecca28714d 100644
---
a/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/timer/TimerDispatcher.java
+++
b/hertzbeat-collector/hertzbeat-collector-common/src/main/java/org/apache/hertzbeat/collector/timer/TimerDispatcher.java
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentLinkedDeque;
+import java.util.concurrent.ThreadLocalRandom;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Supplier;
@@ -116,7 +117,7 @@ public class TimerDispatcher implements TimerDispatch,
DisposableBean {
// Delay dispatcher lookup to avoid a startup cycle with
CommonDispatcher.
WheelTimerTask timerJob = new WheelTimerTask(addJob,
metricsTaskDispatchSupplier);
if (addJob.isCyclic()) {
- Long nextExecutionTime = getNextExecutionInterval(addJob);
+ long nextExecutionTime = initialCyclicDelay(addJob);
Timeout timeout = wheelTimer.newTimeout(timerJob,
nextExecutionTime, TimeUnit.SECONDS);
cancelPreviousTimeout(currentCyclicTaskMap.put(addJob.getId(),
timeout));
} else {
@@ -205,6 +206,21 @@ public class TimerDispatcher implements TimerDispatch,
DisposableBean {
}
}
+ /**
+ * Interval jobs get a random first-run phase: a restart re-adds every job
at once,
+ * and a shared phase makes them all collect at the same instant forever.
Cron jobs
+ * keep their meaningful phase; already-executed or re-added jobs keep
theirs.
+ */
+ long initialCyclicDelay(Job addJob) {
+ long nextExecutionTime = getNextExecutionInterval(addJob);
+ boolean fixedPhase =
ScheduleTypeEnum.CRON.getType().equals(addJob.getScheduleType());
+ if (!fixedPhase && addJob.getDispatchTime() <= 0
+ && !currentCyclicTaskMap.containsKey(addJob.getId()) &&
nextExecutionTime > 1) {
+ nextExecutionTime =
ThreadLocalRandom.current().nextLong(nextExecutionTime) + 1;
+ }
+ return nextExecutionTime;
+ }
+
public Long getNextExecutionInterval(Job job) {
if (ScheduleTypeEnum.CRON.getType().equals(job.getScheduleType()) &&
job.getCronExpression() != null && !job.getCronExpression().isEmpty()) {
try {
diff --git
a/hertzbeat-collector/hertzbeat-collector-common/src/test/java/org/apache/hertzbeat/collector/timer/TimerDispatcherTest.java
b/hertzbeat-collector/hertzbeat-collector-common/src/test/java/org/apache/hertzbeat/collector/timer/TimerDispatcherTest.java
index 17cd06a345..38fabf72bb 100644
---
a/hertzbeat-collector/hertzbeat-collector-common/src/test/java/org/apache/hertzbeat/collector/timer/TimerDispatcherTest.java
+++
b/hertzbeat-collector/hertzbeat-collector-common/src/test/java/org/apache/hertzbeat/collector/timer/TimerDispatcherTest.java
@@ -18,8 +18,10 @@
package org.apache.hertzbeat.collector.timer;
import java.lang.reflect.Field;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Set;
import java.util.concurrent.TimeUnit;
import org.apache.hertzbeat.collector.constants.ScheduleTypeEnum;
import org.apache.hertzbeat.collector.dispatch.MetricsTaskDispatch;
@@ -299,4 +301,41 @@ public class TimerDispatcherTest {
throw new AssertionError(e);
}
}
+
+ @Test
+ void testFirstCyclicDelayJittersWithinInterval() {
+ when(job.isCyclic()).thenReturn(true);
+ when(job.getScheduleType()).thenReturn(null);
+ when(job.getDispatchTime()).thenReturn(0L);
+ when(job.getInterval()).thenReturn(600L);
+ when(job.getId()).thenReturn(4242L);
+
+ Set<Long> delays = new HashSet<>();
+ for (int i = 0; i < 50; i++) {
+ long delay = timerDispatcher.initialCyclicDelay(job);
+ assertTrue(delay >= 1 && delay <= 600, "delay out of interval: " +
delay);
+ delays.add(delay);
+ }
+ assertTrue(delays.size() > 1, "no jitter observed across 50 samples");
+ }
+
+ @Test
+ void testExecutedJobKeepsRemainingIntervalOnReAdd() {
+ when(job.getScheduleType()).thenReturn(null);
+ when(job.getDispatchTime()).thenReturn(System.currentTimeMillis() -
10_000L);
+ when(job.getInterval()).thenReturn(600L);
+
+ long expected = timerDispatcher.getNextExecutionInterval(job);
+ long actual = timerDispatcher.initialCyclicDelay(job);
+ assertTrue(Math.abs(actual - expected) <= 1, "remaining interval must
not be jittered");
+ }
+
+ @Test
+ void testCronScheduleKeepsFixedPhase() {
+
when(job.getScheduleType()).thenReturn(ScheduleTypeEnum.CRON.getType());
+ when(job.getCronExpression()).thenReturn("0 0 3 * * ?");
+ long expected = timerDispatcher.getNextExecutionInterval(job);
+ long actual = timerDispatcher.initialCyclicDelay(job);
+ assertTrue(Math.abs(actual - expected) <= 1, "cron phase must not be
jittered");
+ }
}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
