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 7f2dfd3ed fix(metrics,litetopic): stop cancelled jobs after
checkpoints (#7188)
7f2dfd3ed is described below
commit 7f2dfd3ed7fd9536b1b46f0efb35a84aadc451a6
Author: Yexi Xiang <[email protected]>
AuthorDate: Wed Oct 28 01:08:40 2026 +0800
fix(metrics,litetopic): stop cancelled jobs after checkpoints (#7188)
* fix(metrics): stop cancelled jobs after checkpoints (#7188)
A cancelled collection pass was only checked at the persist entry point, so
a job
cancelled while it was blocked in a checkpoint went on to write its samples
and
run the alert side effects afterwards. Re-check cancellation and the
heartbeat
lease at each of the three points - before the lease refresh, before the
snapshot
write, and before alert processing - so a late result from a cancelled or
de-leased pass is discarded instead of committed. The checks are
deliberately
redundant: each one covers a different interleaving.
* fix(litetopic): exclude unrelated group clients from summaries (#7203)
buildSummary counted every client connected to the consumer group, including
clients that had no LiteTopic under the parent topic, so the consumer count
was
larger than the number of groups actually consuming lite topics. Resolve the
client once and skip it when it has no lite topics; the same resolved
record also
supplies lastAccessTime, which lets the separate clientLastAccess lookup go
away.
Batch integration of two fixes by the same contributor; #7203 is closed as
merged
into this one. Verified on the combined branch: 65 backend tests across the
affected classes plus the full suite, and both changes are
mutation-sensitive -
removing the post-checkpoint cancellation check fails 2 of the 3 cases in
CollectorCancellationCheckpointTest, and removing the lite-topic filter
fails
listLiteTopicsShouldCountOnlyClientsSubscribedToTheParentTopicTest.
---
.../studio/cluster/metrics/CollectorScheduler.java | 14 +-
.../provider/apache/RocketMQLiteTopicProvider.java | 18 +-
.../CollectorCancellationCheckpointTest.java | 241 +++++++++++++++++++++
.../apache/RocketMQLiteTopicProviderTest.java | 37 +++-
4 files changed, 293 insertions(+), 17 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 ba4d2e18d..ee83592c9 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
@@ -339,14 +339,26 @@ public class CollectorScheduler {
MetricCollectionScope actualScope = new MetricCollectionScope(domain,
instance.getName(), scopeMetricKeys);
// Refresh immediately before persisting so a slow remote collection
cannot write after lease loss.
// This second ownership check is defense-in-depth against a heartbeat
tick racing this write.
- if (!heartbeat.isHeld() || !collectionLease.tryAcquire() ||
!heartbeat.isHeld()) {
+ if (!heartbeat.isHeld() || cancelled.get()) {
+ log.debug("Discarding native metric samples before lease refresh");
+ return;
+ }
+ if (!collectionLease.tryAcquire() || !heartbeat.isHeld()) {
heartbeat.markLost();
log.debug("Discarding native metric samples because the collection
lease was lost");
return;
}
+ if (cancelled.get()) {
+ log.debug("Discarding native metric samples after cancellation
during lease refresh");
+ return;
+ }
if (!collected.isEmpty()) {
snapshotRepository.saveAll(collected);
}
+ if (!heartbeat.isHeld() || cancelled.get()) {
+ log.debug("Discarding native metric samples after cancellation or
lease loss");
+ return;
+ }
alertProcessor.processSuccessfulCollection(actualScope, collected);
}
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProvider.java
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProvider.java
index 82cf494be..9be5f5473 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProvider.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProvider.java
@@ -237,10 +237,15 @@ public class RocketMQLiteTopicProvider implements
LiteTopicProvider {
parent.parentTopic, MAX_LITE_SESSION_SCAN);
break;
}
+ LocatedClient located = locateClient(admin, masters,
parent.parentTopic, group,
+ connection.getClientId());
+ if (located == null || located.body.getLiteTopicCount() <= 0) {
+ continue;
+ }
consumerCount++;
sessionIds.add(encodeSessionId(parent.parentTopic, group,
connection.getClientId()));
- Long active = clientLastAccess(admin, masters,
parent.parentTopic, group,
- connection.getClientId());
+ Long active = located.body.getLastAccessTime() > 0
+ ? located.body.getLastAccessTime() : null;
if (active != null && (lastActive == null || active >
lastActive)) {
lastActive = active;
}
@@ -597,15 +602,6 @@ public class RocketMQLiteTopicProvider implements
LiteTopicProvider {
return null;
}
- private Long clientLastAccess(MQAdminExt admin, List<String> masters,
String parentTopic,
- String group, String clientId) {
- LocatedClient located = locateClient(admin, masters, parentTopic,
group, clientId);
- if (located == null || located.body.getLastAccessTime() <= 0) {
- return null;
- }
- return located.body.getLastAccessTime();
- }
-
private Set<Connection> consumerConnections(MQAdminExt admin, String
group) {
try {
ConsumerConnection connection =
admin.examineConsumerConnectionInfo(group);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorCancellationCheckpointTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorCancellationCheckpointTest.java
new file mode 100644
index 000000000..bf219497c
--- /dev/null
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/CollectorCancellationCheckpointTest.java
@@ -0,0 +1,241 @@
+/*
+ * 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.rocketmq.studio.cluster.metrics;
+
+import org.apache.rocketmq.studio.instance.InstanceRepository;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.apache.rocketmq.studio.ops.alert.AlertDomain;
+import org.apache.rocketmq.studio.ops.alert.NativeAlertProcessor;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.atomic.AtomicReference;
+
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class CollectorCancellationCheckpointTest {
+ enum Checkpoint {
+ LEASE,
+ SNAPSHOT
+ }
+
+ @ParameterizedTest
+ @EnumSource(Checkpoint.class)
+ void
cancellationDuringBlockingCheckpointStopsSubsequentSideEffectsTest(Checkpoint
checkpoint) throws Exception {
+ AlertingProperties properties = new AlertingProperties();
+ properties.setCollectionParallelism(3);
+ properties.setCollectionTimeout("PT1S");
+ InstanceVO slow = instance("slow");
+ InstanceVO fastA = instance("fast-a");
+ InstanceVO fastB = instance("fast-b");
+ InstanceVO blockerA = instance("blocker-a");
+ InstanceVO blockerB = instance("blocker-b");
+ InstanceRepository instances = mock(InstanceRepository.class);
+ when(instances.findAll()).thenReturn(List.of(slow, fastA, fastB,
blockerA, blockerB));
+
+ CountDownLatch blockersStarted = new CountDownLatch(2);
+ CountDownLatch releaseBlockers = new CountDownLatch(1);
+ CountDownLatch slowFinished = new CountDownLatch(1);
+ BlockingCheckpoint checkpointGate = new BlockingCheckpoint();
+ AtomicReference<Thread> slowThread = new AtomicReference<>();
+ MetricSample slowSample = sample(slow);
+ ClusterMetricsCollector collector =
mock(ClusterMetricsCollector.class);
+ when(collector.supports(any())).thenReturn(true);
+
when(collector.metricKeys()).thenReturn(Set.of("nameserver.availability"));
+ when(collector.collect(slow)).thenAnswer(invocation -> {
+ slowThread.set(Thread.currentThread());
+ return List.of(slowSample);
+ });
+ when(collector.collect(fastA)).thenReturn(List.of());
+ when(collector.collect(fastB)).thenReturn(List.of());
+ when(collector.collect(blockerA)).thenAnswer(invocation -> {
+ blockersStarted.countDown();
+ awaitUninterruptibly(releaseBlockers);
+ return List.of();
+ });
+ when(collector.collect(blockerB)).thenAnswer(invocation -> {
+ blockersStarted.countDown();
+ awaitUninterruptibly(releaseBlockers);
+ return List.of();
+ });
+
+ AlertCollectionLease lease = mock(AlertCollectionLease.class);
+ when(lease.tryAcquire()).thenAnswer(invocation -> {
+ if (checkpoint == Checkpoint.LEASE && Thread.currentThread() ==
slowThread.get()) {
+ checkpointGate.awaitReleaseAbsorbingCancellation();
+ }
+ return true;
+ });
+ MetricSnapshotRepository snapshots =
mock(MetricSnapshotRepository.class);
+ doAnswer(invocation -> {
+ if (checkpoint == Checkpoint.SNAPSHOT && Thread.currentThread() ==
slowThread.get()) {
+ checkpointGate.awaitReleaseAbsorbingCancellation();
+ }
+ return null;
+ }).when(snapshots).saveAll(any());
+ NativeAlertProcessor processor = mock(NativeAlertProcessor.class);
+ ExecutorService executor = new ThreadPoolExecutor(3, 3, 0,
TimeUnit.MILLISECONDS,
+ new LinkedBlockingQueue<>()) {
+ @Override
+ protected void afterExecute(Runnable task, Throwable error) {
+ super.afterExecute(task, error);
+ if (Thread.currentThread() == slowThread.get()) {
+ slowFinished.countDown();
+ }
+ }
+ };
+ CollectorScheduler scheduler = new CollectorScheduler(properties,
instances, List.of(collector), List.of(),
+ snapshots, processor, lease, executor);
+ AtomicReference<Throwable> passFailure = new AtomicReference<>();
+ Thread pass = new Thread(() -> {
+ try {
+ scheduler.collect();
+ } catch (Throwable error) {
+ passFailure.set(error);
+ }
+ });
+ try {
+ pass.start();
+ assertTrue(checkpointGate.entered.await(3, TimeUnit.SECONDS),
+ "slow job must reach the blocking checkpoint");
+ assertTrue(blockersStarted.await(3, TimeUnit.SECONDS),
+ "younger jobs must keep the collection pass alive");
+ assertTrue(checkpointGate.cancelled.await(5, TimeUnit.SECONDS),
+ "the scheduler must cancel the slow job after its
timeout");
+ checkpointGate.release.countDown();
+ assertTrue(slowFinished.await(3, TimeUnit.SECONDS), "slow worker
must finish after release");
+ releaseBlockers.countDown();
+ pass.join(TimeUnit.SECONDS.toMillis(3));
+
+ if (checkpoint == Checkpoint.LEASE) {
+ verify(snapshots, never()).saveAll(List.of(slowSample));
+ } else {
+ verify(snapshots).saveAll(List.of(slowSample));
+ }
+ verify(processor, never()).processSuccessfulCollection(any(),
+ org.mockito.ArgumentMatchers.argThat(rows ->
rows.contains(slowSample)));
+ } finally {
+ checkpointGate.release.countDown();
+ releaseBlockers.countDown();
+ pass.join(TimeUnit.SECONDS.toMillis(3));
+ scheduler.stopCollectionExecutor();
+ }
+ assertNull(passFailure.get(), "collection pass must not fail");
+ }
+
+ @Test
+ void leaseLossDuringSnapshotStopsAlertProcessingTest() throws Exception {
+ AlertingProperties properties = new AlertingProperties();
+ InstanceVO instance = instance("lease-lost-during-snapshot");
+ InstanceRepository instances = mock(InstanceRepository.class);
+ when(instances.findAll()).thenReturn(List.of(instance));
+ MetricSample sample = sample(instance);
+ ClusterMetricsCollector collector =
mock(ClusterMetricsCollector.class);
+ when(collector.supports(instance)).thenReturn(true);
+
when(collector.metricKeys()).thenReturn(Set.of("nameserver.availability"));
+ when(collector.collect(instance)).thenReturn(List.of(sample));
+
+ AlertCollectionLease lease = mock(AlertCollectionLease.class);
+ when(lease.tryAcquire()).thenReturn(true);
+ when(lease.renew()).thenReturn(false);
+ ScheduledExecutorService renewals =
mock(ScheduledExecutorService.class);
+ ScheduledFuture<?> renewalTask = mock(ScheduledFuture.class);
+ AtomicReference<Runnable> renewal = new AtomicReference<>();
+ when(renewals.scheduleAtFixedRate(any(Runnable.class), anyLong(),
anyLong(), eq(TimeUnit.MILLISECONDS)))
+ .thenAnswer(invocation -> {
+ renewal.set(invocation.getArgument(0));
+ return renewalTask;
+ });
+ MetricSnapshotRepository snapshots =
mock(MetricSnapshotRepository.class);
+ doAnswer(invocation -> {
+ renewal.get().run();
+ return null;
+ }).when(snapshots).saveAll(List.of(sample));
+ NativeAlertProcessor processor = mock(NativeAlertProcessor.class);
+ CollectorScheduler scheduler = new CollectorScheduler(properties,
instances, List.of(collector), List.of(),
+ snapshots, processor, lease, newSingleThreadExecutor(),
renewals);
+ try {
+ scheduler.collect();
+ verify(snapshots).saveAll(List.of(sample));
+ verify(processor, never()).processSuccessfulCollection(any(),
any());
+ } finally {
+ scheduler.stopCollectionExecutor();
+ }
+ }
+
+ private static InstanceVO instance(String name) {
+ return InstanceVO.builder().name(name).build();
+ }
+
+ private static MetricSample sample(InstanceVO instance) {
+ return new MetricSample("nameserver.availability",
AlertDomain.CLUSTER, instance.getName(), null,
+ null, 1D, MetricAvailability.AVAILABLE, Instant.now());
+ }
+
+ private static ExecutorService newSingleThreadExecutor() {
+ return new ThreadPoolExecutor(1, 1, 0, TimeUnit.MILLISECONDS, new
LinkedBlockingQueue<>());
+ }
+
+ private static void awaitUninterruptibly(CountDownLatch release) {
+ while (true) {
+ try {
+ release.await();
+ return;
+ } catch (InterruptedException ignored) {
+ // A sibling cancellation must not abort the healthy blocker.
+ }
+ }
+ }
+
+ private static final class BlockingCheckpoint {
+ private final CountDownLatch entered = new CountDownLatch(1);
+ private final CountDownLatch cancelled = new CountDownLatch(1);
+ private final CountDownLatch release = new CountDownLatch(1);
+
+ private void awaitReleaseAbsorbingCancellation() {
+ entered.countDown();
+ while (true) {
+ try {
+ release.await();
+ return;
+ } catch (InterruptedException interrupted) {
+ cancelled.countDown();
+ }
+ }
+ }
+ }
+}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
index f0b9bcf56..2a5bb8162 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQLiteTopicProviderTest.java
@@ -206,6 +206,29 @@ class RocketMQLiteTopicProviderTest {
});
}
+ @Test
+ void listLiteTopicsShouldCountOnlyClientsSubscribedToTheParentTopicTest()
throws Exception {
+ when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
+
when(admin.getBrokerLiteInfo(BROKER_A)).thenReturn(brokerLiteInfo(PARENT, 30,
2, GROUP));
+ when(admin.getParentTopicInfo(BROKER_A,
PARENT)).thenReturn(parentTopicInfo(PARENT, 30, 2));
+ when(admin.getLiteGroupInfo(BROKER_A, GROUP, null,
1)).thenReturn(lag(7));
+ when(admin.examineConsumerConnectionInfo(GROUP))
+ .thenReturn(consumerConnection("c1", "10.0.0.1:1234", "c2",
"10.0.0.2:1234"));
+ when(admin.getLiteClientInfo(BROKER_A, PARENT, GROUP, "c1"))
+ .thenReturn(clientInfo(1, System.currentTimeMillis(),
LiteUtil.toLmqName(PARENT, "bob")));
+ // The group also has a client, but it has no LiteTopic under this
parent.
+ when(admin.getLiteClientInfo(BROKER_A, PARENT, GROUP, "c2"))
+ .thenReturn(clientInfo(0, System.currentTimeMillis()));
+
+ List<LiteTopicSummary> summaries = provider.listLiteTopics(null, null);
+
+ assertThat(summaries).singleElement().satisfies(summary -> {
+ assertThat(summary.getConsumerCount()).isEqualTo(1);
+ assertThat(summary.getSessionIds())
+
.containsExactly(RocketMQLiteTopicProvider.encodeSessionId(PARENT, GROUP,
"c1"));
+ });
+ }
+
@Test
void listLiteTopicsFiltersByPatternCaseInsensitively() throws Exception {
when(admin.examineBrokerClusterInfo()).thenReturn(cluster(BROKER_A));
@@ -635,12 +658,16 @@ class RocketMQLiteTopicProviderTest {
return body;
}
- private static ConsumerConnection consumerConnection(String clientId,
String clientAddr) {
- Connection connection = new Connection();
- connection.setClientId(clientId);
- connection.setClientAddr(clientAddr);
+ private static ConsumerConnection consumerConnection(String...
clientPairs) {
+ HashSet<Connection> connections = new HashSet<>();
+ for (int i = 0; i < clientPairs.length; i += 2) {
+ Connection connection = new Connection();
+ connection.setClientId(clientPairs[i]);
+ connection.setClientAddr(clientPairs[i + 1]);
+ connections.add(connection);
+ }
ConsumerConnection consumerConnection = new ConsumerConnection();
- consumerConnection.setConnectionSet(new HashSet<>(Set.of(connection)));
+ consumerConnection.setConnectionSet(connections);
return consumerConnection;
}
