(rocketmq) branch develop updated: [ISSUE #11248] Fix Lite topic consumer losing messages when notified before they are readable (#11249)

[ji...@apache.org]

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 <zs...@foxmail.com>
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";