This is an automated email from the ASF dual-hosted git repository.
tomsun28 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 3c87b977ed [bugfix] Fix duplicate collector job timeouts (#4169)
3c87b977ed is described below
commit 3c87b977ed421dee3b2c300028e250865ab5469a
Author: hutiefang76 <[email protected]>
AuthorDate: Sat Jul 4 16:52:52 2026 +0800
[bugfix] Fix duplicate collector job timeouts (#4169)
Co-authored-by: hutiefang <[email protected]>
Co-authored-by: Tomsun28 <[email protected]>
---
.../hertzbeat/collector/timer/TimerDispatcher.java | 10 ++-
.../collector/timer/TimerDispatcherTest.java | 91 +++++++++++++++++++++-
2 files changed, 98 insertions(+), 3 deletions(-)
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 a8d6a60002..c75e14220e 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
@@ -118,14 +118,14 @@ public class TimerDispatcher implements TimerDispatch,
DisposableBean {
if (addJob.isCyclic()) {
Long nextExecutionTime = getNextExecutionInterval(addJob);
Timeout timeout = wheelTimer.newTimeout(timerJob,
nextExecutionTime, TimeUnit.SECONDS);
- currentCyclicTaskMap.put(addJob.getId(), timeout);
+ cancelPreviousTimeout(currentCyclicTaskMap.put(addJob.getId(),
timeout));
} else {
for (Metrics metric : addJob.getMetrics()) {
metric.setInterval(0L);
}
addJob.setIntervals(new ConcurrentLinkedDeque<>(List.of(0L)));
Timeout timeout = wheelTimer.newTimeout(timerJob,
addJob.getInterval(), TimeUnit.SECONDS);
- currentTempTaskMap.put(addJob.getId(), timeout);
+ cancelPreviousTimeout(currentTempTaskMap.put(addJob.getId(),
timeout));
eventListeners.put(addJob.getId(), eventListener);
}
}
@@ -199,6 +199,12 @@ public class TimerDispatcher implements TimerDispatch,
DisposableBean {
this.wheelTimer.stop();
}
+ private void cancelPreviousTimeout(Timeout previousTimeout) {
+ if (previousTimeout != null) {
+ previousTimeout.cancel();
+ }
+ }
+
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 4554ca8f0a..c0d609a687 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
@@ -17,8 +17,14 @@
package org.apache.hertzbeat.collector.timer;
+import java.lang.reflect.Field;
+import java.util.List;
+import java.util.Map;
import org.apache.hertzbeat.collector.constants.ScheduleTypeEnum;
+import
org.apache.hertzbeat.collector.dispatch.entrance.internal.CollectResponseEventListener;
import org.apache.hertzbeat.common.entity.job.Job;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.timer.Timeout;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
import org.mockito.InjectMocks;
@@ -26,6 +32,9 @@ import org.mockito.Mock;
import org.mockito.MockitoAnnotations;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.Mockito.when;
/**
@@ -179,4 +188,84 @@ public class TimerDispatcherTest {
// Verify - Should fall back to interval value
assertEquals(180L, result);
}
-}
\ No newline at end of file
+
+ @Test
+ void testAddCyclicJobCancelsPreviousTimeoutForSameJob() {
+ Job firstJob = Job.builder()
+ .id(1L)
+ .app("test")
+ .isCyclic(true)
+ .configmap(List.of())
+ .metrics(List.of(Metrics.builder().interval(60L).build()))
+ .build();
+ timerDispatcher.addJob(firstJob, null);
+
+ Timeout firstTimeout = currentCyclicTaskMap().get(1L);
+
+ Job updatedJob = Job.builder()
+ .id(1L)
+ .app("test")
+ .isCyclic(true)
+ .configmap(List.of())
+ .metrics(List.of(Metrics.builder().interval(60L).build()))
+ .build();
+ timerDispatcher.addJob(updatedJob, null);
+
+ Timeout updatedTimeout = currentCyclicTaskMap().get(1L);
+ assertNotSame(firstTimeout, updatedTimeout);
+ assertTrue(firstTimeout.isCancelled());
+ assertFalse(updatedTimeout.isCancelled());
+ }
+
+ @Test
+ void testAddTemporaryJobCancelsPreviousTimeoutForSameJob() {
+ Job firstJob = Job.builder()
+ .id(1L)
+ .app("test")
+ .isCyclic(false)
+ .configmap(List.of())
+ .metrics(List.of(Metrics.builder().interval(60L).build()))
+ .build();
+ timerDispatcher.addJob(firstJob, new CollectResponseEventListener() {
+ });
+
+ Timeout firstTimeout = currentTempTaskMap().get(1L);
+
+ Job updatedJob = Job.builder()
+ .id(1L)
+ .app("test")
+ .isCyclic(false)
+ .configmap(List.of())
+ .metrics(List.of(Metrics.builder().interval(60L).build()))
+ .build();
+ timerDispatcher.addJob(updatedJob, new CollectResponseEventListener() {
+ });
+
+ Timeout updatedTimeout = currentTempTaskMap().get(1L);
+ assertNotSame(firstTimeout, updatedTimeout);
+ assertTrue(firstTimeout.isCancelled());
+ assertFalse(updatedTimeout.isCancelled());
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map<Long, Timeout> currentCyclicTaskMap() {
+ try {
+ Field field =
TimerDispatcher.class.getDeclaredField("currentCyclicTaskMap");
+ field.setAccessible(true);
+ return (Map<Long, Timeout>) field.get(timerDispatcher);
+ } catch (ReflectiveOperationException e) {
+ throw new AssertionError(e);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private Map<Long, Timeout> currentTempTaskMap() {
+ try {
+ Field field =
TimerDispatcher.class.getDeclaredField("currentTempTaskMap");
+ field.setAccessible(true);
+ return (Map<Long, Timeout>) field.get(timerDispatcher);
+ } catch (ReflectiveOperationException e) {
+ throw new AssertionError(e);
+ }
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
