This is an automated email from the ASF dual-hosted git repository.
lizhimins 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 6291ce1ed3 [ISSUE #11191] Fix Lite topic consumer losing messages on
event re-dispatch during a pop (#11192)
6291ce1ed3 is described below
commit 6291ce1ed398f5eb2571006d8964ac7afa2adcd4
Author: Quan <[email protected]>
AuthorDate: Tue Sep 22 11:07:00 2026 +0800
[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/processor/PopLiteMessageProcessor.java | 15 +-
.../PopLiteMessageProcessorEventLossTest.java | 353 +++++++++++++++++++++
.../processor/PopLiteMessageProcessorTest.java | 48 ++-
3 files changed, 395 insertions(+), 21 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 167597dab3..8fd3797437 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
@@ -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);
diff --git
a/broker/src/test/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessorEventLossTest.java
b/broker/src/test/java/org/apache/rocketmq/broker/processor/PopLiteMessageProcessorEventLossTest.java
new file mode 100644
index 0000000000..6d3d28d1df
--- /dev/null
+++
b/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);
+ }
+}
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 ee6d6101b1..7a3fca9931 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
@@ -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());
}
}
