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 aa9677b9e fix(metrics): preserve delay incidents without a timestamp
(#6187)
aa9677b9e is described below
commit aa9677b9ea2aafa592503b3c1abfdcb58f883c75
Author: zmuxuny <[email protected]>
AuthorDate: Mon Oct 12 00:17:45 2026 -0700
fix(metrics): preserve delay incidents without a timestamp (#6187)
`ApacheRocketMqBusinessMetricsCollector.java:96-98` added a
`consumer.delay.seconds` sample only when
`group.isConsumptionTimestampAvailable()`, so a group whose consume stats
report a backlog but no consumption timestamp vanished from the collection
round entirely. `NativeAlertProcessor.reconcileMissingActiveStates` reads a key
that is absent from `presentKeys` as a disappeared resource and advances it
with the synthetic AVAILABLE clear built at `:139`, which resolves a delay
alert t [...]
One new collector test and a two-instance parameterized processor test pin
this end to end through the real collector and the real processor. Removing the
`else` empties the sample in
`reportsUnavailableDelayWhenTheConsumptionTimestampIsMissingTest` and makes
`preservesDelayIncidentUntilARealTimestampConfirmsRecoveryTest` resolve both
the FIRING and the ACKED state. The second half of that test flips
`consumptionTimestampAvailable` to true with a zero delay and asserts the state
does [...]
---
.../ApacheRocketMqBusinessMetricsCollector.java | 6 +++
...ApacheRocketMqBusinessMetricsCollectorTest.java | 23 +++++++++
.../studio/ops/alert/NativeAlertProcessorTest.java | 57 ++++++++++++++++++++++
3 files changed, 86 insertions(+)
diff --git
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqBusinessMetricsCollector.java
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqBusinessMetricsCollector.java
index c0e10eaa5..3cc3bdf37 100644
---
a/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqBusinessMetricsCollector.java
+++
b/server/src/main/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqBusinessMetricsCollector.java
@@ -95,6 +95,12 @@ public class ApacheRocketMqBusinessMetricsCollector
implements BusinessMetricsCo
}
if (group.isConsumptionTimestampAvailable()) {
samples.add(delaySample(instance, group, collectedAt));
+ } else {
+ // Stats without a consumed-message timestamp leave the
delay unknown; emit it as
+ // unavailable instead of omitting the sample, so
reconciliation does not see the
+ // group's delay series vanish and resolve an active delay
alert for it.
+ samples.add(unavailable(CONSUMER_DELAY_SECONDS, instance,
group.getClusterId(),
+ Map.of("consumerGroup", group.getName()),
collectedAt, "CONSUMER_TIMESTAMP_UNAVAILABLE"));
}
samples.addAll(queueLagSamples(provider, instance, group,
collectedAt));
}
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqBusinessMetricsCollectorTest.java
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqBusinessMetricsCollectorTest.java
index f84483319..9ed3b4c64 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqBusinessMetricsCollectorTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/cluster/metrics/collectors/ApacheRocketMqBusinessMetricsCollectorTest.java
@@ -187,6 +187,29 @@ class ApacheRocketMqBusinessMetricsCollectorTest {
});
}
+ @Test
+ void reportsUnavailableDelayWhenTheConsumptionTimestampIsMissingTest() {
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ InstanceProvider provider = mock(InstanceProvider.class);
+ ConsumerGroupVO orders = group("orders", "cluster-a", 42);
+ orders.setConsumptionTimestampAvailable(false);
+ when(registry.byInstanceId("local")).thenReturn(Optional.of(provider));
+ when(provider.listConsumerGroups("local",
null)).thenReturn(List.of(orders));
+ when(provider.getGroupProgress("local",
"orders")).thenReturn(List.of(QueueProgressVO.builder()
+ .topic("orders-topic").diffTotal(17).build()));
+
+ List<MetricSample> samples = new
ApacheRocketMqBusinessMetricsCollector(registry).collect(apacheInstance());
+
+ assertThat(samples).filteredOn(sample -> sample.metricKey().equals(
+ ApacheRocketMqBusinessMetricsCollector.CONSUMER_DELAY_SECONDS))
+ .singleElement().satisfies(sample -> {
+
assertThat(sample.availability()).isEqualTo(MetricAvailability.UNAVAILABLE);
+ assertThat(sample.value()).isNull();
+ assertThat(sample.clusterId()).isEqualTo("cluster-a");
+
assertThat(sample.unavailableReason()).isEqualTo("CONSUMER_TIMESTAMP_UNAVAILABLE");
+ });
+ }
+
private static ConsumerGroupVO group(String name, String clusterId, long
lag) {
ConsumerGroupVO group = new ConsumerGroupVO();
group.setName(name);
diff --git
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
index 86860162c..ce0e77e09 100644
---
a/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
+++
b/server/src/test/java/org/apache/rocketmq/studio/ops/alert/NativeAlertProcessorTest.java
@@ -24,9 +24,12 @@ import
org.apache.rocketmq.studio.cluster.metrics.collectors.ApacheRocketMqBusin
import org.apache.rocketmq.studio.common.domain.enums.InstanceVendor;
import org.apache.rocketmq.studio.instance.InstanceVO;
import org.apache.rocketmq.studio.instance.group.ConsumerGroupVO;
+import org.apache.rocketmq.studio.instance.group.QueueProgressVO;
import org.apache.rocketmq.studio.provider.InstanceProvider;
import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.EnumSource;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionDefinition;
import org.springframework.transaction.TransactionStatus;
@@ -983,6 +986,60 @@ class NativeAlertProcessorTest {
verify(outbox, never()).enqueue(any(), any(), any());
}
+ @ParameterizedTest
+ @EnumSource(value = AlertStateStatus.class, names = {"FIRING", "ACKED"})
+ void
preservesDelayIncidentUntilARealTimestampConfirmsRecoveryTest(AlertStateStatus
status) {
+ AlertRuleVO rule = rule("local", "orders", 1);
+ rule.setMetric("consumer.delay.seconds");
+ AlertService service = mock(AlertService.class);
+
when(service.listRules(AlertDomain.BUSINESS)).thenReturn(List.of(rule));
+ Map<String, String> labels = Map.of("consumerGroup", "orders");
+ AlertStateKey key = new AlertStateKey(rule.getId(),
AlertFingerprint.of(rule.getId(), "local", labels));
+ Instant firedAt = Instant.now().minusSeconds(60);
+ AtomicReference<AlertRuleState> saved = new AtomicReference<>(new
AlertRuleState(status, 1, 120D,
+ firedAt, firedAt, firedAt, null));
+ AlertStateRepository states = mock(AlertStateRepository.class);
+ when(states.find(eq(key))).thenAnswer(invocation ->
Optional.of(saved.get()));
+ when(states.save(eq(key),
any(AlertRuleState.class))).thenAnswer(invocation -> {
+ saved.set(invocation.getArgument(1));
+ return true;
+ });
+ when(states.findActive(any(MetricCollectionScope.class),
any())).thenAnswer(invocation ->
+ saved.get().status() == AlertStateStatus.RESOLVED ? List.of()
+ : List.of(new ActiveAlertState(key, saved.get(),
"local", labels)));
+ AlertRepository alerts = mock(AlertRepository.class);
+ NativeAlertProcessor processor = processor(service, states, alerts);
+ InstanceProviderRegistry registry =
mock(InstanceProviderRegistry.class);
+ InstanceProvider provider = mock(InstanceProvider.class);
+ ConsumerGroupVO group = new ConsumerGroupVO();
+ group.setName("orders");
+ group.setConsumeStatsAvailable(true);
+ group.setTotalLag(100);
+ // A known backlog does not imply that the broker supplied a
consumption timestamp.
+ group.setConsumptionTimestampAvailable(false);
+ when(registry.byInstanceId("local")).thenReturn(Optional.of(provider));
+ when(provider.listConsumerGroups("local",
null)).thenReturn(List.of(group));
+ when(provider.getGroupProgress("local",
"orders")).thenReturn(List.of(QueueProgressVO.builder()
+ .topic("orders-topic").diffTotal(100).build()));
+ ApacheRocketMqBusinessMetricsCollector collector = new
ApacheRocketMqBusinessMetricsCollector(registry);
+ InstanceVO instance =
InstanceVO.builder().name("local").vendor(InstanceVendor.APACHE).build();
+ MetricCollectionScope scope = new
MetricCollectionScope(AlertDomain.BUSINESS, "local", collector.metricKeys());
+
+ processor.processSuccessfulCollection(scope,
collector.collect(instance));
+
+ assertThat(saved.get().status()).isEqualTo(status);
+ assertThat(saved.get().currentValue()).isEqualTo(120D);
+ verify(alerts, never()).saveAlert(any(SystemAlertVO.class));
+
+ group.setConsumptionTimestampAvailable(true);
+ group.setDelaySeconds(0);
+ processor.processSuccessfulCollection(scope,
collector.collect(instance));
+
+ assertThat(saved.get().status()).isEqualTo(AlertStateStatus.RESOLVED);
+ verify(alerts).saveAlert(org.mockito.ArgumentMatchers.argThat(event ->
+ "RESOLVED".equals(event.getTransition())));
+ }
+
private static PlatformTransactionManager mockTxManager() {
PlatformTransactionManager txManager =
mock(PlatformTransactionManager.class);
TransactionStatus status = mock(TransactionStatus.class);
