Skip to content

[ISSUE #10722] [Bug] Lite Topic may not be cleaned up when message store timestamp is unavailable #10723

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
lizhimins merged 2 commits into from
Aug 3, 2026
+68 −26
Merged

[ISSUE #10722] [Bug] Lite Topic may not be cleaned up when message store timestamp is unavailable #10723

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
c0db702
[ISSUE #10722] Fix Lite Topic expiration check when store timestamp i…
f1amingo Jul 31, 2026
0b6fddd
chore: retrigger CI
f1amingo Aug 3, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 43 additions & 16 deletions broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,6 +34,7 @@
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;

import static org.apache.rocketmq.broker.offset.ConsumerOffsetManager.TOPIC_GROUP_SEPARATOR;
Expand All @@ -44,15 +45,16 @@
*/
public abstract class AbstractLiteLifecycleManager extends ServiceThread {
private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.ROCKETMQ_POP_LITE_LOGGER_NAME);
private static final int MAX_INVALID_SCAN_COUNT = 5;
static final int MAX_INVALID_SCAN_COUNT = 5;

protected final BrokerController brokerController;
protected final String brokerName;
protected final LiteSharding liteSharding;
protected MessageStore messageStore;
protected Map<String, Integer> ttlMap = Collections.emptyMap();
protected Map<String, Set<String>> subscriberGroupMap = Collections.emptyMap();
protected Map<String, Integer> invalidScanCountMap = new ConcurrentHashMap<>();
protected Map<String, Integer> offsetInvalidScanCountMap = new ConcurrentHashMap<>();
protected Map<String, Integer> storeTimeInvalidScanCountMap = new ConcurrentHashMap<>();

public AbstractLiteLifecycleManager(BrokerController brokerController, LiteSharding liteSharding) {
this.brokerController = brokerController;
Expand Down Expand Up @@ -167,20 +169,19 @@ public boolean isLiteTopicExpired(String parentTopic, String lmqName, long maxOf
if (!LiteUtil.isLiteTopicQueue(lmqName)) {
return false;
}
if (maxOffset <= 0) {
int invalidCount = invalidScanCountMap.getOrDefault(lmqName, 0) + 1;
LOGGER.warn("unexpected condition, max offset <= 0, {}, {}, scanCount:{}", lmqName, maxOffset, invalidCount);
if (invalidCount > MAX_INVALID_SCAN_COUNT) { // check more times in case of concurrent issue
invalidScanCountMap.remove(lmqName);
return true;
}
invalidScanCountMap.put(lmqName, invalidCount);
return false;
} else {
invalidScanCountMap.remove(lmqName);
int offsetInvalidCount = trackInvalidCount(lmqName, maxOffset <= 0, offsetInvalidScanCountMap);
if (offsetInvalidCount > 0) {
// check more times in case of concurrent issue
LOGGER.warn("unexpected condition, max offset <= 0, {}, {}, scanCount:{}", lmqName, maxOffset, offsetInvalidCount);
return offsetInvalidCount > MAX_INVALID_SCAN_COUNT;
}
long latestStoreTime = messageStore.getMessageStoreTimeStamp(lmqName, 0, maxOffset - 1);
int storeTimeInvalidCount = trackInvalidCount(lmqName, latestStoreTime <= 0, storeTimeInvalidScanCountMap);
if (storeTimeInvalidCount > 0) {
// bypass TTL protection on purpose, but debounce against transient read failures
LOGGER.warn("latest store time <= 0, {}, {}, scanCount:{}", lmqName, latestStoreTime, storeTimeInvalidCount);
return storeTimeInvalidCount > MAX_INVALID_SCAN_COUNT;
}
long latestStoreTime =
this.brokerController.getMessageStore().getMessageStoreTimeStamp(lmqName, 0, maxOffset - 1);
long inactiveTime = System.currentTimeMillis() - latestStoreTime;
if (inactiveTime < brokerController.getBrokerConfig().getMinLiteTTl()) {
return false;
Expand All @@ -196,7 +197,32 @@ public boolean isLiteTopicExpired(String parentTopic, String lmqName, long maxOf
if (hasConsumerLag(lmqName, maxOffset, latestStoreTime, parentTopic)) {
return false;
}
return inactiveTime > minutes * 60 * 1000;
return inactiveTime > TimeUnit.MINUTES.toMillis(minutes);
}

/**
* Track the invalid state of the given lmq: increase the count when invalid, reset when recovered.
* The counter is removed automatically once it exceeds {@link #MAX_INVALID_SCAN_COUNT}.
*
* @return the current invalid count, 0 means healthy (and the counter has been reset)
*/
private int trackInvalidCount(String lmqName, boolean invalid, Map<String, Integer> invalidCountMap) {
if (!invalid) {
invalidCountMap.remove(lmqName);
return 0;
}
int invalidCount = invalidCountMap.getOrDefault(lmqName, 0) + 1;
if (invalidCount > MAX_INVALID_SCAN_COUNT) {
invalidCountMap.remove(lmqName);
} else {
invalidCountMap.put(lmqName, invalidCount);
}
return invalidCount;
}

private void removeInvalidCount(String lmqName) {
offsetInvalidScanCountMap.remove(lmqName);
storeTimeInvalidScanCountMap.remove(lmqName);
}

public void deleteLmq(String parentTopic, String lmqName) {
Expand All @@ -214,6 +240,7 @@ public void deleteLmq(String parentTopic, String lmqName) {
brokerController.getLiteSubscriptionRegistry().cleanSubscription(lmqName, false);
brokerController.getConsumerOffsetManager().getPullOffsetTable().remove(
lmqName + TOPIC_GROUP_SEPARATOR + MixAll.TOOLS_CONSUMER_GROUP);
removeInvalidCount(lmqName);
LOGGER.info("delete lmq finish. {}, sharding:{}", lmqName, sharding);
} catch (Exception e) {
LOGGER.error("delete lmq error. {}", lmqName, e);
Expand Down
13 changes: 12 additions & 1 deletion broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,6 +47,7 @@
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;

import static org.apache.rocketmq.broker.lite.AbstractLiteLifecycleManager.MAX_INVALID_SCAN_COUNT;
import static org.apache.rocketmq.broker.offset.ConsumerOffsetManager.TOPIC_GROUP_SEPARATOR;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
Expand Down Expand Up @@ -165,7 +166,17 @@ public void testIsLiteTopicExpired() {
Assert.assertFalse(lifecycleManager.isLiteTopicExpired(PARENT_TOPIC, "whatever", 10L));

// maxOffset invalid
Assert.assertFalse(lifecycleManager.isLiteTopicExpired(PARENT_TOPIC, EXIST_LMQ_NAME, 0L));
for (int i = 0; i < MAX_INVALID_SCAN_COUNT; i++) {
Assert.assertFalse(lifecycleManager.isLiteTopicExpired(PARENT_TOPIC, EXIST_LMQ_NAME, 0L));
}
Assert.assertTrue(lifecycleManager.isLiteTopicExpired(PARENT_TOPIC, EXIST_LMQ_NAME, 0L));

// storeTime invalid
when(messageStore.getMessageStoreTimeStamp(anyString(), anyInt(), anyLong())).thenReturn(-1L);
for (int i = 0; i < MAX_INVALID_SCAN_COUNT; i++) {
Assert.assertFalse(lifecycleManager.isLiteTopicExpired(PARENT_TOPIC, EXIST_LMQ_NAME, 100L));
}
Assert.assertTrue(lifecycleManager.isLiteTopicExpired(PARENT_TOPIC, EXIST_LMQ_NAME, 100L));

// less than minLiteTTl
long mockStoreTime = System.currentTimeMillis();
Expand Down
1 change: 1 addition & 0 deletions test/src/test/java/org/apache/rocketmq/test/offset/LagCalculationIT.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -200,6 +200,7 @@ public void testEstimateLag() throws Exception {
});
producer.send(msgMap);
}
waitForFullyDispatched();

// test lag estimation for tag consumer
for (BrokerController controller : brokerControllerList) {
Expand Down
21 changes: 12 additions & 9 deletions ...re/src/test/java/org/apache/rocketmq/tieredstore/core/MessageStoreDispatcherImplTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -146,15 +146,18 @@ public void dispatchFromCommitLogTest() throws Exception {
new SelectMappedBufferResult(0L, buffer.asReadOnlyBuffer(), buffer.remaining(), null));
dispatcher.doScheduleDispatch(flatFile, true).join();

Awaitility.await().pollInterval(Duration.ofSeconds(1)).atMost(Duration.ofSeconds(30)).until(() -> {
List<IndexItem> resultList1 = indexService.queryAsync(
mq.getTopic(), "uk", 32, 0L, System.currentTimeMillis()).join();
List<IndexItem> resultList2 = indexService.queryAsync(
mq.getTopic(), "uk", 120, 0L, System.currentTimeMillis()).join();
Assert.assertEquals(32, resultList1.size());
Assert.assertEquals(100, resultList2.size());
return true;
});
// Index construction is submitted to the buffer commit executor asynchronously,
// so it may not be visible right after doScheduleDispatch returns. Use untilAsserted
// here, only it retries when the assertion inside fails.
Awaitility.await().pollDelay(Duration.ZERO).pollInterval(Duration.ofMillis(100))
.atMost(Duration.ofSeconds(30)).untilAsserted(() -> {
List<IndexItem> resultList1 = indexService.queryAsync(
mq.getTopic(), "uk", 32, 0L, System.currentTimeMillis()).join();
List<IndexItem> resultList2 = indexService.queryAsync(
mq.getTopic(), "uk", 120, 0L, System.currentTimeMillis()).join();
Assert.assertEquals(32, resultList1.size());
Assert.assertEquals(100, resultList2.size());
});

Assert.assertEquals(100L, flatFile.getConsumeQueueMinOffset());
Assert.assertEquals(200L, flatFile.getConsumeQueueMaxOffset());
Expand Down
Loading