This is an automated email from the ASF dual-hosted git repository.
lizhimins pushed a commit to branch rocketmq-studio
in repository https://gitbox.apache.org/repos/asf/rocketmq-dashboard.git
The following commit(s) were added to refs/heads/rocketmq-studio by this push:
new d03f5d144 fix(metrics): discard results from cancelled collection jobs
(#6690)
d03f5d144 is described below
commit d03f5d1441e78e61af3a02bf888d9aff047f4cfb
Author: 健康宝 <[email protected]>
AuthorDate: Thu Oct 15 18:19:11 2026 +0800
fix(metrics): discard results from cancelled collection jobs (#6690)
---
.../studio/cluster/metrics/CollectorScheduler.java | 55 +++++++++----
.../cluster/metrics/CollectorSchedulerTest.java | 96 ++++++++++++++++++++++
2 files changed, 136 insertions(+), 15 deletions(-)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java
index d5e982569..ba4d2e18d 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/CollectorScheduler.java
@@ -174,15 +174,21 @@ public class CollectorScheduler {
}
InstanceVO instance = pending.removeFirst();
try {
+ // Per-job cancellation state: Future.cancel(true) only sets
the
+ // interrupt bit, which a collector can absorb and keep going,
+ // and the pass-level heartbeat stays held while sibling jobs
+ // run. Without this flag a cancelled job's late results could
+ // still pass the lease check and reach the repository.
+ java.util.concurrent.atomic.AtomicBoolean cancelled = new
java.util.concurrent.atomic.AtomicBoolean(false);
Future<InstanceVO> future = completionService.submit(() -> {
- if (heartbeat.isHeld()) {
- collectClusterMetrics(instance, heartbeat);
- collectBusinessMetrics(instance, heartbeat);
+ if (heartbeat.isHeld() && !cancelled.get()) {
+ collectClusterMetrics(instance, heartbeat, cancelled);
+ collectBusinessMetrics(instance, heartbeat, cancelled);
}
return instance;
});
leaseJobs.add(future);
- running.add(new CollectionJob(instance, future,
now.plus(timeout)));
+ running.add(new CollectionJob(instance, future,
now.plus(timeout), cancelled));
} catch (RejectedExecutionException error) {
pending.addFirst(instance);
throw error;
@@ -231,12 +237,12 @@ public class CollectorScheduler {
while (iterator.hasNext()) {
CollectionJob job = iterator.next();
if (!passDeadline.isAfter(now)) {
- job.future().cancel(true);
+ job.cancel();
iterator.remove();
log.warn("Native metric collection pass exceeded {};
cancelling instance {}", passTimeout,
job.instance().getName());
} else if (!job.deadline().isAfter(now)) {
- job.future().cancel(true);
+ job.cancel();
iterator.remove();
log.warn("Native metric collection exceeded {} for instance {}
and was cancelled", timeout,
job.instance().getName());
@@ -246,7 +252,7 @@ public class CollectorScheduler {
private void cancelRunningJobs(List<CollectionJob> running) {
for (CollectionJob job : running) {
- job.future().cancel(true);
+ job.cancel();
}
running.clear();
}
@@ -279,15 +285,16 @@ public class CollectorScheduler {
snapshotRepository.deleteBefore(Instant.now().minus(retention));
}
- private void collectClusterMetrics(InstanceVO instance, LeaseHeartbeat
heartbeat) {
+ private void collectClusterMetrics(InstanceVO instance, LeaseHeartbeat
heartbeat,
+ java.util.concurrent.atomic.AtomicBoolean cancelled) {
for (ClusterMetricsCollector collector : clusterCollectors) {
- if (!heartbeat.isHeld()) {
+ if (!heartbeat.isHeld() || cancelled.get()) {
return;
}
try {
if (collector.supports(instance)) {
List<MetricSample> samples = collector.collect(instance);
- persist(AlertDomain.CLUSTER, instance,
collector.metricKeys(), samples, heartbeat);
+ persist(AlertDomain.CLUSTER, instance,
collector.metricKeys(), samples, heartbeat, cancelled);
}
} catch (RuntimeException error) {
log.warn("Native metric collector failed for instance {}: {}",
instance.getName(), error.getMessage());
@@ -295,15 +302,16 @@ public class CollectorScheduler {
}
}
- private void collectBusinessMetrics(InstanceVO instance, LeaseHeartbeat
heartbeat) {
+ private void collectBusinessMetrics(InstanceVO instance, LeaseHeartbeat
heartbeat,
+ java.util.concurrent.atomic.AtomicBoolean cancelled) {
for (BusinessMetricsCollector collector : businessCollectors) {
- if (!heartbeat.isHeld()) {
+ if (!heartbeat.isHeld() || cancelled.get()) {
return;
}
try {
if (collector.supports(instance)) {
List<MetricSample> samples = collector.collect(instance);
- persist(AlertDomain.BUSINESS, instance,
collector.metricKeys(), samples, heartbeat);
+ persist(AlertDomain.BUSINESS, instance,
collector.metricKeys(), samples, heartbeat, cancelled);
}
} catch (RuntimeException error) {
log.warn("Native metric collector failed for instance {}: {}",
instance.getName(), error.getMessage());
@@ -312,7 +320,17 @@ public class CollectorScheduler {
}
private void persist(AlertDomain domain, InstanceVO instance, Set<String>
declaredMetricKeys,
- List<MetricSample> samples, LeaseHeartbeat heartbeat) {
+ List<MetricSample> samples, LeaseHeartbeat heartbeat,
+ java.util.concurrent.atomic.AtomicBoolean cancelled) {
+ // A job that was cancelled or expired must not start persisting: its
+ // samples predate the cancellation decision and feeding them (and the
+ // alert processor) after the scheduler gave up on the instance
+ // contradicts that decision. The interrupt bit alone cannot enforce
+ // this — collectors may absorb it.
+ if (cancelled.get()) {
+ log.debug("Discarding native metric samples from a cancelled
collection job");
+ return;
+ }
List<MetricSample> collected = samples == null ? List.of() : samples;
Set<String> scopeMetricKeys = metricKeys(declaredMetricKeys,
collected);
if (scopeMetricKeys.isEmpty()) {
@@ -442,7 +460,14 @@ public class CollectorScheduler {
}
}
- private record CollectionJob(InstanceVO instance, Future<InstanceVO>
future, Instant deadline) {
+ private record CollectionJob(InstanceVO instance, Future<InstanceVO>
future, Instant deadline,
+ java.util.concurrent.atomic.AtomicBoolean cancelled) {
+
+ /** Marks the job cancelled for its collection/persist guards, then
interrupts the worker. */
+ void cancel() {
+ cancelled.set(true);
+ future.cancel(true);
+ }
}
private static Duration parsePositiveDuration(String value, Duration
fallback) {
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorSchedulerTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorSchedulerTest.java
index 8273169d3..9847a642c 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorSchedulerTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorSchedulerTest.java
@@ -190,6 +190,102 @@ class CollectorSchedulerTest {
scheduler.stopCollectionExecutor();
}
+ @Test
+ void discardsLateResultsFromACancelledJobTest() throws Exception {
+ AlertingProperties properties = new AlertingProperties();
+ // Four instances, parallelism 2, staggered by a gated quick job: the
+ // slow job's deadline expires while two later-submitted jobs are
+ // still running, so the pass — and its lease heartbeat — stay alive
+ // when the cancelled job's late result arrives.
+ properties.setCollectionParallelism(2);
+ properties.setCollectionTimeout("PT0.5S");
+ InstanceRepository instances = mock(InstanceRepository.class);
+ InstanceVO slow =
InstanceVO.builder().name("slow").endpoint("slow:9876").build();
+ InstanceVO quick =
InstanceVO.builder().name("quick").endpoint("quick:9876").build();
+ InstanceVO blocker1 =
InstanceVO.builder().name("blocker1").endpoint("blocker1:9876").build();
+ InstanceVO blocker2 =
InstanceVO.builder().name("blocker2").endpoint("blocker2:9876").build();
+ when(instances.findAll()).thenReturn(List.of(slow, quick, blocker1,
blocker2));
+
+ CountDownLatch releaseQuick = new CountDownLatch(1);
+ CountDownLatch releaseBlocker1 = new CountDownLatch(1);
+ CountDownLatch releaseBlocker2 = new CountDownLatch(1);
+ CountDownLatch releaseSlow = new CountDownLatch(1);
+ MetricSample slowSample = new MetricSample("nameserver.availability",
AlertDomain.CLUSTER, "slow", null,
+ null, 1D, MetricAvailability.AVAILABLE, Instant.now());
+ MetricSample blockerSample = new
MetricSample("nameserver.availability", AlertDomain.CLUSTER, "blocker1",
+ null, null, 1D, MetricAvailability.AVAILABLE, Instant.now());
+ ClusterMetricsCollector collector =
mock(ClusterMetricsCollector.class);
+ when(collector.supports(any())).thenReturn(true);
+
when(collector.metricKeys()).thenReturn(java.util.Set.of("nameserver.availability"));
+ when(collector.collect(slow)).thenAnswer(invocation -> {
+ // Absorbs the cancellation interrupt and keeps waiting, like a
+ // collector that only checks the interrupt at safe points: the
+ // late result must still be discarded. The flag is deliberately
+ // not re-set — re-setting it would make await() spin instead of
+ // parking again.
+ while (true) {
+ try {
+ releaseSlow.await();
+ break;
+ } catch (InterruptedException absorbed) {
+ // swallowed on purpose
+ }
+ }
+ return List.of(slowSample);
+ });
+ when(collector.collect(quick)).thenAnswer(invocation -> {
+ releaseQuick.await();
+ return List.of();
+ });
+ when(collector.collect(blocker1)).thenAnswer(invocation -> {
+ releaseBlocker1.await();
+ return List.of(blockerSample);
+ });
+ when(collector.collect(blocker2)).thenAnswer(invocation -> {
+ releaseBlocker2.await();
+ return List.of();
+ });
+
+ MetricSnapshotRepository snapshots =
mock(MetricSnapshotRepository.class);
+ CountDownLatch persisted = new CountDownLatch(1);
+ org.mockito.Mockito.doAnswer(invocation -> {
+ if (invocation.getArgument(0).toString().contains("blocker1")) {
+ persisted.countDown();
+ }
+ return null;
+ }).when(snapshots).saveAll(any());
+ NativeAlertProcessor processor = mock(NativeAlertProcessor.class);
+ AlertCollectionLease lease = mock(AlertCollectionLease.class);
+ when(lease.tryAcquire()).thenReturn(true);
+ ExecutorService executor = Executors.newFixedThreadPool(2);
+ CollectorScheduler scheduler = new CollectorScheduler(properties,
instances, List.of(collector), List.of(),
+ snapshots, processor, lease, executor);
+
+ Thread pass = new Thread(scheduler::collect);
+ pass.start();
+ // t=0: slow + quick start. t=400ms: quick finishes, blocker1 starts
+ // with a fresh deadline. t=500ms: slow's deadline cancels it while
+ // blocker1/blocker2 keep the pass alive; blocker2 takes the freed
+ // slot with another fresh deadline.
+ TimeUnit.MILLISECONDS.sleep(400);
+ releaseQuick.countDown();
+ TimeUnit.MILLISECONDS.sleep(200);
+ releaseBlocker1.countDown();
+ assertTrue(persisted.await(1, TimeUnit.SECONDS), "the healthy instance
still persists");
+ // The cancelled job's late result lands while the pass is alive and
+ // the lease heartbeat is held; only afterwards is the last blocker
+ // released to let the pass finish.
+ releaseSlow.countDown();
+ TimeUnit.MILLISECONDS.sleep(200);
+ releaseBlocker2.countDown();
+ pass.join(TimeUnit.SECONDS.toMillis(3));
+ assertTrue(!pass.isAlive(), "the collection pass should finish");
+ verify(snapshots, never()).saveAll(List.of(slowSample));
+ verify(processor, never()).processSuccessfulCollection(any(),
+ argThat(list -> list.contains(slowSample)));
+ scheduler.stopCollectionExecutor();
+ }
+
@Test
void appliesCollectionTimeoutToTheWholePassInsteadOfEachInstanceTest() {
AlertingProperties properties = new AlertingProperties();
