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 2525797134 [fix] prevent duplicate dispatch when collector task times
out (#4342)
2525797134 is described below
commit 2525797134b1b3a7f4f0b99c1a3e4b2b1acbe813
Author: Tanay Paul <[email protected]>
AuthorDate: Sun Aug 23 13:39:12 2026 +0530
[fix] prevent duplicate dispatch when collector task times out (#4342)
---
.../collector/dispatch/CommonDispatcher.java | 34 ++--
.../collector/dispatch/MetricsCollect.java | 4 +
.../collector/dispatch/CommonDispatcherTest.java | 218 +++++++++++++++++++++
3 files changed, 244 insertions(+), 12 deletions(-)
diff --git
a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/dispatch/CommonDispatcher.java
b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/dispatch/CommonDispatcher.java
index 686d3fe6c1..7c01127df0 100644
---
a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/dispatch/CommonDispatcher.java
+++
b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/dispatch/CommonDispatcher.java
@@ -158,18 +158,8 @@ public class CommonDispatcher implements
MetricsTaskDispatch, CollectDataDispatc
for (Map.Entry<String, MetricsTime> entry :
metricsTimeoutMonitorMap.entrySet()) {
MetricsTime metricsTime = entry.getValue();
if (metricsTime.getStartTime() < deadline) {
- // Metrics collection timeout
- MetricsTime removedMetricsTime =
metricsTimeoutMonitorMap.remove(entry.getKey());
- if (removedMetricsTime == null) {
- continue;
- }
WheelTimerTask timerJob = (WheelTimerTask)
metricsTime.getTimeout().task();
Job job = timerJob.getJob();
- // timeout metrics
- if (metricsCollector != null) {
- long duration = System.currentTimeMillis() -
removedMetricsTime.getStartTime();
- metricsCollector.recordCollectMetrics(job, duration,
"timeout");
- }
CollectRep.MetricsData metricsData =
CollectRep.MetricsData.newBuilder()
.setId(job.getMonitorId())
@@ -184,7 +174,17 @@ public class CommonDispatcher implements
MetricsTaskDispatch, CollectDataDispatc
.setCode(CollectRep.Code.TIMEOUT).setMsg("collect
timeout").build();
log.error("[Collect Timeout]: \n{}", metricsData);
if (metricsData.getPriority() == 0) {
+ // dispatchCollectData removes the map entry as a
once-wins gate;
+ // cancel afterwards so cyclicJob() inside it still
fires normally.
dispatchCollectData(metricsTime.timeout,
metricsTime.getMetrics(), metricsData);
+ metricsTime.getTimeout().cancel();
+ } else {
+ // remove the stale entry and cancel the in-flight
collect so a
+ // late result does not produce a duplicate dispatch.
+ MetricsTime removed =
metricsTimeoutMonitorMap.remove(entry.getKey());
+ if (removed != null) {
+ metricsTime.getTimeout().cancel();
+ }
}
}
}
@@ -229,12 +229,22 @@ public class CommonDispatcher implements
MetricsTaskDispatch, CollectDataDispatc
}
MetricsTime metricsTime = metricsTimeoutMonitorMap.remove(monitorKey);
- // job completed metrics
if (metricsTime != null && metricsCollector != null) {
long duration = System.currentTimeMillis() -
metricsTime.getStartTime();
- String status = metricsData.getCode() == CollectRep.Code.SUCCESS ?
"success" : "fail";
+ String status;
+ if (metricsData.getCode() == CollectRep.Code.SUCCESS) {
+ status = "success";
+ } else if (metricsData.getCode() == CollectRep.Code.TIMEOUT) {
+ status = "timeout";
+ } else {
+ status = "fail";
+ }
metricsCollector.recordCollectMetrics(job, duration, status);
}
+ // if the entry was already removed by the timeout monitor, skip the
duplicate result.
+ if (metricsTime == null && !metrics.isHasSubTask() &&
metrics.getPrometheus() == null) {
+ return;
+ }
if (metrics.isHasSubTask()) {
boolean isLastTask = metrics.consumeSubTaskResponse(metricsData);
if (isLastTask) {
diff --git
a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/dispatch/MetricsCollect.java
b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/dispatch/MetricsCollect.java
index d896e1404b..670eb8c030 100644
---
a/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/dispatch/MetricsCollect.java
+++
b/hertzbeat-collector/hertzbeat-collector-collector/src/main/java/org/apache/hertzbeat/collector/dispatch/MetricsCollect.java
@@ -183,6 +183,10 @@ public class MetricsCollect implements Runnable,
Comparable<MetricsCollect> {
*/
metricsData =
PrometheusAutoCollectImpl.getInstance().collect(response, metrics);
validateResponse(metricsData == null ? null :
metricsData.stream().findFirst().orElse(null));
+ // if the timeout monitor already cancelled this cycle, skip the
late result.
+ if (fastFailed()) {
+ return;
+ }
collectDataDispatch.dispatchCollectData(timeout, metrics,
metricsData);
return;
}
diff --git
a/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/dispatch/CommonDispatcherTest.java
b/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/dispatch/CommonDispatcherTest.java
new file mode 100644
index 0000000000..7f0e5375a6
--- /dev/null
+++
b/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/dispatch/CommonDispatcherTest.java
@@ -0,0 +1,218 @@
+/*
+ * 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.hertzbeat.collector.dispatch;
+
+import
org.apache.hertzbeat.collector.dispatch.entrance.internal.CollectJobService;
+import org.apache.hertzbeat.collector.timer.WheelTimerTask;
+import org.apache.hertzbeat.common.entity.job.Job;
+import org.apache.hertzbeat.common.entity.job.Metrics;
+import org.apache.hertzbeat.common.entity.message.CollectRep;
+import org.apache.hertzbeat.common.timer.Timeout;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+
+import java.lang.reflect.Field;
+import java.lang.reflect.Method;
+import java.util.Collections;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+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;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+/**
+ * Test case for {@link CommonDispatcher}.
+ * Regression coverage for issue #4203 (duplicate dispatch on collection
timeout).
+ */
+class CommonDispatcherTest {
+
+ private static final long JOB_ID = 1L;
+
+ /**
+ * Minimal {@link Timeout} stub that tracks cancellation state and returns
+ * a mock {@link WheelTimerTask} backed by a real {@link Job}.
+ */
+ private static class CancellableTimeout implements Timeout {
+
+ private volatile boolean cancelled = false;
+
+ @Override
+ public boolean cancel() {
+ cancelled = true;
+ return true;
+ }
+
+ @Override
+ public boolean isCancelled() {
+ return cancelled;
+ }
+
+ @Override
+ public boolean isExpired() {
+ return false;
+ }
+
+ @Override
+ public org.apache.hertzbeat.common.timer.Timer timer() {
+ return null;
+ }
+
+ @Override
+ public org.apache.hertzbeat.common.timer.TimerTask task() {
+ WheelTimerTask task = mock(WheelTimerTask.class);
+ Job job = Job.builder()
+ .id(JOB_ID)
+ .monitorId(JOB_ID)
+ .tenantId(0L)
+ .app("test-app")
+ .labels(Collections.emptyMap())
+ .annotations(Collections.emptyMap())
+ .metadata(Collections.emptyMap())
+ .build();
+ when(task.getJob()).thenReturn(job);
+ return task;
+ }
+ }
+
+ private CancellableTimeout timeout;
+
+ @BeforeEach
+ void setUp() {
+ timeout = new CancellableTimeout();
+ }
+
+ @Test
+ void timeoutMonitor_priorityZero_dispatchesOnceAndCancels() throws
Exception {
+ long expiredStart = System.currentTimeMillis() - 300_000L;
+ Metrics availability =
Metrics.builder().name("availability").priority((byte) 0).build();
+ CommonDispatcher.MetricsTime entry =
+ new CommonDispatcher.MetricsTime(expiredStart, availability,
timeout);
+
+ AtomicInteger dispatchCount = new AtomicInteger(0);
+ Map<String, CommonDispatcher.MetricsTime> monitorMap = new
ConcurrentHashMap<>();
+ monitorMap.put(JOB_ID + "-availability", entry);
+
+ CommonDispatcher dispatcher = buildDispatcher(monitorMap,
dispatchCount);
+ invokeMonitorCollectTaskTimeout(dispatcher);
+
+ assertEquals(1, dispatchCount.get(), "dispatchCollectData must be
called exactly once");
+ assertTrue(timeout.isCancelled(), "Timeout must be cancelled after
dispatch");
+ assertTrue(monitorMap.isEmpty(), "Map must be empty after timeout is
handled");
+ }
+
+ @Test
+ void timeoutMonitor_secondScan_doesNotDoubleDispatch() throws Exception {
+ long expiredStart = System.currentTimeMillis() - 300_000L;
+ Metrics availability =
Metrics.builder().name("availability").priority((byte) 0).build();
+ CommonDispatcher.MetricsTime entry =
+ new CommonDispatcher.MetricsTime(expiredStart, availability,
timeout);
+
+ AtomicInteger dispatchCount = new AtomicInteger(0);
+ Map<String, CommonDispatcher.MetricsTime> monitorMap = new
ConcurrentHashMap<>();
+ monitorMap.put(JOB_ID + "-availability", entry);
+
+ CommonDispatcher dispatcher = buildDispatcher(monitorMap,
dispatchCount);
+
+ invokeMonitorCollectTaskTimeout(dispatcher);
+ assertEquals(1, dispatchCount.get(), "First scan must dispatch once");
+
+ invokeMonitorCollectTaskTimeout(dispatcher);
+ assertEquals(1, dispatchCount.get(), "Second scan must not produce a
duplicate dispatch");
+ }
+
+ @Test
+ void timeoutMonitor_nonZeroPriority_cancelsWithoutDispatching() throws
Exception {
+ long expiredStart = System.currentTimeMillis() - 300_000L;
+ Metrics cpu = Metrics.builder().name("cpu").priority((byte) 1).build();
+ CommonDispatcher.MetricsTime entry =
+ new CommonDispatcher.MetricsTime(expiredStart, cpu, timeout);
+
+ AtomicInteger dispatchCount = new AtomicInteger(0);
+ Map<String, CommonDispatcher.MetricsTime> monitorMap = new
ConcurrentHashMap<>();
+ monitorMap.put(JOB_ID + "-cpu", entry);
+
+ CommonDispatcher dispatcher = buildDispatcher(monitorMap,
dispatchCount);
+ invokeMonitorCollectTaskTimeout(dispatcher);
+
+ assertEquals(0, dispatchCount.get(), "Non-zero priority timeout must
not dispatch");
+ assertTrue(timeout.isCancelled(), "Timeout must be cancelled");
+ assertTrue(monitorMap.isEmpty(), "Map must be empty after timeout is
handled");
+ }
+
+ @Test
+ void timeoutMonitor_nonExpiredEntry_isLeftUntouched() throws Exception {
+ long recentStart = System.currentTimeMillis() - 60_000L;
+ Metrics availability =
Metrics.builder().name("availability").priority((byte) 0).build();
+ CommonDispatcher.MetricsTime entry =
+ new CommonDispatcher.MetricsTime(recentStart, availability,
timeout);
+
+ AtomicInteger dispatchCount = new AtomicInteger(0);
+ Map<String, CommonDispatcher.MetricsTime> monitorMap = new
ConcurrentHashMap<>();
+ monitorMap.put(JOB_ID + "-availability", entry);
+
+ CommonDispatcher dispatcher = buildDispatcher(monitorMap,
dispatchCount);
+ invokeMonitorCollectTaskTimeout(dispatcher);
+
+ assertEquals(0, dispatchCount.get(), "Non-expired entry must not be
dispatched");
+ assertFalse(timeout.isCancelled(), "Non-expired timeout must not be
cancelled");
+ assertFalse(monitorMap.isEmpty(), "Non-expired entry must remain in
the map");
+ }
+
+ private CommonDispatcher buildDispatcher(
+ Map<String, CommonDispatcher.MetricsTime> monitorMap,
+ AtomicInteger dispatchCount) throws Exception {
+
+ CollectJobService jobService = mock(CollectJobService.class);
+ when(jobService.getCollectorIdentity()).thenReturn("test-collector");
+ WorkerPool workerPool = mock(WorkerPool.class);
+
+ CommonDispatcher dispatcher = new CommonDispatcher(
+ null, null, null, workerPool, jobService, null) {
+
+ @Override
+ public void start() {
+ }
+
+ @Override
+ public void dispatchCollectData(Timeout t, Metrics m,
CollectRep.MetricsData data) {
+ WheelTimerTask task = (WheelTimerTask) t.task();
+ String key = task.getJob().getId() + "-" + m.getName();
+ if (monitorMap.remove(key) == null) {
+ return;
+ }
+ dispatchCount.incrementAndGet();
+ }
+ };
+
+ Field mapField =
CommonDispatcher.class.getDeclaredField("metricsTimeoutMonitorMap");
+ mapField.setAccessible(true);
+ mapField.set(dispatcher, monitorMap);
+
+ return dispatcher;
+ }
+
+ private void invokeMonitorCollectTaskTimeout(CommonDispatcher dispatcher)
throws Exception {
+ Method method =
CommonDispatcher.class.getDeclaredMethod("monitorCollectTaskTimeout");
+ method.setAccessible(true);
+ method.invoke(dispatcher);
+ }
+}
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
