From bfd488224c9d85ec2c62de45d49c9a163f353257 Mon Sep 17 00:00:00 2001 From: hengyuss <1183660933@qq.com> Date: Sat, 11 Jul 2026 18:25:13 +0800 Subject: [PATCH 1/4] fix: dispatch timeout metrics for all priorities with staggered delay --- .../hertzbeat/collector/dispatch/CommonDispatcher.java | 9 +++++++++ 1 file changed, 9 insertions(+) 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 f915f2ee4eb..a81baec7516 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 @@ -92,6 +92,10 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc private final String collectorIdentity; + private final ScheduledThreadPoolExecutor dispatchDelayExecutor; + + private static final long DELAY_PER_PRIORITY = 500L; + @Autowired private HertzBeatMetricsCollector metricsCollector; @@ -108,6 +112,7 @@ public CommonDispatcher(MetricsCollectorQueue jobRequestQueue, this.workerPool = workerPool; this.collectorIdentity = collectJobService.getCollectorIdentity(); this.metricsTimeoutMonitorMap = new ConcurrentHashMap<>(16); + this.dispatchDelayExecutor = new ScheduledThreadPoolExecutor(1, new ThreadFactoryBuilder().setNameFormat("dispatch-delay-%d").setDaemon(true).build()); this.start(); } @@ -182,6 +187,10 @@ private void monitorCollectTaskTimeout() { log.error("[Collect Timeout]: \n{}", metricsData); if (metricsData.getPriority() == 0) { dispatchCollectData(metricsTime.timeout, metricsTime.getMetrics(), metricsData); + } else { + long delay = (long) metricsData.getPriority() * DELAY_PER_PRIORITY; + dispatchDelayExecutor.schedule(() -> dispatchCollectData(metricsTime.timeout, metricsTime.getMetrics(), metricsData), + delay, TimeUnit.MILLISECONDS); } } } From 0f5fc7c387048f498cd022e616a1ea2727a8316e Mon Sep 17 00:00:00 2001 From: hengyuss <1183660933@qq.com> Date: Sun, 12 Jul 2026 16:16:40 +0800 Subject: [PATCH 2/4] fix: add tests --- .../dispatch/CommonDispatcherTest.java | 193 ++++++++++++++++++ 1 file changed, 193 insertions(+) create mode 100644 hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/dispatch/CommonDispatcherTest.java 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 00000000000..d04372176d5 --- /dev/null +++ b/hertzbeat-collector/hertzbeat-collector-collector/src/test/java/org/apache/hertzbeat/collector/dispatch/CommonDispatcherTest.java @@ -0,0 +1,193 @@ +package org.apache.hertzbeat.collector.dispatch; + +import org.apache.hertzbeat.collector.dispatch.entrance.internal.CollectJobService; +import org.apache.hertzbeat.collector.dispatch.unit.UnitConvert; +import org.apache.hertzbeat.collector.timer.TimerDispatch; +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.queue.CommonDataQueue; +import org.apache.hertzbeat.common.timer.Timeout; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.Mockito; +import org.mockito.MockitoAnnotations; + +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.List; +import java.util.Map; + +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +class CommonDispatcherTest { + + @Mock + private MetricsCollectorQueue jobRequestQueue; + + @Mock + private TimerDispatch timerDispatch; + + @Mock + private CommonDataQueue commonDataQueue; + + @Mock + private WorkerPool workerPool; + + @Mock + private CollectJobService collectJobService; + + @Mock + private Timeout timeout; + + @Mock + private WheelTimerTask wheelTimerTask; + + private List unitConvertList = List.of(); + + private CommonDispatcher dispatcher; + + @BeforeEach + void setUp() { + MockitoAnnotations.openMocks(this); + when(collectJobService.getCollectorIdentity()).thenReturn("test-collector"); + doNothing().when(workerPool).executeLongRunning(any()); + dispatcher = new CommonDispatcher(jobRequestQueue, timerDispatch, commonDataQueue, workerPool, collectJobService, unitConvertList); + } + + @Test + void dispatchCollectDataExecute_When_Priority0MetricsTimeout() throws Exception { + Job job = buildCycleJobWithMetrics(); + job.constructPriorMetrics(); + Metrics availabilityMetrics = job.getMetrics().get(0); + + when(timeout.task()).thenReturn(wheelTimerTask); + when(wheelTimerTask.getJob()).thenReturn(job); + when(timeout.isCancelled()).thenReturn(false); + + Map metricsTimeoutMonitorMap = getMetricsTimeoutMonitorMap(); + CommonDispatcher.MetricsTime metricsTime = new CommonDispatcher.MetricsTime( + System.currentTimeMillis() - 300_000L, + availabilityMetrics, + timeout + ); + metricsTimeoutMonitorMap.put(job.getId() + "-" + availabilityMetrics.getName(), metricsTime); + + invokeMonitorCollectTaskTimeout(); + + verify(commonDataQueue).sendMetricsData(any()); + } + + @Test + void dispatchCollectDataExecute_When_Priority_gt_0_MetricsTimeout() throws Exception { + Job job = buildCycleJobWithMetrics(); + job.constructPriorMetrics(); + Metrics availabilityMetrics = job.getMetrics().get(0); + Metrics cpu = job.getMetrics().get(1); + job.getNextCollectMetrics(availabilityMetrics, false); + + when(timeout.task()).thenReturn(wheelTimerTask); + when(wheelTimerTask.getJob()).thenReturn(job); + when(timeout.isCancelled()).thenReturn(false); + + Map metricsTimeoutMonitorMap = getMetricsTimeoutMonitorMap(); + CommonDispatcher.MetricsTime metricsTime = new CommonDispatcher.MetricsTime( + System.currentTimeMillis() - 300_000L, + cpu, + timeout + ); + metricsTimeoutMonitorMap.put(job.getId() + "-" + cpu.getName(), metricsTime); + + invokeMonitorCollectTaskTimeout(); + + verify(commonDataQueue, never()).sendMetricsData(any()); + + verify(commonDataQueue, Mockito.timeout(2000)).sendMetricsData(any()); + + } + + + @Test + void monitorPipelineNotStall_When_Priority0MetricsTimeout() throws Exception { + Job job = buildCycleJobWithMetrics(); + job.constructPriorMetrics(); + Metrics availabilityMetrics = job.getMetrics().get(0); + + when(timeout.task()).thenReturn(wheelTimerTask); + when(wheelTimerTask.getJob()).thenReturn(job); + when(timeout.isCancelled()).thenReturn(false); + + Map metricsTimeoutMonitorMap = getMetricsTimeoutMonitorMap(); + CommonDispatcher.MetricsTime metricsTime = new CommonDispatcher.MetricsTime( + System.currentTimeMillis() - 300_000L, + availabilityMetrics, + timeout + ); + metricsTimeoutMonitorMap.put(job.getId() + "-" + availabilityMetrics.getName(), metricsTime); + + invokeMonitorCollectTaskTimeout(); + + verify(timerDispatch).cyclicJob(any(WheelTimerTask.class)); + } + + @Test + void monitorPipelineNotStall_When_Priority_gt_0_MetricsTimeout() throws Exception { + Job job = buildCycleJobWithMetrics(); + job.constructPriorMetrics(); + Metrics availabilityMetrics = job.getMetrics().get(0); + Metrics cpu = job.getMetrics().get(1); + job.getNextCollectMetrics(availabilityMetrics, false); + + when(timeout.task()).thenReturn(wheelTimerTask); + when(wheelTimerTask.getJob()).thenReturn(job); + when(timeout.isCancelled()).thenReturn(false); + + Map metricsTimeoutMonitorMap = getMetricsTimeoutMonitorMap(); + CommonDispatcher.MetricsTime metricsTime = new CommonDispatcher.MetricsTime( + System.currentTimeMillis() - 300_000L, + cpu, + timeout + ); + metricsTimeoutMonitorMap.put(job.getId() + "-" + cpu.getName(), metricsTime); + + invokeMonitorCollectTaskTimeout(); + + verify(jobRequestQueue, Mockito.timeout(2000)).addJob(any(MetricsCollect.class)); + } + + @SuppressWarnings("unchecked") + private Map getMetricsTimeoutMonitorMap() throws Exception { + Field field = CommonDispatcher.class.getDeclaredField("metricsTimeoutMonitorMap"); + field.setAccessible(true); + return (Map) field.get(dispatcher); + } + + private void invokeMonitorCollectTaskTimeout() throws Exception { + Method method = CommonDispatcher.class.getDeclaredMethod("monitorCollectTaskTimeout"); + method.setAccessible(true); + method.invoke(dispatcher); + } + + private Job buildCycleJobWithMetrics() { + return Job.builder() + .id(1L) + .monitorId(100L) + .app("linux") + .isCyclic(true) + .isSd(false) + .configmap(List.of()) + .metrics(List.of( + Metrics.builder().name("availability").priority((byte) 0).interval(60L).build(), + Metrics.builder().name("cpu").priority((byte) 1).interval(60L).build(), + Metrics.builder().name("memory").priority((byte) 2).interval(60L).build() + )) + .build(); + } + + +} From 09d38cbca551ba40c62c471e96c848ad0560c186 Mon Sep 17 00:00:00 2001 From: hengyuss <1183660933@qq.com> Date: Sun, 12 Jul 2026 17:25:31 +0800 Subject: [PATCH 3/4] fix: add license header --- .../dispatch/CommonDispatcherTest.java | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) 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 index d04372176d5..3135eb3f4b4 100644 --- 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 @@ -1,3 +1,20 @@ +/* + * 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; From 2108218917169b1fdc18a5b16776dfc8fb416160 Mon Sep 17 00:00:00 2001 From: hengyuss <1183660933@qq.com> Date: Mon, 13 Jul 2026 10:06:10 +0800 Subject: [PATCH 4/4] fix: remove delay handle and test --- .../collector/dispatch/CommonDispatcher.java | 13 +---- .../dispatch/CommonDispatcherTest.java | 54 +++++-------------- 2 files changed, 13 insertions(+), 54 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 a81baec7516..d496c0bd4ba 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 @@ -92,10 +92,6 @@ public class CommonDispatcher implements MetricsTaskDispatch, CollectDataDispatc private final String collectorIdentity; - private final ScheduledThreadPoolExecutor dispatchDelayExecutor; - - private static final long DELAY_PER_PRIORITY = 500L; - @Autowired private HertzBeatMetricsCollector metricsCollector; @@ -112,7 +108,6 @@ public CommonDispatcher(MetricsCollectorQueue jobRequestQueue, this.workerPool = workerPool; this.collectorIdentity = collectJobService.getCollectorIdentity(); this.metricsTimeoutMonitorMap = new ConcurrentHashMap<>(16); - this.dispatchDelayExecutor = new ScheduledThreadPoolExecutor(1, new ThreadFactoryBuilder().setNameFormat("dispatch-delay-%d").setDaemon(true).build()); this.start(); } @@ -185,13 +180,7 @@ private void monitorCollectTaskTimeout() { .setTime(System.currentTimeMillis()) .setCode(CollectRep.Code.TIMEOUT).setMsg("collect timeout").build(); log.error("[Collect Timeout]: \n{}", metricsData); - if (metricsData.getPriority() == 0) { - dispatchCollectData(metricsTime.timeout, metricsTime.getMetrics(), metricsData); - } else { - long delay = (long) metricsData.getPriority() * DELAY_PER_PRIORITY; - dispatchDelayExecutor.schedule(() -> dispatchCollectData(metricsTime.timeout, metricsTime.getMetrics(), metricsData), - delay, TimeUnit.MILLISECONDS); - } + dispatchCollectData(metricsTime.timeout, metricsTime.getMetrics(), metricsData); } } } catch (Exception e) { 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 index 3135eb3f4b4..656ba0a1849 100644 --- 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 @@ -17,6 +17,15 @@ package org.apache.hertzbeat.collector.dispatch; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.doNothing; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; + +import java.lang.reflect.Field; +import java.lang.reflect.Method; +import java.util.List; +import java.util.Map; import org.apache.hertzbeat.collector.dispatch.entrance.internal.CollectJobService; import org.apache.hertzbeat.collector.dispatch.unit.UnitConvert; import org.apache.hertzbeat.collector.timer.TimerDispatch; @@ -28,20 +37,8 @@ import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.Mock; -import org.mockito.Mockito; import org.mockito.MockitoAnnotations; -import java.lang.reflect.Field; -import java.lang.reflect.Method; -import java.util.List; -import java.util.Map; - -import static org.mockito.ArgumentMatchers.any; -import static org.mockito.Mockito.doNothing; -import static org.mockito.Mockito.never; -import static org.mockito.Mockito.verify; -import static org.mockito.Mockito.when; - class CommonDispatcherTest { @Mock @@ -74,7 +71,8 @@ void setUp() { MockitoAnnotations.openMocks(this); when(collectJobService.getCollectorIdentity()).thenReturn("test-collector"); doNothing().when(workerPool).executeLongRunning(any()); - dispatcher = new CommonDispatcher(jobRequestQueue, timerDispatch, commonDataQueue, workerPool, collectJobService, unitConvertList); + dispatcher = new CommonDispatcher(jobRequestQueue, timerDispatch, commonDataQueue, workerPool, + collectJobService, unitConvertList); } @Test @@ -100,34 +98,6 @@ void dispatchCollectDataExecute_When_Priority0MetricsTimeout() throws Exception verify(commonDataQueue).sendMetricsData(any()); } - @Test - void dispatchCollectDataExecute_When_Priority_gt_0_MetricsTimeout() throws Exception { - Job job = buildCycleJobWithMetrics(); - job.constructPriorMetrics(); - Metrics availabilityMetrics = job.getMetrics().get(0); - Metrics cpu = job.getMetrics().get(1); - job.getNextCollectMetrics(availabilityMetrics, false); - - when(timeout.task()).thenReturn(wheelTimerTask); - when(wheelTimerTask.getJob()).thenReturn(job); - when(timeout.isCancelled()).thenReturn(false); - - Map metricsTimeoutMonitorMap = getMetricsTimeoutMonitorMap(); - CommonDispatcher.MetricsTime metricsTime = new CommonDispatcher.MetricsTime( - System.currentTimeMillis() - 300_000L, - cpu, - timeout - ); - metricsTimeoutMonitorMap.put(job.getId() + "-" + cpu.getName(), metricsTime); - - invokeMonitorCollectTaskTimeout(); - - verify(commonDataQueue, never()).sendMetricsData(any()); - - verify(commonDataQueue, Mockito.timeout(2000)).sendMetricsData(any()); - - } - @Test void monitorPipelineNotStall_When_Priority0MetricsTimeout() throws Exception { @@ -174,7 +144,7 @@ void monitorPipelineNotStall_When_Priority_gt_0_MetricsTimeout() throws Exceptio invokeMonitorCollectTaskTimeout(); - verify(jobRequestQueue, Mockito.timeout(2000)).addJob(any(MetricsCollect.class)); + verify(jobRequestQueue).addJob(any(MetricsCollect.class)); } @SuppressWarnings("unchecked")