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 bef2f4810c [ISSUE #11248] Fix Lite topic consumer losing messages when
notified before they are readable (#11249)
bef2f4810c is described below
commit bef2f4810cb5f3b6b2cae90fb708e52b5529f2a0
Author: Quan <[email protected]>
AuthorDate: Thu Oct 8 09:53:49 2026 +0800
[ISSUE #11248] Fix Lite topic consumer losing messages when notified before
they are readable (#11249)
* [ISSUE #11248] Fix Lite topic consumer losing messages when notified
before they are readable
- Mark a lite topic in the per-request dedup set only after messages are
read
- Retry a re-dispatch within the same pop after an empty read instead of
dropping it
- Add unit tests for the empty-read path of popLiteTopic and popByClientId
* [ISSUE #11248] Re-trigger CI
---
.../broker/processor/PopLiteMessageProcessor.java | 11 ++++--
.../processor/PopLiteMessageProcessorTest.java | 43 ++++++++++++++++++++++
2 files changed, 50 insertions(+), 4 deletions(-)
diff --git
a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessor.java
b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessor.java
index 8fd3797437..acf158026b 100644
---
a/broker/src/main/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessor.java
+++
b/broker/src/main/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessor.java
@@ -258,8 +258,8 @@ public class PopLiteMessageProcessor implements
NettyRequestProcessor {
break;
}
if (processed.contains(lmqName)) {
- // Already handled in this pop; skip the duplicate. A
FIFO-blocked lmq is never marked by
- // popLiteTopic, so its later ack-unblock re-dispatch is
retried instead of being deduped away.
+ // Already read in this pop; skip the duplicate. A
FIFO-blocked or empty-read lmq is never marked
+ // by popLiteTopic, so its later re-dispatch (ack-unblock or
CQ commit) is retried, not deduped away.
continue;
}
// Tombstone check: reject pull if this client was evicted from
the liteTopic (exclusive mode)
@@ -298,10 +298,13 @@ public class PopLiteMessageProcessor implements
NettyRequestProcessor {
// Leave this lmq unmarked so a later ack-unblock re-dispatch
is retried, not deduped away.
return null;
}
- // Holding the lock and not blocked: this lmq is handled in this
pop, mark it to dedup further events.
- processed.add(lmqName);
final long consumeOffset = getPopOffset(group, lmqName);
GetMessageResult result = getMessage(clientHost, group, lmqName,
consumeOffset, (int) maxNum);
+ if (result != null && result.getMessageCount() > 0) {
+ // Mark only after messages are read: an empty read (e.g. CQ
entry not yet committed) must not
+ // dedup away a later re-dispatch of this lmq within the same
pop.
+ processed.add(lmqName);
+ }
return handleGetMessageResult(result, parentTopic, group, lmqName,
popTime, invisibleTime, attemptId);
} catch (Throwable e) {
LOGGER.error("popLiteTopic error. {}, {}", group, lmqName, e);
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessorTest.java
b/broker/src/test/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessorTest.java
index 7a3fca9931..5ae6b46d8c 100644
---
a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessorTest.java
+++
b/broker/src/test/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessorTest.java
@@ -305,6 +305,31 @@ public class PopLiteMessageProcessorTest {
verify(popLiteMessageProcessor).popLiteTopic(eq("parentTopic"),
eq("clientHost"), eq("group"), eq(event2), eq(31L), eq(pollTime), eq(6000L),
eq("attemptId"), anySet());
}
+ @SuppressWarnings("unchecked")
+ @Test
+ public void testPopByClientId_reDispatchAfterEmptyRead() {
+ String event = "lmqName";
+ long pollTime = System.currentTimeMillis();
+
+ // The lmq is dispatched before its CQ entry is readable, then
re-dispatched within the same pop.
+ Iterator<String> mockIterator = mock(Iterator.class);
+ when(mockIterator.hasNext()).thenReturn(true, true, false);
+ when(mockIterator.next()).thenReturn(event, event);
+
when(liteEventDispatcher.getEventIterator("clientId")).thenReturn(mockIterator);
+ when(lockService.tryLock(anyString())).thenReturn(true);
+ when(consumerOrderInfoManager.checkBlock(anyString(), anyString(),
anyString(), anyInt(), anyLong()))
+ .thenReturn(false);
+ GetMessageResult emptyResult =
mockGetMessageResult(GetMessageStatus.NO_MESSAGE_IN_QUEUE, 0, 0L);
+ GetMessageResult foundResult =
mockGetMessageResult(GetMessageStatus.FOUND, 1, 1L);
+ when(messageStore.getMessage("group", event, 0, 0, 32,
null)).thenReturn(emptyResult, foundResult);
+
+ Pair<StringBuilder, GetMessageResult> result =
popLiteMessageProcessor.popByClientId(
+ "clientHost", "parentTopic", "group", "clientId", pollTime, 6000L,
32, "attemptId");
+
+ assertEquals(1, result.getObject2().getMessageCount());
+ verify(messageStore, times(2)).getMessage("group", event, 0, 0, 32,
null);
+ }
+
@Test
public void testGetMessage_found() {
String group = "group";
@@ -414,6 +439,24 @@ public class PopLiteMessageProcessorTest {
verify(lockService).unlock(anyString());
}
+ @Test
+ public void testPopLiteTopic_emptyRead() {
+ when(lockService.tryLock(anyString())).thenReturn(true);
+ when(consumerOrderInfoManager.checkBlock(anyString(), anyString(),
anyString(), anyInt(), anyLong()))
+ .thenReturn(false);
+ GetMessageResult mockResult =
mockGetMessageResult(GetMessageStatus.NO_MESSAGE_IN_QUEUE, 0, 0L);
+ when(messageStore.getMessage("group", "lmqName", 0, 0, 32,
null)).thenReturn(mockResult);
+
+ Set<String> processed = new HashSet<>();
+ Pair<StringBuilder, GetMessageResult> result =
popLiteMessageProcessor.popLiteTopic("parentTopic",
+ "clientHost", "group", "lmqName", 32L, System.currentTimeMillis(),
6000L, "attemptId", processed);
+
+ assertEquals(0, result.getObject2().getMessageCount());
+ assertFalse(processed.contains("lmqName"));
+ verify(lockService).tryLock(anyString());
+ verify(lockService).unlock(anyString());
+ }
+
@Test
public void testPreCheck() {
final String parentTopic = "parentTopic";
