[ISSUE #10384] Enforce Lite exclusive subscription eviction via server-side tombstones

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.
+ * <p>
+ * 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.
+ * <p>
+ * Key format: clientId$lmqName
+ */
+public class ExclusiveEvictionTombstones {
+
+    private static final char KEY_SEPARATOR = LiteUtil.SEPARATOR;
+
+    private final Set<String> 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<String> 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;
+    }
+}

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();

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);
     }
 
 }

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<StringBuilder, GetMessageResult> popByClientId(String clientHost, St
         StringBuilder orderCountInfoAll = new StringBuilder();
         AtomicLong total = new AtomicLong(0);
 
+        boolean isExclusiveGroup = LiteMetadataUtil.isSubLiteExclusive(group, brokerController);
         Set<String> processed = new HashSet<>(); // deduplication in one request
         Iterator<String> iterator = liteEventDispatcher.getEventIterator(clientId);
         while (total.get() < maxNum && iterator.hasNext()) {
@@ -263,6 +265,11 @@ public Pair<StringBuilder, GetMessageResult> 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<StringBuilder, GetMessageResult> pair = popLiteTopic(parentTopic, clientHost, group, lmqName,
                 maxNum - total.get(), popTime, invisibleTime, attemptId);
             if (null == pair || pair.getObject2().getMessageCount() <= 0) {

broker/src/test/java/org/apache/rocketmq/broker/lite/ExclusiveEvictionTombstonesTest.java

@@ -0,0 +1,158 @@
+/*
+ * 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<String> 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.
+     */
+    @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<String> 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());
+    }
+}