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.

Author: f1amingo

broker/src/main/java/org/apache/rocketmq/broker/lite/ExclusiveEvictionTombstones.java

@@ -0,0 +1,90 @@
+/*
+ * 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;
+
+/**
+ * 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 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 + "@";
+        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 retainOnly(String clientId, Set<String> activeLmqNames) {
+        String prefix = clientId + "@";
+        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 + "@" + 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.retainOnly(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) {