From 33c0242e7fc7da9917c33ad51b4fa27a0c7732d9 Mon Sep 17 00:00:00 2001 From: Quan Date: Thu, 28 May 2026 13:46:35 +0800 Subject: [PATCH 1/2] Enforce Lite exclusive subscription eviction via server-side tombstones Lite exclusive eviction relied on a fire-and-forget unsubscribe notification. When the client missed it, the old consumer kept pulling, causing double consumption. Add a broker-local, in-memory tombstone keyed by (clientId, lmqName): - Written synchronously on eviction, before client notification. - Checked at the Lite pull entry; rejected pulls reuse "no message available", requiring no client change. - Lifecycle-aligned with the subscription set: cleared on resubscription, full-sync reconciliation, and disconnect/timeout. - Self-heals lost notifications by re-sending unsubscribe when a heartbeat still reports a tombstoned lmqName. Tests: add ExclusiveEvictionTombstonesTest; extend registry and pull-path tests to cover eviction, self-clear, reconciliation, and cleanup. --- .../lite/ExclusiveEvictionTombstones.java | 93 ++++++++ .../broker/lite/LiteSubscriptionRegistry.java | 2 + .../lite/LiteSubscriptionRegistryImpl.java | 36 +++ .../processor/PopLiteMessageProcessor.java | 7 + .../lite/ExclusiveEvictionTombstonesTest.java | 159 +++++++++++++ .../LiteSubscriptionRegistryImplTest.java | 225 ++++++++++++++++++ .../PopLiteMessageProcessorTest.java | 70 ++++++ 7 files changed, 592 insertions(+) create mode 100644 broker/src/main/java/org/apache/rocketmq/broker/lite/ExclusiveEvictionTombstones.java create mode 100644 broker/src/test/java/org/apache/rocketmq/broker/lite/ExclusiveEvictionTombstonesTest.java diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/ExclusiveEvictionTombstones.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/ExclusiveEvictionTombstones.java new file mode 100644 index 00000000000..0efaf03c67f --- /dev/null +++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/ExclusiveEvictionTombstones.java @@ -0,0 +1,93 @@ +/* + * 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.lite; + +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import org.apache.rocketmq.common.lite.LiteUtil; + +/** + * Manages tombstones for exclusive subscription eviction. + *

+ * When a client is evicted from a liteTopic (exclusive mode), an entry is placed here + * to prevent the evicted client from pulling messages until the tombstone is cleared + * during the next full subscription sync or client removal. + *

+ * Key format: clientId$lmqName + */ +public class ExclusiveEvictionTombstones { + + private static final char KEY_SEPARATOR = LiteUtil.SEPARATOR; + + private final Set tombstones = ConcurrentHashMap.newKeySet(); + + /** + * Check whether a tombstone exists for the given client and lmqName. + */ + public boolean contains(String clientId, String lmqName) { + return tombstones.contains(buildKey(clientId, lmqName)); + } + + /** + * Add a tombstone for the given client and lmqName. + */ + public void add(String clientId, String lmqName) { + tombstones.add(buildKey(clientId, lmqName)); + } + + /** + * Remove the tombstone for the given client and lmqName, if present. + */ + public void remove(String clientId, String lmqName) { + tombstones.remove(buildKey(clientId, lmqName)); + } + + /** + * Remove all tombstones belonging to the specified client. + */ + public void removeAllOf(String clientId) { + String prefix = clientId + KEY_SEPARATOR; + tombstones.removeIf(key -> key.startsWith(prefix)); + } + + /** + * For a given client, remove tombstones whose lmqName is NOT in the provided active set. + * This is used during full subscription sync to clear stale tombstones. + */ + public void removeStale(String clientId, Set activeLmqNames) { + String prefix = clientId + KEY_SEPARATOR; + tombstones.removeIf(key -> { + if (!key.startsWith(prefix)) { + return false; + } + String lmqName = key.substring(prefix.length()); + return !activeLmqNames.contains(lmqName); + }); + } + + /** + * Return the current number of tombstones (for monitoring/testing). + */ + public int size() { + return tombstones.size(); + } + + private static String buildKey(String clientId, String lmqName) { + return clientId + KEY_SEPARATOR + lmqName; + } +} diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistry.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistry.java index 50fbc373f23..965ed180fc6 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistry.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistry.java @@ -50,6 +50,8 @@ public interface LiteSubscriptionRegistry { void cleanSubscription(String lmqName, boolean notifyClient); + boolean hasExclusiveEvictionTombstone(String clientId, String lmqName); + void start(); void shutdown(); diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java index 3ced9331271..b487b8757f4 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java @@ -61,6 +61,8 @@ public class LiteSubscriptionRegistryImpl extends ServiceThread implements LiteS private final BrokerController brokerController; private final AbstractLiteLifecycleManager liteLifecycleManager; + private final ExclusiveEvictionTombstones exclusiveEvictionTombstones = new ExclusiveEvictionTombstones(); + public LiteSubscriptionRegistryImpl(BrokerController brokerController, AbstractLiteLifecycleManager liteLifecycleManager) { this.brokerController = brokerController; @@ -99,6 +101,10 @@ public void addPartialSubscription(String clientId, String group, String topic, // First remove the old subscription if (LiteMetadataUtil.isSubLiteExclusive(group, brokerController)) { excludeClientByLmqName(clientId, group, lmqName); + // Boundary case: this client may have a stale tombstone from a previous eviction. + // Since it is now actively re-claiming the lmqName, clear its own tombstone so + // subsequent popLiteTopic is not blocked by the stale mark. + exclusiveEvictionTombstones.remove(clientId, lmqName); } resetOffset(lmqName, group, clientId, offsetOption); addTopicGroup(clientGroup, lmqName); @@ -144,12 +150,31 @@ public void addCompleteSubscription(String clientId, String group, String topic, thisSub.addLiteTopic(lmqName); addTopicGroup(clientGroup, lmqName); }); + // Tombstone operations only apply to exclusive groups. + if (LiteMetadataUtil.isSubLiteExclusive(group, brokerController)) { + // Boundary case: if any lmqName in the client's reported full subscription still has + // a tombstone, the previous notifyUnsubscribeLite was likely lost. Re-send the + // unsubscribe notification to drive the client's local state to converge. + lmqNameNew.stream() + .filter(lmqName -> exclusiveEvictionTombstones.contains(clientId, lmqName)) + .forEach(lmqName -> { + LOGGER.info("re-notify unsubscribe for tombstoned lmqName, clientId:{}, group:{}, lmqName:{}", + clientId, group, lmqName); + notifyUnsubscribeLite(clientId, group, lmqName); + }); + // Clean exclusive-eviction tombstones for liteTopics no longer in the client's full subscription set + exclusiveEvictionTombstones.removeStale(clientId, lmqNameNew); + } } @Override public void removeCompleteSubscription(String clientId) { clientChannels.remove(clientId); LiteSubscription thisSub = client2Subscription.remove(clientId); + // Only clean tombstones for exclusive groups. + if (thisSub == null || LiteMetadataUtil.isSubLiteExclusive(thisSub.getGroup(), brokerController)) { + exclusiveEvictionTombstones.removeAllOf(clientId); + } if (thisSub == null) { return; } @@ -299,6 +324,7 @@ protected void excludeClientByLmqName(String newClientId, String group, String l client2Subscription.remove(clientGroup.clientId); } } + exclusiveEvictionTombstones.add(clientGroup.clientId, lmqName); notifyUnsubscribeLite(clientGroup.clientId, clientGroup.group, lmqName); boolean resetOffset = LiteMetadataUtil.isResetOffsetInExclusiveMode(group, brokerController); LOGGER.info("excludeClientByLmqName group:{}, lmqName:{}, resetOffset:{}, clientId:{} -> {}", @@ -464,6 +490,16 @@ protected void cleanupExpiredSubscriptions(long checkTimeout) { LOGGER.info("Remove expired LiteSubscription, topic: {}, group: {}, clientId: {}, timeout: {}ms, expired: {}ms", topic, group, clientId, checkTimeout, System.currentTimeMillis() - liteSubscription.getUpdateTime()); }); + + int tombstoneSize = exclusiveEvictionTombstones.size(); + if (tombstoneSize > 0) { + LOGGER.info("ExclusiveEvictionTombstones size: {}", tombstoneSize); + } + } + + @Override + public boolean hasExclusiveEvictionTombstone(String clientId, String lmqName) { + return exclusiveEvictionTombstones.contains(clientId, lmqName); } } \ No newline at end of file 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 9314dab734e..7baca876947 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 @@ -22,6 +22,7 @@ import io.opentelemetry.api.common.Attributes; import org.apache.rocketmq.broker.BrokerController; import org.apache.rocketmq.broker.lite.LiteEventDispatcher; +import org.apache.rocketmq.broker.lite.LiteMetadataUtil; import org.apache.rocketmq.broker.longpolling.PollingResult; import org.apache.rocketmq.broker.longpolling.PopLiteLongPollingService; import org.apache.rocketmq.broker.metrics.LiteConsumerLagCalculator; @@ -253,6 +254,7 @@ public Pair popByClientId(String clientHost, St StringBuilder orderCountInfoAll = new StringBuilder(); AtomicLong total = new AtomicLong(0); + boolean isExclusiveGroup = LiteMetadataUtil.isSubLiteExclusive(group, brokerController); Set processed = new HashSet<>(); // deduplication in one request Iterator iterator = liteEventDispatcher.getEventIterator(clientId); while (total.get() < maxNum && iterator.hasNext()) { @@ -263,6 +265,11 @@ public Pair popByClientId(String clientHost, St if (!processed.add(lmqName)) { continue; // wait for next pop request or re-fetch in current process, here prefer the former approach } + // Tombstone check: reject pull if this client was evicted from the liteTopic (exclusive mode) + if (isExclusiveGroup && brokerController.getLiteSubscriptionRegistry().hasExclusiveEvictionTombstone(clientId, lmqName)) { + LOGGER.info("popLiteTopic rejected by tombstone: clientId={}, group={}, lmqName={}", clientId, group, lmqName); + continue; + } Pair pair = popLiteTopic(parentTopic, clientHost, group, lmqName, maxNum - total.get(), popTime, invisibleTime, attemptId); if (null == pair || pair.getObject2().getMessageCount() <= 0) { diff --git a/broker/src/test/java/org/apache/rocketmq/broker/lite/ExclusiveEvictionTombstonesTest.java b/broker/src/test/java/org/apache/rocketmq/broker/lite/ExclusiveEvictionTombstonesTest.java new file mode 100644 index 00000000000..3df0f07d246 --- /dev/null +++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/ExclusiveEvictionTombstonesTest.java @@ -0,0 +1,159 @@ +/* + * 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.lite; + +import java.util.HashSet; +import java.util.Set; +import org.junit.Before; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class ExclusiveEvictionTombstonesTest { + + private ExclusiveEvictionTombstones tombstones; + + @Before + public void setUp() { + tombstones = new ExclusiveEvictionTombstones(); + } + + @Test + public void testAddAndContains() { + // empty store + assertFalse(tombstones.contains("client1", "lmq1")); + assertEquals(0, tombstones.size()); + + // basic add + tombstones.add("client1", "lmq1"); + assertTrue(tombstones.contains("client1", "lmq1")); + assertFalse(tombstones.contains("client1", "lmq2")); // same client, different lmq + assertFalse(tombstones.contains("client2", "lmq1")); // different client, same lmq + assertEquals(1, tombstones.size()); + + // duplicate add is idempotent + tombstones.add("client1", "lmq1"); + assertEquals(1, tombstones.size()); + } + + @Test + public void testRemoveAllOf() { + tombstones.add("client1", "lmq1"); + tombstones.add("client1", "lmq2"); + tombstones.add("client2", "lmq1"); + + // non-existent client is no-op + tombstones.removeAllOf("client3"); + assertEquals(3, tombstones.size()); + + // removes only target client + tombstones.removeAllOf("client1"); + assertFalse(tombstones.contains("client1", "lmq1")); + assertFalse(tombstones.contains("client1", "lmq2")); + assertTrue(tombstones.contains("client2", "lmq1")); + assertEquals(1, tombstones.size()); + } + + @Test + public void testRemoveStale() { + tombstones.add("client1", "lmq1"); + tombstones.add("client1", "lmq2"); + tombstones.add("client1", "lmq3"); + tombstones.add("client2", "lmq1"); + + // removes stale, keeps active, does not affect other clients + Set activeSet = new HashSet<>(); + activeSet.add("lmq2"); + activeSet.add("lmq3"); + tombstones.removeStale("client1", activeSet); + + assertFalse(tombstones.contains("client1", "lmq1")); // removed (not in activeSet) + assertTrue(tombstones.contains("client1", "lmq2")); // retained + assertTrue(tombstones.contains("client1", "lmq3")); // retained + assertTrue(tombstones.contains("client2", "lmq1")); // unaffected + assertEquals(3, tombstones.size()); + + // empty activeSet clears all for that client + tombstones.removeStale("client1", new HashSet<>()); + assertFalse(tombstones.contains("client1", "lmq2")); + assertFalse(tombstones.contains("client1", "lmq3")); + assertTrue(tombstones.contains("client2", "lmq1")); // still unaffected + } + + @Test + public void testSize() { + assertEquals(0, tombstones.size()); + + tombstones.add("c1", "l1"); + assertEquals(1, tombstones.size()); + + tombstones.add("c1", "l2"); + assertEquals(2, tombstones.size()); + + tombstones.add("c2", "l1"); + assertEquals(3, tombstones.size()); + + tombstones.removeAllOf("c1"); + assertEquals(1, tombstones.size()); + } + + /** + * Verifies that real RocketMQ clientId formats (containing '@') and lmqName formats + * (e.g. topic@group for wildcard) do not cause cross-client collisions. + * This is the core safety guarantee of using '\0' separator instead of '@'. + */ + @Test + public void testRealClientIdFormat_NoCrossClientCollision() { + // RocketMQ clientId: IP@instanceName vs IP@instanceName@unitName + // clientA is a strict prefix of clientB if '@' were the separator + String clientA = "10.0.0.1@DEFAULT"; + String clientB = "10.0.0.1@DEFAULT@unit1"; + String lmqPlain = "lmq1"; + String lmqWildcard = "parentTopic@wildcardGroup"; // lmqName also contains '@' + + tombstones.add(clientA, lmqPlain); + tombstones.add(clientA, lmqWildcard); + tombstones.add(clientB, lmqPlain); + + // basic isolation + assertTrue(tombstones.contains(clientA, lmqPlain)); + assertTrue(tombstones.contains(clientA, lmqWildcard)); + assertTrue(tombstones.contains(clientB, lmqPlain)); + assertFalse(tombstones.contains(clientA, "parentTopic")); // partial lmqName no match + assertFalse(tombstones.contains("10.0.0.1@DEFAULT", lmqPlain + "@extra")); // no false match + assertEquals(3, tombstones.size()); + + // removeStale for clientA does NOT affect clientB + Set activeSet = new HashSet<>(); + activeSet.add(lmqWildcard); + tombstones.removeStale(clientA, activeSet); + + assertFalse(tombstones.contains(clientA, lmqPlain)); // removed + assertTrue(tombstones.contains(clientA, lmqWildcard)); // retained + assertTrue(tombstones.contains(clientB, lmqPlain)); // unaffected + assertEquals(2, tombstones.size()); + + // removeAllOf clientA does NOT affect clientB + tombstones.removeAllOf(clientA); + assertFalse(tombstones.contains(clientA, lmqWildcard)); + assertTrue(tombstones.contains(clientB, lmqPlain)); + assertEquals(1, tombstones.size()); + } +} diff --git a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java index d301e134633..7645a470962 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java @@ -699,4 +699,229 @@ public void testCleanupExpiredSubscriptions_CleanExpired() { assertFalse(registry.client2Subscription.containsKey(clientId)); assertEquals(0, registry.getActiveSubscriptionNum()); } + + // ==================== Exclusive Eviction Tombstone Tests ==================== + + /** + * Test: When clientB takes over lmq in exclusive mode, clientA gets a tombstone + */ + @Test + public void testExclusiveEviction_TombstoneCreatedOnEviction() { + String clientA = "clientA"; + String clientB = "clientB"; + String group = "exclusiveGroup"; + String topic = "testTopic"; + String lmqName = "lmq1"; + Set lmqNameSet = Collections.singleton(lmqName); + + // Configure exclusive mode + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + groupConfig.setLiteSubExclusive(true); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + when(mockLifecycleManager.isSubscriptionActive(topic, lmqName)).thenReturn(true); + + // clientA subscribes + registry.updateClientChannel(clientA, mock(Channel.class)); + registry.addPartialSubscription(clientA, group, topic, lmqNameSet, null); + + assertFalse(registry.hasExclusiveEvictionTombstone(clientA, lmqName)); + + // clientB takes over → clientA should get tombstone + registry.updateClientChannel(clientB, mock(Channel.class)); + registry.addPartialSubscription(clientB, group, topic, lmqNameSet, null); + + assertTrue(registry.hasExclusiveEvictionTombstone(clientA, lmqName)); + assertFalse(registry.hasExclusiveEvictionTombstone(clientB, lmqName)); + } + + /** + * Test: addCompleteSubscription clears stale tombstones but keeps active ones + */ + @Test + public void testExclusiveEviction_CompleteSyncClearsStale() { + String clientA = "clientA"; + String clientB = "clientB"; + String group = "exclusiveGroup"; + String topic = "testTopic"; + + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + groupConfig.setLiteSubExclusive(true); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + // clientA subscribes to lmq1 and lmq2 + Set initialSet = new HashSet<>(); + initialSet.add("lmq1"); + initialSet.add("lmq2"); + registry.updateClientChannel(clientA, mock(Channel.class)); + registry.addPartialSubscription(clientA, group, topic, initialSet, null); + + // clientB takes over lmq1 and lmq2 from clientA + registry.updateClientChannel(clientB, mock(Channel.class)); + registry.addPartialSubscription(clientB, group, topic, Collections.singleton("lmq1"), null); + registry.addPartialSubscription(clientB, group, topic, Collections.singleton("lmq2"), null); + + assertTrue(registry.hasExclusiveEvictionTombstone(clientA, "lmq1")); + assertTrue(registry.hasExclusiveEvictionTombstone(clientA, "lmq2")); + + // clientA does full sync with only lmq2 in active set → lmq1 tombstone should be cleaned + Set newActiveSet = new HashSet<>(); + newActiveSet.add("lmq2"); + newActiveSet.add("lmq3"); + registry.addCompleteSubscription(clientA, group, topic, newActiveSet, 2L); + + assertFalse(registry.hasExclusiveEvictionTombstone(clientA, "lmq1")); // cleared (not in newActiveSet) + assertTrue(registry.hasExclusiveEvictionTombstone(clientA, "lmq2")); // retained (in newActiveSet) + } + + /** + * Test: removeCompleteSubscription clears all tombstones for the client + */ + @Test + public void testExclusiveEviction_RemoveClientClearsTombstones() { + String clientA = "clientA"; + String clientB = "clientB"; + String group = "exclusiveGroup"; + String topic = "testTopic"; + + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + groupConfig.setLiteSubExclusive(true); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + // Setup: clientA subscribes then gets evicted by clientB + registry.updateClientChannel(clientA, mock(Channel.class)); + registry.addPartialSubscription(clientA, group, topic, Collections.singleton("lmq1"), null); + registry.addPartialSubscription(clientA, group, topic, Collections.singleton("lmq2"), null); + + registry.updateClientChannel(clientB, mock(Channel.class)); + registry.addPartialSubscription(clientB, group, topic, Collections.singleton("lmq1"), null); + registry.addPartialSubscription(clientB, group, topic, Collections.singleton("lmq2"), null); + + assertTrue(registry.hasExclusiveEvictionTombstone(clientA, "lmq1")); + assertTrue(registry.hasExclusiveEvictionTombstone(clientA, "lmq2")); + + // clientA disconnects + registry.removeCompleteSubscription(clientA); + + assertFalse(registry.hasExclusiveEvictionTombstone(clientA, "lmq1")); + assertFalse(registry.hasExclusiveEvictionTombstone(clientA, "lmq2")); + // clientB unaffected + assertFalse(registry.hasExclusiveEvictionTombstone(clientB, "lmq1")); + } + + /** + * Test: expired subscription cleanup also clears tombstones + */ + @Test + public void testExclusiveEviction_ExpiredCleanupClearsTombstones() { + String clientA = "clientA"; + String clientB = "clientB"; + String group = "exclusiveGroup"; + String topic = "testTopic"; + String lmqName = "lmq1"; + + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + groupConfig.setLiteSubExclusive(true); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + when(mockLifecycleManager.isSubscriptionActive(topic, lmqName)).thenReturn(true); + + // clientA subscribes, then gets evicted + registry.updateClientChannel(clientA, mock(Channel.class)); + registry.addPartialSubscription(clientA, group, topic, Collections.singleton(lmqName), null); + + registry.updateClientChannel(clientB, mock(Channel.class)); + registry.addPartialSubscription(clientB, group, topic, Collections.singleton(lmqName), null); + + assertTrue(registry.hasExclusiveEvictionTombstone(clientA, lmqName)); + + // Simulate clientA becoming expired (sub was already removed, manually add back for timeout test) + LiteSubscription expiredSub = new LiteSubscription(); + expiredSub.setGroup(group); + expiredSub.setTopic(topic); + expiredSub.setUpdateTime(System.currentTimeMillis() - 60000L); + registry.client2Subscription.put(clientA, expiredSub); + + registry.cleanupExpiredSubscriptions(10000L); + + assertFalse(registry.hasExclusiveEvictionTombstone(clientA, lmqName)); + } + + /** + * Test: addPartialSubscription clears stale tombstone when client re-claims an lmqName + */ + @Test + public void testExclusiveEviction_ReClaimClearsStaleTombstone() { + String clientA = "clientA"; + String clientB = "clientB"; + String group = "exclusiveGroup"; + String topic = "testTopic"; + String lmqName = "lmq1"; + Set lmqNameSet = Collections.singleton(lmqName); + + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + groupConfig.setLiteSubExclusive(true); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + when(mockLifecycleManager.isSubscriptionActive(topic, lmqName)).thenReturn(true); + + // clientA subscribes first + registry.updateClientChannel(clientA, mock(Channel.class)); + registry.addPartialSubscription(clientA, group, topic, lmqNameSet, null); + + // clientB takes over → clientA gets tombstone + registry.updateClientChannel(clientB, mock(Channel.class)); + registry.addPartialSubscription(clientB, group, topic, lmqNameSet, null); + + assertTrue(registry.hasExclusiveEvictionTombstone(clientA, lmqName)); + + // clientA re-claims the lmqName → tombstone should be cleared + registry.addPartialSubscription(clientA, group, topic, lmqNameSet, null); + + assertFalse(registry.hasExclusiveEvictionTombstone(clientA, lmqName)); + } + + /** + * Test: addCompleteSubscription re-sends unsubscribe for tombstoned lmqNames still in active set + */ + @Test + public void testExclusiveEviction_CompleteSyncReNotifiesForTombstonedLmq() { + String clientA = "clientA"; + String clientB = "clientB"; + String group = "exclusiveGroup"; + String topic = "testTopic"; + String lmqName = "lmq1"; + + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + groupConfig.setLiteSubExclusive(true); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + Channel clientAChannel = mock(Channel.class); + registry.updateClientChannel(clientA, clientAChannel); + registry.addPartialSubscription(clientA, group, topic, Collections.singleton(lmqName), null); + + // clientB takes over → clientA gets tombstone + registry.updateClientChannel(clientB, mock(Channel.class)); + registry.addPartialSubscription(clientB, group, topic, Collections.singleton(lmqName), null); + + assertTrue(registry.hasExclusiveEvictionTombstone(clientA, lmqName)); + + // clientA does full sync still reporting lmq1 → re-notify should be triggered + Set fullSet = new HashSet<>(); + fullSet.add(lmqName); + registry.addCompleteSubscription(clientA, group, topic, fullSet, 2L); + + // Verify notifyUnsubscribeLite was called for the tombstoned lmqName during full sync + // The first call is during eviction, the second is during re-notify + ArgumentCaptor captor = + ArgumentCaptor.forClass(NotifyUnsubscribeLiteRequestHeader.class); + verify(mockBroker2Client, org.mockito.Mockito.atLeast(2)) + .notifyUnsubscribeLite(eq(clientAChannel), captor.capture()); + } } 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 453cb8fd14e..fa9f2dbfaff 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 @@ -22,6 +22,7 @@ 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.longpolling.PopLiteLongPollingService; import org.apache.rocketmq.broker.offset.ConsumerOffsetManager; import org.apache.rocketmq.broker.pop.PopConsumerLockService; @@ -96,6 +97,8 @@ public class PopLiteMessageProcessorTest { private SubscriptionGroupManager subscriptionGroupManager; @Mock private AbstractLiteLifecycleManager liteLifecycleManager; + @Mock + private LiteSubscriptionRegistry liteSubscriptionRegistry; private BrokerConfig brokerConfig; private PopLiteMessageProcessor popLiteMessageProcessor; @@ -110,6 +113,7 @@ public void setUp() throws Exception { when(brokerController.getTopicConfigManager()).thenReturn(topicConfigManager); when(brokerController.getSubscriptionGroupManager()).thenReturn(subscriptionGroupManager); when(brokerController.getLiteLifecycleManager()).thenReturn(liteLifecycleManager); + when(brokerController.getLiteSubscriptionRegistry()).thenReturn(liteSubscriptionRegistry); PopLiteMessageProcessor testObject = new PopLiteMessageProcessor(brokerController, liteEventDispatcher); FieldUtils.writeDeclaredField(testObject, "popLiteLongPollingService", popLiteLongPollingService, true); @@ -487,4 +491,70 @@ private GetMessageResult mockGetMessageResult(GetMessageStatus status, int messa } return getMessageResult; } + + @SuppressWarnings("unchecked") + @Test + public void testPopByClientId_tombstoneRejectsPull() { + String clientId = "clientId"; + String group = "exclusiveGroup"; + String lmqName = "lmqName"; + + // Configure exclusive mode + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + groupConfig.setLiteSubExclusive(true); + when(subscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + + // Tombstone exists for this client+lmqName + when(liteSubscriptionRegistry.hasExclusiveEvictionTombstone(clientId, lmqName)).thenReturn(true); + + Iterator mockIterator = mock(Iterator.class); + when(mockIterator.hasNext()).thenReturn(true, false); + when(mockIterator.next()).thenReturn(lmqName); + when(liteEventDispatcher.getEventIterator(clientId)).thenReturn(mockIterator); + + Pair result = popLiteMessageProcessor.popByClientId( + "clientHost", "parentTopic", group, clientId, System.currentTimeMillis(), 6000L, 32, "attemptId"); + + // Should return empty result since tombstone blocks the pull + 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()); + } + + @SuppressWarnings("unchecked") + @Test + public void testPopByClientId_noTombstoneAllowsPull() { + String clientId = "clientId"; + String group = "exclusiveGroup"; + String lmqName = "lmqName"; + int msgCount = 1; + GetMessageResult mockResult = mockGetMessageResult(GetMessageStatus.FOUND, msgCount, 100L); + long pollTime = System.currentTimeMillis(); + + // Configure exclusive mode + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + groupConfig.setLiteSubExclusive(true); + when(subscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + + // No tombstone + when(liteSubscriptionRegistry.hasExclusiveEvictionTombstone(clientId, lmqName)).thenReturn(false); + + Iterator mockIterator = mock(Iterator.class); + when(mockIterator.hasNext()).thenReturn(true, false); + when(mockIterator.next()).thenReturn(lmqName); + when(liteEventDispatcher.getEventIterator(clientId)).thenReturn(mockIterator); + doReturn(new Pair<>(new StringBuilder("0"), mockResult)) + .when(popLiteMessageProcessor) + .popLiteTopic(anyString(), anyString(), anyString(), anyString(), anyLong(), anyLong(), anyLong(), anyString()); + + Pair 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"); + } } From a47ead1c316028d0f8aa4b3ff4b5623e2d487b00 Mon Sep 17 00:00:00 2001 From: Quan Date: Fri, 29 May 2026 13:45:19 +0800 Subject: [PATCH 2/2] docs(broker): remove implementation detail from tombstone test javadoc --- .../rocketmq/broker/lite/ExclusiveEvictionTombstonesTest.java | 1 - 1 file changed, 1 deletion(-) diff --git a/broker/src/test/java/org/apache/rocketmq/broker/lite/ExclusiveEvictionTombstonesTest.java b/broker/src/test/java/org/apache/rocketmq/broker/lite/ExclusiveEvictionTombstonesTest.java index 3df0f07d246..70f88ee14fd 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/lite/ExclusiveEvictionTombstonesTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/ExclusiveEvictionTombstonesTest.java @@ -117,7 +117,6 @@ public void testSize() { /** * Verifies that real RocketMQ clientId formats (containing '@') and lmqName formats * (e.g. topic@group for wildcard) do not cause cross-client collisions. - * This is the core safety guarantee of using '\0' separator instead of '@'. */ @Test public void testRealClientIdFormat_NoCrossClientCollision() {