[ISSUE #11191] Fix Lite topic consumer losing messages on event re-dispatch during a pop (#11192)

- Mark the dedup entry only after a lite topic is handled under the pop lock, not when it is FIFO-blocked, so a re-dispatched event is retried instead of dropped
- Move the processed-set marking into popLiteTopic and narrow its visibility
- Add a regression test reproducing the concurrent ack-unblock event loss

broker/src/main/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessor.java

@@ -257,8 +257,10 @@ public class PopLiteMessageProcessor implements NettyRequestProcessor {
             if (null == lmqName) {
                 break;
             }
-            if (!processed.add(lmqName)) {
-                continue; // wait for next pop request or re-fetch in current process, here prefer the former approach
+            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.
+                continue;
             }
             // Tombstone check: reject pull if this client was evicted from the liteTopic (exclusive mode)
             if (isExclusiveGroup && brokerController.getLiteSubscriptionRegistry().hasExclusiveEvictionTombstone(clientId, lmqName)) {
@@ -266,7 +268,7 @@ public class PopLiteMessageProcessor implements NettyRequestProcessor {
                 continue;
             }
             Pair<StringBuilder, GetMessageResult> pair = popLiteTopic(parentTopic, clientHost, group, lmqName,
-                maxNum - total.get(), popTime, invisibleTime, attemptId);
+                maxNum - total.get(), popTime, invisibleTime, attemptId, processed);
             if (null == pair || pair.getObject2().getMessageCount() <= 0) {
                 continue;
             }
@@ -285,16 +287,19 @@ public class PopLiteMessageProcessor implements NettyRequestProcessor {
     }
 
     @VisibleForTesting
-    public Pair<StringBuilder, GetMessageResult> popLiteTopic(String parentTopic, String clientHost, String group,
-        String lmqName, long maxNum, long popTime, long invisibleTime, String attemptId) {
+    Pair<StringBuilder, GetMessageResult> popLiteTopic(String parentTopic, String clientHost, String group,
+        String lmqName, long maxNum, long popTime, long invisibleTime, String attemptId, Set<String> processed) {
         String lockKey = KeyBuilder.buildPopLiteLockKey(group, lmqName);
         if (!lockService.tryLock(lockKey)) {
             return null;
         }
         try {
             if (isFifoBlocked(attemptId, group, lmqName, invisibleTime)) {
+                // 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);
             return handleGetMessageResult(result, parentTopic, group, lmqName, popTime, invisibleTime, attemptId);

broker/src/test/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessorEventLossTest.java

@@ -0,0 +1,353 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.rocketmq.broker.processor;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.UUID;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import org.apache.rocketmq.broker.BrokerController;
+import org.apache.rocketmq.broker.lite.AbstractLiteLifecycleManager;
+import org.apache.rocketmq.broker.lite.LiteEventDispatcher;
+import org.apache.rocketmq.broker.lite.LiteSubscriptionRegistry;
+import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
+import org.apache.rocketmq.broker.pop.PopConsumerLockService;
+import org.apache.rocketmq.broker.pop.orderly.ConsumerOrderInfoManager;
+import org.apache.rocketmq.broker.subscription.SubscriptionGroupManager;
+import org.apache.rocketmq.common.BrokerConfig;
+import org.apache.rocketmq.common.KeyBuilder;
+import org.apache.rocketmq.common.Pair;
+import org.apache.rocketmq.common.TopicConfig;
+import org.apache.rocketmq.common.UtilAll;
+import org.apache.rocketmq.common.entity.ClientGroup;
+import org.apache.rocketmq.common.lite.LiteSubscription;
+import org.apache.rocketmq.common.lite.LiteUtil;
+import org.apache.rocketmq.remoting.protocol.subscription.SubscriptionGroupConfig;
+import org.apache.rocketmq.store.GetMessageResult;
+import org.apache.rocketmq.store.MessageStore;
+import org.junit.AfterClass;
+import org.junit.Assert;
+import org.junit.BeforeClass;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doNothing;
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.when;
+
+/**
+ * Reproduces the Lite pop event-loss bug: within one popByClientId, a lmq already visited in this
+ * request (added to the {@code processed} dedup set) gets re-dispatched by a concurrent ack that
+ * advances the offset and unblocks FIFO. The buggy code drops the re-enqueued event, so its
+ * remaining messages are never read (queue becomes empty, offset does not advance further).
+ *
+ * <p>Symptom asserted per iteration: consumer subscribing lt1 and lt2 with 5 messages each must
+ * eventually receive all 5+5. On the buggy code lt1 only yields its first message (offset 0) while
+ * lt2 yields all 5 (6 total), matching the failing E2E case.
+ *
+ * <p>The harness boots real components once ({@link MessageStore}, {@link LiteEventDispatcher},
+ * {@link PopLiteMessageProcessor} with its internal FIFO order-info + lock service,
+ * {@link ConsumerOffsetManager}) and loops the send/consume scenario internally. The ack of lt1's
+ * first message is fired while popByClientId is mid-iteration through a test-only white-box spy on
+ * {@code popLiteTopic}; this is purely to make the concurrency race deterministic rather than
+ * timing-dependent, and is not a dependency of the production code path.
+ */
+public class PopLiteMessageProcessorEventLossTest {
+
+    private static final String PARENT_TOPIC = "LitePopLossParent";
+    private static final String GROUP = "LitePopLossGroup";
+    private static final String CLIENT_ID = "clientId-repro";
+    private static final String CLIENT_HOST = "127.0.0.1:0";
+    private static final long INVISIBLE = 60_000L;
+    private static final int MAX_NUM = 32;
+    private static final int ITERATIONS = 5;
+    private static final int DRAIN_ROUNDS = 6;
+
+    private static final BrokerConfig BROKER_CONFIG = new BrokerConfig();
+    private static final ConcurrentMap<String, TopicConfig> TOPIC_CONFIG_TABLE = new ConcurrentHashMap<>();
+
+    private static String storePathRootDir;
+    private static MessageStore messageStore;
+    private static ConsumerOffsetManager consumerOffsetManager;
+    private static LiteEventDispatcher liteEventDispatcher;
+    private static PopLiteMessageProcessor popLiteMessageProcessor;
+    private static LiteSubscriptionRegistry liteSubscriptionRegistry;
+
+    // Per-pop capture populated by the popLiteTopic spy: lmqName -> offsets read in the current pop.
+    private static final Map<String, List<Long>> LAST_POP_READS = new HashMap<>();
+    // Cumulative reception across the whole scenario: lmqName -> offsets RECEIVED.
+    private static final Map<String, Set<Long>> RECEIVED = new HashMap<>();
+
+    // Injection state: when armed, firing the ack of (injectLmq, offset 0) while popLiteTopic runs.
+    private static final AtomicBoolean INJECT_ARMED = new AtomicBoolean(false);
+    private static volatile String injectLmq;
+    private static volatile long injectPopTime;
+
+    @BeforeClass
+    public static void setUp() throws Exception {
+        storePathRootDir = System.getProperty("java.io.tmpdir") + File.separator + "store-litePopLoss";
+        UtilAll.deleteFile(new File(storePathRootDir));
+
+        BROKER_CONFIG.setMaxClientEventCount(1000);
+
+        BrokerController brokerController = Mockito.mock(BrokerController.class);
+        SubscriptionGroupManager subscriptionGroupManager = Mockito.mock(SubscriptionGroupManager.class);
+        AbstractLiteLifecycleManager liteLifecycleManager = Mockito.mock(AbstractLiteLifecycleManager.class);
+        liteSubscriptionRegistry = Mockito.mock(LiteSubscriptionRegistry.class);
+        PopMessageProcessor popMessageProcessor = Mockito.mock(PopMessageProcessor.class);
+
+        SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig();
+        groupConfig.setGroupName(GROUP);
+        groupConfig.setLiteBindTopic(PARENT_TOPIC);
+        when(subscriptionGroupManager.findSubscriptionGroupConfig(GROUP)).thenReturn(groupConfig);
+
+        doReturn(BROKER_CONFIG).when(brokerController).getBrokerConfig();
+        doReturn(subscriptionGroupManager).when(brokerController).getSubscriptionGroupManager();
+        doReturn(liteSubscriptionRegistry).when(brokerController).getLiteSubscriptionRegistry();
+        doReturn(liteLifecycleManager).when(brokerController).getLiteLifecycleManager();
+        doReturn(popMessageProcessor).when(brokerController).getPopMessageProcessor();
+        when(liteLifecycleManager.getMaxOffsetInQueue(anyString())).thenReturn(1_000_000L);
+
+        consumerOffsetManager = new ConsumerOffsetManager(brokerController);
+        doReturn(consumerOffsetManager).when(brokerController).getConsumerOffsetManager();
+
+        messageStore = org.apache.rocketmq.broker.lite.LiteTestUtil.buildMessageStore(
+            storePathRootDir, BROKER_CONFIG, TOPIC_CONFIG_TABLE, false, null);
+        doReturn(messageStore).when(brokerController).getMessageStore();
+        messageStore.load();
+        messageStore.start();
+
+        liteEventDispatcher = new LiteEventDispatcher(brokerController, liteSubscriptionRegistry, liteLifecycleManager);
+        popLiteMessageProcessor = Mockito.spy(new PopLiteMessageProcessor(brokerController, liteEventDispatcher));
+        NotificationProcessor notificationProcessor = new NotificationProcessor(brokerController);
+
+        doReturn(liteEventDispatcher).when(brokerController).getLiteEventDispatcher();
+        doReturn(popLiteMessageProcessor).when(brokerController).getPopLiteMessageProcessor();
+        doReturn(notificationProcessor).when(brokerController).getNotificationProcessor();
+
+        // Subscribers fan-out: any lmq under GROUP resolves to our single client.
+        Map<String, List<ClientGroup>> subscriberMap = new HashMap<>();
+        subscriberMap.put(GROUP, Collections.singletonList(new ClientGroup(CLIENT_ID, GROUP)));
+        when(liteSubscriptionRegistry.getAllSubscribers(eq(GROUP), anyString())).thenReturn(subscriberMap);
+        when(liteSubscriptionRegistry.hasExclusiveEvictionTombstone(anyString(), anyString())).thenReturn(false);
+
+        // Avoid metrics manager dependency.
+        doNothing().when(popLiteMessageProcessor).recordPopLiteMetrics(any(), anyString(), anyString());
+
+        // Spy hook on popLiteTopic: capture per-lmq offsets, and inject the mid-pop ack when armed.
+        Mockito.doAnswer(invocation -> {
+            Object real = invocation.callRealMethod();
+            String lmq = invocation.getArgument(3);
+            @SuppressWarnings("unchecked")
+            Pair<StringBuilder, GetMessageResult> pair = (Pair<StringBuilder, GetMessageResult>) real;
+            if (pair != null && pair.getObject2() != null && pair.getObject2().getMessageCount() > 0) {
+                List<Long> offsets = new ArrayList<>(pair.getObject2().getMessageQueueOffset());
+                synchronized (LAST_POP_READS) {
+                    LAST_POP_READS.computeIfAbsent(lmq, k -> new ArrayList<>()).addAll(offsets);
+                }
+                synchronized (RECEIVED) {
+                    RECEIVED.computeIfAbsent(lmq, k -> new HashSet<>()).addAll(offsets);
+                }
+            }
+            if (INJECT_ARMED.get() && lmq.equals(injectLmq)) {
+                INJECT_ARMED.set(false);
+                // Fire the ack of the first in-flight message concurrently with this pop. This advances
+                // the offset, unblocks FIFO and re-dispatches the lmq back into this client's queue,
+                // reproducing a duplicate for an already-visited lmq within the same popByClientId.
+                ackLite(injectLmq, 0L, injectPopTime);
+            }
+            return real;
+        }).when(popLiteMessageProcessor).popLiteTopic(anyString(), anyString(), anyString(), anyString(),
+            anyLong(), anyLong(), anyLong(), anyString(), any());
+    }
+
+    @AfterClass
+    public static void tearDown() {
+        if (messageStore != null) {
+            messageStore.shutdown();
+            messageStore.destroy();
+        }
+        UtilAll.deleteFile(new File(storePathRootDir));
+    }
+
+    @Test
+    public void testLitePopEventLossAcrossIterations() throws Exception {
+        for (int i = 0; i < ITERATIONS; i++) {
+            runOnce(i);
+        }
+    }
+
+    private void runOnce(int iteration) throws Exception {
+        String lt1 = "lt1-" + iteration + "-" + UUID.randomUUID();
+        String lt2 = "lt2-" + iteration + "-" + UUID.randomUUID();
+        String lmq1 = LiteUtil.toLmqName(PARENT_TOPIC, lt1);
+        String lmq2 = LiteUtil.toLmqName(PARENT_TOPIC, lt2);
+
+        RECEIVED.clear();
+
+        // lt1 initially has only its first message readable; lt2 has all five.
+        send(lt1, 1);
+        send(lt2, 5);
+        awaitReput(lmq1, 1);
+        awaitReput(lmq2, 5);
+
+        // Initialize consumer offsets so getPopOffset does not fall back to init-offset lookup.
+        consumerOffsetManager.commitOffset("init", GROUP, lmq1, 0, 0L);
+        consumerOffsetManager.commitOffset("init", GROUP, lmq2, 0, 0L);
+
+        // The full subscription is used only by the fix's recovery path.
+        LiteSubscription subscription = new LiteSubscription().setGroup(GROUP).setTopic(PARENT_TOPIC);
+        Set<String> lmqSet = new HashSet<>();
+        lmqSet.add(lmq1);
+        lmqSet.add(lmq2);
+        subscription.setLmqSet(lmqSet);
+        when(liteSubscriptionRegistry.getLiteSubscription(CLIENT_ID)).thenReturn(subscription);
+
+        enqueue(lmq1);
+        enqueue(lmq2);
+
+        // POP #1: reads lt1 offset 0 (now in-flight, FIFO-blocked) and lt2 offsets 0..4.
+        long popTime1 = System.currentTimeMillis();
+        pop(popTime1);
+
+        // lt1's remaining four messages become readable now; a single coalesced arrival event enqueues lt1.
+        send(lt1, 4);
+        awaitReput(lmq1, 5);
+        enqueue(lmq1);
+
+        // POP #2: visits lt1 (blocked -> 0 read, added to processed); the spy fires lt1's offset-0 ack
+        // mid-pop, which re-dispatches lt1. The buggy code drops the re-enqueued duplicate.
+        injectLmq = lmq1;
+        injectPopTime = popTime1;
+        INJECT_ARMED.set(true);
+        long popTime2 = System.currentTimeMillis() + 1;
+        pop(popTime2);
+        INJECT_ARMED.set(false);
+
+        // Drain: keep popping and acking; on the fixed code lt1 is recovered and yields offsets 1..4.
+        for (int r = 0; r < DRAIN_ROUNDS; r++) {
+            Set<Long> lt1Received;
+            synchronized (RECEIVED) {
+                lt1Received = new HashSet<>(RECEIVED.getOrDefault(lmq1, Collections.emptySet()));
+            }
+            if (lt1Received.size() >= 5) {
+                break;
+            }
+            long popTime = System.currentTimeMillis() + 2 + r;
+            Map<String, List<Long>> reads = pop(popTime);
+            for (Map.Entry<String, List<Long>> e : reads.entrySet()) {
+                for (Long off : e.getValue()) {
+                    ackLite(e.getKey(), off, popTime);
+                }
+            }
+        }
+
+        assertAllReceived(iteration, lmq1, lmq2);
+    }
+
+    private Map<String, List<Long>> pop(long popTime) {
+        synchronized (LAST_POP_READS) {
+            LAST_POP_READS.clear();
+        }
+        popLiteMessageProcessor.popByClientId(CLIENT_HOST, PARENT_TOPIC, GROUP, CLIENT_ID,
+            popTime, INVISIBLE, MAX_NUM, UUID.randomUUID().toString());
+        synchronized (LAST_POP_READS) {
+            Map<String, List<Long>> copy = new HashMap<>();
+            for (Map.Entry<String, List<Long>> e : LAST_POP_READS.entrySet()) {
+                copy.put(e.getKey(), new ArrayList<>(e.getValue()));
+            }
+            return copy;
+        }
+    }
+
+    private void enqueue(String lmqName) {
+        liteEventDispatcher.tryDispatchToClient(lmqName, CLIENT_ID, GROUP, false);
+    }
+
+    private void send(String liteTopic, int count) {
+        for (int i = 0; i < count; i++) {
+            messageStore.putMessage(org.apache.rocketmq.broker.lite.LiteTestUtil.buildMessage(PARENT_TOPIC, liteTopic));
+        }
+    }
+
+    private void awaitReput(String lmqName, long expectedMaxOffset) throws Exception {
+        long deadline = System.currentTimeMillis() + 5_000L;
+        while (System.currentTimeMillis() < deadline) {
+            if (messageStore.getMaxOffsetInQueue(lmqName, 0) >= expectedMaxOffset) {
+                return;
+            }
+            Thread.sleep(10);
+        }
+        Assert.fail("reput timeout for " + lmqName + ", expected max offset " + expectedMaxOffset
+            + ", actual " + messageStore.getMaxOffsetInQueue(lmqName, 0));
+    }
+
+    /**
+     * Replicates {@link AckMessageProcessor#ackLite} for a single lite message: advance the FIFO
+     * order info, commit the consumer offset, and re-dispatch the lmq when it becomes unblocked.
+     */
+    private static void ackLite(String lmqName, long ackOffset, long popTime) {
+        ConsumerOrderInfoManager orderInfoManager = popLiteMessageProcessor.getConsumerOrderInfoManager();
+        PopConsumerLockService lockService = popLiteMessageProcessor.getLockService();
+        String lockKey = KeyBuilder.buildPopLiteLockKey(GROUP, lmqName);
+        while (!lockService.tryLock(lockKey)) {
+            // spin, matching AckMessageProcessor.ackLite
+        }
+        try {
+            long nextOffset = orderInfoManager.commitAndNext(lmqName, GROUP, 0, ackOffset, popTime);
+            if (nextOffset > -1L) {
+                if (!consumerOffsetManager.hasOffsetReset(lmqName, GROUP, 0)) {
+                    consumerOffsetManager.commitOffset("AckLiteHost", GROUP, lmqName, 0, nextOffset);
+                }
+                if (!orderInfoManager.checkBlock(null, lmqName, GROUP, 0, INVISIBLE)) {
+                    liteEventDispatcher.dispatch(GROUP, lmqName, 0, nextOffset, -1);
+                }
+            }
+        } finally {
+            lockService.unlock(lockKey);
+        }
+    }
+
+    private void assertAllReceived(int iteration, String lmq1, String lmq2) {
+        Set<Long> expected = new HashSet<>();
+        for (long o = 0; o < 5; o++) {
+            expected.add(o);
+        }
+        Set<Long> lt1;
+        Set<Long> lt2;
+        synchronized (RECEIVED) {
+            lt1 = new HashSet<>(RECEIVED.getOrDefault(lmq1, Collections.emptySet()));
+            lt2 = new HashSet<>(RECEIVED.getOrDefault(lmq2, Collections.emptySet()));
+        }
+        Assert.assertEquals("iteration " + iteration + ": lt2 must receive 5 messages", expected, lt2);
+        Assert.assertEquals("iteration " + iteration + ": lt1 must receive 5 messages but got " + lt1
+            + " (event loss reproduced: lt1 stuck after its first message)", expected, lt1);
+    }
+}

broker/src/test/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessorTest.java

@@ -53,6 +53,8 @@ import org.mockito.Mockito;
 import org.mockito.junit.MockitoJUnitRunner;
 
 import java.util.Iterator;
+import java.util.HashSet;
+import java.util.Set;
 
 import static org.assertj.core.api.Assertions.assertThat;
 import static org.junit.Assert.assertEquals;
@@ -63,7 +65,10 @@ import static org.junit.Assert.assertTrue;
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.ArgumentMatchers.anyInt;
 import static org.mockito.ArgumentMatchers.anyLong;
+import static org.mockito.ArgumentMatchers.anySet;
 import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doAnswer;
 import static org.mockito.Mockito.doNothing;
 import static org.mockito.Mockito.when;
 import static org.mockito.Mockito.verify;
@@ -230,14 +235,14 @@ public class PopLiteMessageProcessorTest {
         when(liteEventDispatcher.getEventIterator("clientId")).thenReturn(mockIterator);
         doReturn(new Pair<>(new StringBuilder("0"), mockResult))
             .when(popLiteMessageProcessor)
-            .popLiteTopic(anyString(), anyString(), anyString(), anyString(), anyLong(), anyLong(), anyLong(), anyString());
+            .popLiteTopic(anyString(), anyString(), anyString(), anyString(), anyLong(), anyLong(), anyLong(), anyString(), anySet());
 
         Pair<StringBuilder, GetMessageResult> result = popLiteMessageProcessor.popByClientId(
             "clientHost", "parentTopic", "group", "clientId", pollTime, 6000L, 32, "attemptId");
 
         assertEquals(msgCount, result.getObject2().getMessageCount());
         verify(mockIterator, times(2)).hasNext();
-        verify(popLiteMessageProcessor).popLiteTopic("parentTopic" ,"clientHost", "group", event, 32L, pollTime, 6000L, "attemptId");
+        verify(popLiteMessageProcessor).popLiteTopic(eq("parentTopic"), eq("clientHost"), eq("group"), eq(event), eq(32L), eq(pollTime), eq(6000L), eq("attemptId"), anySet());
     }
 
     @SuppressWarnings("unchecked")
@@ -255,7 +260,7 @@ public class PopLiteMessageProcessorTest {
         when(liteEventDispatcher.getEventIterator("clientId")).thenReturn(mockIterator);
         doReturn(new Pair<>(new StringBuilder("0"), mockResult))
             .when(popLiteMessageProcessor)
-            .popLiteTopic(anyString(), anyString(), anyString(), anyString(), anyLong(), anyLong(), anyLong(), anyString());
+            .popLiteTopic(anyString(), anyString(), anyString(), anyString(), anyLong(), anyLong(), anyLong(), anyString(), anySet());
 
         Pair<StringBuilder, GetMessageResult> result = popLiteMessageProcessor.popByClientId(
             "clientHost", "parentTopic", "group", "clientId", pollTime, 6000L, 2, "attemptId");
@@ -263,8 +268,8 @@ public class PopLiteMessageProcessorTest {
         assertEquals(2, result.getObject2().getMessageCount());
         assertEquals("0;0", result.getObject1().toString());
         verify(mockIterator, times(2)).hasNext();
-        verify(popLiteMessageProcessor).popLiteTopic("parentTopic", "clientHost", "group", event1, 2L, pollTime, 6000L, "attemptId");
-        verify(popLiteMessageProcessor).popLiteTopic("parentTopic", "clientHost", "group", event2, 1L, pollTime, 6000L, "attemptId");
+        verify(popLiteMessageProcessor).popLiteTopic(eq("parentTopic"), eq("clientHost"), eq("group"), eq(event1), eq(2L), eq(pollTime), eq(6000L), eq("attemptId"), anySet());
+        verify(popLiteMessageProcessor).popLiteTopic(eq("parentTopic"), eq("clientHost"), eq("group"), eq(event2), eq(1L), eq(pollTime), eq(6000L), eq("attemptId"), anySet());
     }
 
     @SuppressWarnings("unchecked")
@@ -281,9 +286,14 @@ public class PopLiteMessageProcessorTest {
         when(mockIterator.hasNext()).thenReturn(true, true, true, false);
         when(mockIterator.next()).thenReturn(event1, event2, event3);
         when(liteEventDispatcher.getEventIterator("clientId")).thenReturn(mockIterator);
-        doReturn(new Pair<>(new StringBuilder("0"), mockResult))
-            .when(popLiteMessageProcessor)
-            .popLiteTopic(anyString(), anyString(), anyString(), anyString(), anyLong(), anyLong(), anyLong(), anyString());
+        // popLiteTopic now owns the processed mark (added under the lock when not blocked); the stub
+        // mirrors that so the caller's contains-based dedup can drop the duplicate third event.
+        doAnswer(invocation -> {
+            Set<String> processed = invocation.getArgument(8);
+            processed.add(invocation.getArgument(3));
+            return new Pair<>(new StringBuilder("0"), mockResult);
+        }).when(popLiteMessageProcessor)
+            .popLiteTopic(anyString(), anyString(), anyString(), anyString(), anyLong(), anyLong(), anyLong(), anyString(), anySet());
 
         Pair<StringBuilder, GetMessageResult> result = popLiteMessageProcessor.popByClientId(
             "clientHost", "parentTopic", "group", "clientId", pollTime, 6000L, 32, "attemptId");
@@ -291,8 +301,8 @@ public class PopLiteMessageProcessorTest {
         assertEquals(2, result.getObject2().getMessageCount());
         assertEquals("0;0", result.getObject1().toString());
         verify(mockIterator, times(4)).hasNext();
-        verify(popLiteMessageProcessor).popLiteTopic("parentTopic", "clientHost", "group", event1, 32L, pollTime, 6000L, "attemptId");
-        verify(popLiteMessageProcessor).popLiteTopic("parentTopic", "clientHost", "group", event2, 31L, pollTime, 6000L, "attemptId");
+        verify(popLiteMessageProcessor).popLiteTopic(eq("parentTopic"), eq("clientHost"), eq("group"), eq(event1), eq(32L), eq(pollTime), eq(6000L), eq("attemptId"), anySet());
+        verify(popLiteMessageProcessor).popLiteTopic(eq("parentTopic"), eq("clientHost"), eq("group"), eq(event2), eq(31L), eq(pollTime), eq(6000L), eq("attemptId"), anySet());
     }
 
     @Test
@@ -360,10 +370,12 @@ public class PopLiteMessageProcessorTest {
     public void testPopLiteTopic_lockFailed() {
         when(lockService.tryLock(anyString())).thenReturn(false);
 
+        Set<String> processed = new HashSet<>();
         Pair<StringBuilder, GetMessageResult> result = popLiteMessageProcessor.popLiteTopic("parentTopic",
-            "clientHost", "group", "lmqName", 32L, System.currentTimeMillis(), 6000L, "attemptId");
+            "clientHost", "group", "lmqName", 32L, System.currentTimeMillis(), 6000L, "attemptId", processed);
 
         assertNull(result);
+        assertFalse(processed.contains("lmqName"));
         verify(lockService).tryLock(anyString());
         verify(lockService, never()).unlock(anyString());
     }
@@ -374,10 +386,12 @@ public class PopLiteMessageProcessorTest {
         when(consumerOrderInfoManager.checkBlock(anyString(), anyString(), anyString(), anyInt(), anyLong()))
             .thenReturn(true);
 
+        Set<String> processed = new HashSet<>();
         Pair<StringBuilder, GetMessageResult> result = popLiteMessageProcessor.popLiteTopic("parentTopic",
-            "clientHost", "group", "lmqName", 32L, System.currentTimeMillis(), 6000L, "attemptId");
+            "clientHost", "group", "lmqName", 32L, System.currentTimeMillis(), 6000L, "attemptId", processed);
 
         assertThat(result).isNull();
+        assertFalse(processed.contains("lmqName"));
         verify(lockService).tryLock(anyString());
         verify(lockService).unlock(anyString());
     }
@@ -390,10 +404,12 @@ public class PopLiteMessageProcessorTest {
         GetMessageResult mockResult = mockGetMessageResult(GetMessageStatus.FOUND, 1, 100L);
         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");
+            "clientHost", "group", "lmqName", 32L, System.currentTimeMillis(), 6000L, "attemptId", processed);
 
         assertEquals(mockResult, result.getObject2());
+        assertTrue(processed.contains("lmqName"));
         verify(lockService).tryLock(anyString());
         verify(lockService).unlock(anyString());
     }
@@ -512,7 +528,7 @@ public class PopLiteMessageProcessorTest {
         assertEquals(0, result.getObject2().getMessageCount());
         // popLiteTopic should never be called for the tombstoned lmqName
         verify(popLiteMessageProcessor, never())
-            .popLiteTopic(anyString(), anyString(), anyString(), anyString(), anyLong(), anyLong(), anyLong(), anyString());
+            .popLiteTopic(anyString(), anyString(), anyString(), anyString(), anyLong(), anyLong(), anyLong(), anyString(), anySet());
     }
 
     @SuppressWarnings("unchecked")
@@ -540,13 +556,13 @@ public class PopLiteMessageProcessorTest {
         when(liteEventDispatcher.getEventIterator(clientId)).thenReturn(mockIterator);
         doReturn(new Pair<>(new StringBuilder("0"), mockResult))
             .when(popLiteMessageProcessor)
-            .popLiteTopic(anyString(), anyString(), anyString(), anyString(), anyLong(), anyLong(), anyLong(), anyString());
+            .popLiteTopic(anyString(), anyString(), anyString(), anyString(), anyLong(), anyLong(), anyLong(), anyString(), anySet());
 
         Pair<StringBuilder, GetMessageResult> result = popLiteMessageProcessor.popByClientId(
             "clientHost", "parentTopic", group, clientId, pollTime, 6000L, 32, "attemptId");
 
         // Should return the message since no tombstone blocks
         assertEquals(msgCount, result.getObject2().getMessageCount());
-        verify(popLiteMessageProcessor).popLiteTopic("parentTopic", "clientHost", group, lmqName, 32L, pollTime, 6000L, "attemptId");
+        verify(popLiteMessageProcessor).popLiteTopic(eq("parentTopic"), eq("clientHost"), eq(group), eq(lmqName), eq(32L), eq(pollTime), eq(6000L), eq("attemptId"), anySet());
     }
 }