You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@rocketmq.apache.org by ji...@apache.org on 2026/10/13 01:22:36 UTC
(rocketmq) branch develop updated: [ISSUE #11223] Fix stale lag time info remaining in getLiteGroupInfo after lite topic expired (#11224)
This is an automated email from the ASF dual-hosted git repository.
RongtongJin pushed a commit to branch develop
in repository https://gitbox.apache.org/repos/asf/rocketmq.git
The following commit(s) were added to refs/heads/develop by this push:
new a7edeecfb6 [ISSUE #11223] Fix stale lag time info remaining in getLiteGroupInfo after lite topic expired (#11224)
a7edeecfb6 is described below
commit a7edeecfb662f54427dea79b1a37fb034f957195
Author: majialong <ma...@gmail.com>
AuthorDate: Tue Oct 13 09:22:28 2026 +0800
[ISSUE #11223] Fix stale lag time info remaining in getLiteGroupInfo after lite topic expired (#11224)
---
.../broker/lite/AbstractLiteLifecycleManager.java | 2 ++
.../broker/metrics/LiteConsumerLagCalculator.java | 11 +++++++
.../lite/AbstractLiteLifecycleManagerTest.java | 10 +++++++
.../metrics/LiteConsumerLagCalculatorTest.java | 35 ++++++++++++++++++++++
4 files changed, 58 insertions(+)
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java
index f2b89d8043..92c614bd4d 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java
@@ -331,6 +331,8 @@ public abstract class AbstractLiteLifecycleManager extends ServiceThread {
brokerController.getLiteSubscriptionRegistry().cleanSubscription(lmqName, false);
brokerController.getConsumerOffsetManager().getPullOffsetTable().remove(
lmqName + TOPIC_GROUP_SEPARATOR + MixAll.TOOLS_CONSUMER_GROUP);
+ brokerController.getBrokerMetricsManager().getLiteConsumerLagCalculator()
+ .removeLagInfoByLmq(parentTopic, lmqName);
removeInvalidCount(lmqName);
onLmqDelete(lmqName);
LOGGER.info("delete lmq finish. {}, sharding:{}", lmqName, sharding);
diff --git a/broker/src/main/java/org/apache/rocketmq/broker/metrics/LiteConsumerLagCalculator.java b/broker/src/main/java/org/apache/rocketmq/broker/metrics/LiteConsumerLagCalculator.java
index 7f0dcbed5c..8e4c771725 100644
--- a/broker/src/main/java/org/apache/rocketmq/broker/metrics/LiteConsumerLagCalculator.java
+++ b/broker/src/main/java/org/apache/rocketmq/broker/metrics/LiteConsumerLagCalculator.java
@@ -61,6 +61,17 @@ public class LiteConsumerLagCalculator {
}
}
+ /**
+ * Remove lag info of the given lmq for all groups bound to the parent topic.
+ */
+ public void removeLagInfoByLmq(String bindTopic, String lmqName) {
+ topicGroupLagTimeMap.forEach((topicGroup, lagHeap) -> {
+ if (topicGroup.topic.equals(bindTopic)) {
+ lagHeap.removeIf(info -> info.getLmqName().equals(lmqName));
+ }
+ });
+ }
+
public void updateLagInfo(String group, String bindTopic, String lmqName, long storeTimestamp) {
PriorityBlockingQueue<LagTimeInfo> lagHeap = topicGroupLagTimeMap.computeIfAbsent(
new TopicGroup(bindTopic, group),
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java b/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java
index 4a15311a79..9293bd3e5d 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java
@@ -28,6 +28,8 @@ import java.util.function.Function;
import org.apache.commons.lang3.tuple.Triple;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.config.v1.RocksDBConsumerOffsetManager;
+import org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
+import org.apache.rocketmq.broker.metrics.LiteConsumerLagCalculator;
import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager;
import org.apache.rocketmq.broker.processor.PopLiteMessageProcessor;
import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
@@ -82,6 +84,10 @@ public class AbstractLiteLifecycleManagerTest {
private ConsumerOrderInfoManager consumerOrderInfoManager;
@Mock
private LiteSubscriptionRegistry liteSubscriptionRegistry;
+ @Mock
+ private BrokerMetricsManager brokerMetricsManager;
+ @Mock
+ private LiteConsumerLagCalculator liteConsumerLagCalculator;
private TestLiteLifecycleManager lifecycleManager;
private BrokerConfig brokerConfig;
@@ -102,6 +108,8 @@ public class AbstractLiteLifecycleManagerTest {
when(brokerController.getPopLiteMessageProcessor()).thenReturn(popLiteMessageProcessor);
when(popLiteMessageProcessor.getConsumerOrderInfoManager()).thenReturn(consumerOrderInfoManager);
when(brokerController.getLiteSubscriptionRegistry()).thenReturn(liteSubscriptionRegistry);
+ when(brokerController.getBrokerMetricsManager()).thenReturn(brokerMetricsManager);
+ when(brokerMetricsManager.getLiteConsumerLagCalculator()).thenReturn(liteConsumerLagCalculator);
topicConfig.getAttributes().put(
TopicAttributes.TOPIC_MESSAGE_TYPE_ATTRIBUTE.getName(), TopicMessageType.LITE.getValue());
@@ -215,6 +223,7 @@ public class AbstractLiteLifecycleManagerTest {
verify(messageStore).deleteTopics(Collections.singleton(EXIST_LMQ_NAME));
verify(liteSubscriptionRegistry).cleanSubscription(EXIST_LMQ_NAME, false);
verify(consumerOrderInfoManager, times(1)).remove(EXIST_LMQ_NAME, GROUP);
+ verify(liteConsumerLagCalculator, times(1)).removeLagInfoByLmq(PARENT_TOPIC, EXIST_LMQ_NAME);
// not sharding to this broker
when(liteSharding.shardingByLmqName(PARENT_TOPIC, EXIST_LMQ_NAME)).thenReturn("otherBrokerName");
@@ -239,6 +248,7 @@ public class AbstractLiteLifecycleManagerTest {
verify(consumerOffsetManager).removeConsumerOffset(removeKey);
verify(messageStore).deleteTopics(Collections.singleton(EXIST_LMQ_NAME));
verify(liteSubscriptionRegistry).cleanSubscription(EXIST_LMQ_NAME, false);
+ verify(liteConsumerLagCalculator).removeLagInfoByLmq(PARENT_TOPIC, EXIST_LMQ_NAME);
}
@Test
diff --git a/broker/src/test/java/org/apache/rocketmq/broker/metrics/LiteConsumerLagCalculatorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/metrics/LiteConsumerLagCalculatorTest.java
index d2d6b7244a..3068bb8bcd 100644
--- a/broker/src/test/java/org/apache/rocketmq/broker/metrics/LiteConsumerLagCalculatorTest.java
+++ b/broker/src/test/java/org/apache/rocketmq/broker/metrics/LiteConsumerLagCalculatorTest.java
@@ -146,6 +146,41 @@ public class LiteConsumerLagCalculatorTest {
assertTrue(lagHeap.isEmpty());
}
+ @Test
+ public void testRemoveLagInfoByLmq() {
+ String group1 = "testGroup1";
+ String group2 = "testGroup2";
+ String topic = "testTopic";
+ String otherTopic = "otherTopic";
+ String lmqName = LiteUtil.toLmqName(topic, "lmq1");
+ String keepLmqName = LiteUtil.toLmqName(topic, "lmq2");
+ long storeTimestamp = System.currentTimeMillis();
+
+ // Record the same lmq under two groups, plus another lmq and another topic as controls
+ liteConsumerLagCalculator.updateLagInfo(group1, topic, lmqName, storeTimestamp);
+ liteConsumerLagCalculator.updateLagInfo(group1, topic, keepLmqName, storeTimestamp);
+ liteConsumerLagCalculator.updateLagInfo(group2, topic, lmqName, storeTimestamp);
+ liteConsumerLagCalculator.updateLagInfo(group1, otherTopic, lmqName, storeTimestamp);
+
+ // Remove the lmq for all groups bound to the topic
+ liteConsumerLagCalculator.removeLagInfoByLmq(topic, lmqName);
+
+ // Only the target lmq is removed, other lmq in the same group is kept
+ PriorityBlockingQueue<LiteConsumerLagCalculator.LagTimeInfo> group1Heap =
+ liteConsumerLagCalculator.topicGroupLagTimeMap.get(new TopicGroup(topic, group1));
+ assertEquals(1, group1Heap.size());
+ assertEquals(keepLmqName, group1Heap.peek().getLmqName());
+ // The lmq is also removed from other groups of the same topic
+ assertTrue(liteConsumerLagCalculator.topicGroupLagTimeMap.get(new TopicGroup(topic, group2)).isEmpty());
+ // Lag info under other topics is not affected
+ assertEquals(1, liteConsumerLagCalculator.topicGroupLagTimeMap.get(new TopicGroup(otherTopic, group1)).size());
+
+ // TopK by lag time no longer returns the removed lmq
+ Pair<List<LiteLagInfo>, Long> result = liteConsumerLagCalculator.getLagTimestampTopK(group2, topic, 10);
+ assertTrue(result.getObject1().isEmpty());
+ assertEquals(LiteConsumerLagCalculator.INIT_CONSUME_TIMESTAMP, result.getObject2().longValue());
+ }
+
@Test
public void testOffsetTableForEachByGroup() {
String testTopic = "testTopic";