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 9bd2796119 [fix] cancel the overwritten timeout in cyclicJob to fix
HashedWheelTimeout memory leak (#4201)
9bd2796119 is described below
commit 9bd2796119b558cff2834dd35e3d21f2eb4a53a4
Author: Duansg <[email protected]>
AuthorDate: Sun Jul 12 23:05:40 2026 +0800
[fix] cancel the overwritten timeout in cyclicJob to fix HashedWheelTimeout
memory leak (#4201)
---
.../hertzbeat/collector/timer/TimerDispatcher.java | 2 +-
.../collector/timer/TimerDispatcherTest.java | 35 ++++++++++++++++++++--
2 files changed, 34 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 c75e14220e..797da5254f 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
@@ -140,7 +140,7 @@ public class TimerDispatcher implements TimerDispatch,
DisposableBean {
// whether is the job has been canceled
if (currentCyclicTaskMap.containsKey(jobId)) {
Timeout timeout = wheelTimer.newTimeout(timerTask, interval,
TimeUnit.SECONDS);
- currentCyclicTaskMap.put(timerTask.getJob().getId(), timeout);
+
cancelPreviousTimeout(currentCyclicTaskMap.put(timerTask.getJob().getId(),
timeout));
}
}
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 c0d609a687..17cd06a345 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
@@ -20,7 +20,9 @@ package org.apache.hertzbeat.collector.timer;
import java.lang.reflect.Field;
import java.util.List;
import java.util.Map;
+import java.util.concurrent.TimeUnit;
import org.apache.hertzbeat.collector.constants.ScheduleTypeEnum;
+import org.apache.hertzbeat.collector.dispatch.MetricsTaskDispatch;
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;
@@ -136,7 +138,7 @@ public class TimerDispatcherTest {
long interval = 600L; // 10 minutes in seconds
long spendTime = 300000L; // 5 minutes in milliseconds
long dispatchTime = System.currentTimeMillis() - spendTime;
-
+
when(job.getScheduleType()).thenReturn(ScheduleTypeEnum.INTERVAL.getType());
when(job.getDispatchTime()).thenReturn(dispatchTime);
when(job.getInterval()).thenReturn(interval);
@@ -159,7 +161,7 @@ public class TimerDispatcherTest {
long interval = 600L; // 10 minutes in seconds
long spendTime = 1200000L; // 20 minutes in milliseconds (more than
interval)
long dispatchTime = System.currentTimeMillis() - spendTime;
-
+
when(job.getScheduleType()).thenReturn(ScheduleTypeEnum.INTERVAL.getType());
when(job.getDispatchTime()).thenReturn(dispatchTime);
when(job.getInterval()).thenReturn(interval);
@@ -247,6 +249,35 @@ public class TimerDispatcherTest {
assertFalse(updatedTimeout.isCancelled());
}
+ @Test
+ void testCyclicJobCancelsPreviousTimeoutForSameJob() {
+ Job job = Job.builder()
+ .id(1L)
+ .app("test")
+ .isCyclic(true)
+ .configmap(List.of())
+ .metrics(List.of(Metrics.builder().interval(60L).build()))
+ .build();
+ timerDispatcher.addJob(job, null);
+ Timeout t0 = currentCyclicTaskMap().get(1L);
+
+ WheelTimerTask timerTask = new WheelTimerTask(job,
(MetricsTaskDispatch) timeout -> {
+ });
+
+ timerDispatcher.cyclicJob(timerTask, 60L, TimeUnit.SECONDS);
+ Timeout t1 = currentCyclicTaskMap().get(1L);
+
+ timerDispatcher.cyclicJob(timerTask, 60L, TimeUnit.SECONDS);
+ Timeout t2 = currentCyclicTaskMap().get(1L);
+
+ assertNotSame(t0, t1);
+ assertNotSame(t1, t2);
+ assertFalse(t2.isCancelled());
+
+ assertTrue(t0.isCancelled(), "T0 orphaned: overwritten by cyclicJob
but never cancelled");
+ assertTrue(t1.isCancelled(), "T1 orphaned: overwritten by cyclicJob
but never cancelled");
+ }
+
@SuppressWarnings("unchecked")
private Map<Long, Timeout> currentCyclicTaskMap() {
try {
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
