You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by li...@apache.org on 2026/10/15 10:00:16 UTC
(rocketmq-dashboard) branch rocketmq-studio updated: fix(consumer): avoid delay alerts for caught-up idle groups (#6652)
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 8592c946e fix(consumer): avoid delay alerts for caught-up idle groups (#6652)
8592c946e is described below
commit 8592c946e7435d7e2218b0e69c60e100ecdf3296
Author: zmuxuny <xi...@gmail.com>
AuthorDate: Thu Oct 15 03:00:10 2026 -0700
fix(consumer): avoid delay alerts for caught-up idle groups (#6652)
---
.../provider/apache/RocketMQAdminClientImpl.java | 4 +-
.../provider/apache/RocketMQMetadataProvider.java | 4 +-
.../apache/RocketMQAdminClientImplTest.java | 24 ++++++++
.../apache/RocketMQMetadataProviderTest.java | 66 ++++++++++++++++++++++
4 files changed, 96 insertions(+), 2 deletions(-)
diff --git a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
index c5ab62e94..3f1462856 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImpl.java
@@ -206,7 +206,9 @@ public class RocketMQAdminClientImpl implements AdminClient {
}
vo.setTotalLag(lagUnknown ? ConsumerLagResolver.UNKNOWN : totalLag);
if (newestConsumedTimestamp > 0) {
- long delaySeconds = (System.currentTimeMillis() - newestConsumedTimestamp) / 1000;
+ // An old last-message timestamp measures inactivity once every queue is caught up.
+ long delaySeconds = !lagUnknown && totalLag == 0 ? 0
+ : (System.currentTimeMillis() - newestConsumedTimestamp) / 1000;
vo.setDelaySeconds((int) Math.max(delaySeconds, 0));
vo.setConsumptionTimestampAvailable(true);
}
diff --git a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
index 50bce4e40..6c73400b7 100644
--- a/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
+++ b/server/src/main/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProvider.java
@@ -423,7 +423,9 @@ public class RocketMQMetadataProvider implements MetadataProvider {
int delaySeconds = 0;
boolean timestampAvailable = false;
if (newestConsumedTimestamp > 0) {
- long elapsedSeconds = (System.currentTimeMillis() - newestConsumedTimestamp) / 1000;
+ // An old last-message timestamp measures inactivity once every queue is caught up.
+ long elapsedSeconds = !lagUnknown && totalLag == 0 ? 0
+ : (System.currentTimeMillis() - newestConsumedTimestamp) / 1000;
delaySeconds = (int) Math.max(elapsedSeconds, 0);
timestampAvailable = true;
}
diff --git a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
index 809a99b09..25e722511 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQAdminClientImplTest.java
@@ -67,6 +67,7 @@ import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.NullAndEmptySource;
import org.junit.jupiter.params.provider.ValueSource;
+import org.junit.jupiter.params.provider.CsvSource;
import org.junit.jupiter.api.extension.ExtendWith;
import org.mockito.ArgumentCaptor;
import org.mockito.Mock;
@@ -327,6 +328,29 @@ class RocketMQAdminClientImplTest {
assertThat(group.getDelaySeconds()).isBetween(4, 30);
}
+ @ParameterizedTest
+ @CsvSource({"100,100,3600000,0", "100,99,3600000,3600", "99,100,3600000,3600", "100,100,0,0"})
+ void getConsumerGroupDoesNotTreatCaughtUpIdleQueuesAsDelayedTest(
+ long brokerOffset, long consumerOffset, long ageMillis, int minimumDelay) throws Exception {
+ ConsumeStats stats = new ConsumeStats();
+ OffsetWrapper wrapper = offsetWrapper(brokerOffset, consumerOffset);
+ wrapper.setLastTimestamp(ageMillis == 0 ? 0 : System.currentTimeMillis() - ageMillis);
+ stats.getOffsetTable().put(new MessageQueue("orders-topic", "broker-a", 0), wrapper);
+ stats.getOffsetTable().put(new MessageQueue("orders-topic", "broker-b", 1), offsetWrapper(50, 50));
+ when(adminExt.examineConsumeStats("orders")).thenReturn(stats);
+
+ ConsumerGroupVO group = adminClient.getConsumerGroup(null, "orders");
+
+ assertThat(group.isConsumeStatsAvailable()).isTrue();
+ assertThat(group.isConsumptionTimestampAvailable()).isEqualTo(ageMillis > 0);
+ assertThat(group.getTotalLag()).isEqualTo(brokerOffset - consumerOffset);
+ if (minimumDelay == 0) {
+ assertThat(group.getDelaySeconds()).isZero();
+ } else {
+ assertThat(group.getDelaySeconds()).isBetween(minimumDelay, minimumDelay + 30);
+ }
+ }
+
@Test
void getConsumerGroupReportsUnknownTotalLagWhenAnyQueueOffsetIsUnknownTest() throws Exception {
org.apache.rocketmq.remoting.protocol.body.ConsumerConnection connection =
diff --git a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
index 5bfbeabea..c191e0932 100644
--- a/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
+++ b/server/src/test/java/org/apache/rocketmq/studio/provider/apache/RocketMQMetadataProviderTest.java
@@ -21,6 +21,13 @@ import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.TableInfoHelper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import org.apache.rocketmq.studio.common.domain.PageResult;
+import org.apache.rocketmq.studio.provider.InstanceProviderRegistry;
+import org.apache.rocketmq.studio.provider.InstanceProvider;
+import org.apache.rocketmq.studio.cluster.metrics.collectors.ApacheRocketMqBusinessMetricsCollector;
+import org.apache.rocketmq.studio.instance.InstanceVO;
+import org.apache.rocketmq.studio.ops.alert.AlertRuleVO;
+import org.apache.rocketmq.studio.ops.alert.AlertRuleEvaluator;
+import org.apache.rocketmq.studio.cluster.metrics.MetricAvailability;
import org.apache.ibatis.builder.MapperBuilderAssistant;
import org.apache.rocketmq.studio.cluster.broker.MqAdminExtFactory;
import org.apache.rocketmq.common.message.MessageQueue;
@@ -62,6 +69,7 @@ import java.util.HashSet;
import java.util.List;
import java.util.LinkedHashMap;
import java.util.Map;
+import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.TimeUnit;
@@ -997,6 +1005,64 @@ class RocketMQMetadataProviderTest {
assertThat(groups.get(0).getDelaySeconds()).isBetween(4, 30);
}
+ @ParameterizedTest
+ @CsvSource({"100,100,3600000,0", "100,99,3600000,3600", "99,100,3600000,3600", "100,100,0,0"})
+ void listConsumerGroupsDoesNotTreatCaughtUpIdleQueuesAsDelayedTest(
+ long brokerOffset, long consumerOffset, long ageMillis, int minimumDelay) throws Exception {
+ ConsumerGroupVO group = groupWithIdleQueueStats(brokerOffset, consumerOffset, ageMillis);
+
+ assertThat(group.isConsumeStatsAvailable()).isTrue();
+ assertThat(group.isConsumptionTimestampAvailable()).isEqualTo(ageMillis > 0);
+ assertThat(group.getTotalLag()).isEqualTo(brokerOffset - consumerOffset);
+ if (minimumDelay == 0) {
+ assertThat(group.getDelaySeconds()).isZero();
+ } else {
+ assertThat(group.getDelaySeconds()).isBetween(minimumDelay, minimumDelay + 30);
+ }
+ }
+
+ @Test
+ void caughtUpIdleGroupDoesNotTriggerNativeDelayAlertTest() throws Exception {
+ ConsumerGroupVO group = groupWithIdleQueueStats(100, 100, 3_600_000L);
+ var registry = mock(InstanceProviderRegistry.class);
+ var instanceProvider = mock(InstanceProvider.class);
+ when(registry.byInstanceId("instance-a")).thenReturn(Optional.of(instanceProvider));
+ when(instanceProvider.listConsumerGroups("instance-a", null)).thenReturn(List.of(group));
+ var collector = new ApacheRocketMqBusinessMetricsCollector(registry);
+ var instance = InstanceVO.builder().name("instance-a").build();
+ var sample = collector.collect(instance).stream()
+ .filter(item -> "consumer.delay.seconds".equals(item.metricKey())).findFirst().orElseThrow();
+ var rule = AlertRuleVO.builder()
+ .metric("consumer.delay.seconds").operator(">").threshold(300).build();
+
+ assertThat(sample.availability()).isEqualTo(MetricAvailability.AVAILABLE);
+ assertThat(new AlertRuleEvaluator().evaluate(rule, sample).conditionMet()).isFalse();
+ assertThat(sample.value()).isZero();
+ }
+
+ private ConsumerGroupVO groupWithIdleQueueStats(long brokerOffset, long consumerOffset, long ageMillis)
+ throws Exception {
+ RmqGroup entity = new RmqGroup();
+ entity.setName("cg-idle");
+ entity.setInstanceId("instance-a");
+ when(groupMapper.selectList(any())).thenReturn(List.of(entity));
+ DefaultMQAdminExt admin = mock(DefaultMQAdminExt.class);
+ ConsumeStats stats = new ConsumeStats();
+ OffsetWrapper wrapper = offset(brokerOffset, consumerOffset);
+ wrapper.setLastTimestamp(ageMillis == 0 ? 0 : System.currentTimeMillis() - ageMillis);
+ stats.getOffsetTable().put(new MessageQueue("orders-topic", "broker-a", 0), wrapper);
+ stats.getOffsetTable().put(new MessageQueue("orders-topic", "broker-b", 1), offset(50, 50));
+ when(admin.examineConsumeStats("cg-idle")).thenReturn(stats);
+ when(runtimeAdminClientResolver.execute(eq("instance-a"), any()))
+ .thenAnswer(invocation -> invocation.<MqAdminExtFactory.AdminAction<Object>>getArgument(1).apply(admin));
+ RocketMQMetadataProvider provider = newLiveProvider(admin);
+ try {
+ return provider.listConsumerGroups("instance-a", null, null).getFirst();
+ } finally {
+ provider.shutdownOnlineEnrichmentExecutor();
+ }
+ }
+
@Test
void listConsumerGroupsShouldReportUnknownTotalLagWhenAnyQueueOffsetIsUnknownTest() throws Exception {
RmqGroup entity = new RmqGroup();