[ISSUE #10559] Support wildcard pattern matching for LiteTopic subscriptions

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

@@ -21,7 +21,6 @@
 import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.lang3.tuple.Triple;
 import org.apache.rocketmq.broker.BrokerController;
 import org.apache.rocketmq.broker.offset.ConsumerOffsetManager;
 import org.apache.rocketmq.common.BrokerConfig;
@@ -47,7 +46,6 @@
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
-import java.util.function.Function;
 
 public class LiteEventDispatcher extends ServiceThread {
 
@@ -96,6 +94,13 @@ public void dispatch(String group, String lmqName, int queueId, long offset, lon
         if (queueId != 0 || !LiteUtil.isLiteTopicQueue(lmqName)) {
             return;
         }
+        // On the message-arriving path (group == null), lazily register this lmqName against any
+        // pattern-mode wildcard clients whose patterns match it, so the dispatch below can reach
+        // them even for a newly-created lite-topic. The periodic doFullDispatchForWildcardGroup
+        // re-expand remains as a backstop.
+        if (group == null) {
+            liteSubscriptionRegistry.registerArrivingLmqForPatternClients(lmqName);
+        }
         doDispatch(group, lmqName, null);
     }
 
@@ -253,41 +258,75 @@ public void doFullDispatchForClient(String clientId, String group) {
 
     /**
      * Perform a full dispatch for wildcard group which was previously marked for a delayed full dispatch.
-     * It iterates through all LMQ topics in CQ table, so it may be a heavy work.
+     *
+     * <p>For legacy wildcard groups (receiving all lite-topics under the parent topic) it iterates
+     * the lite-topics under the parent topic via {@code collectByParentTopic} (O(M)) instead of
+     * scanning the entire LMQ table. For pattern-mode wildcard groups it first re-expands the
+     * client patterns (picking up lite-topics created since the initial subscription) and then
+     * delegates to {@link #doFullDispatchForClient(String, String)} to clear backlog.
      */
     public void doFullDispatchForWildcardGroup(String group) {
         String parentTopic = LiteMetadataUtil.getLiteBindTopic(group, brokerController);
         if (null == parentTopic || !LiteMetadataUtil.isWildcardGroup(group, brokerController)) {
             return;
         }
-        List<ClientGroup> clients =  liteSubscriptionRegistry.getWildcardSubscriber(group, parentTopic).getClients();
-        if (CollectionUtils.isEmpty(clients)) {
-            return;
+
+        // Partition clients into pattern-mode (re-expand + per-client dispatch) and legacy
+        // (single shared dispatch over all parent-topic lite-topics).
+        List<String> clientIds = liteSubscriptionRegistry.getAllClientIdByGroup(group);
+        List<String> patternClientIds = new ArrayList<>();
+        boolean hasLegacyClient = false;
+        for (String clientId : clientIds) {
+            LiteSubscription subscription = liteSubscriptionRegistry.getLiteSubscription(clientId);
+            if (subscription != null && !subscription.getWildcardPatterns().isEmpty()) {
+                patternClientIds.add(clientId);
+            } else {
+                hasLegacyClient = true;
+            }
         }
-        AtomicInteger count = new AtomicInteger();
-        Function<Triple<String, Long, Long>, Boolean> function = triple -> {
-            String lmqName = triple.getLeft();
-            long maxOffset = triple.getMiddle();
-            if (!LiteUtil.belongsTo(lmqName, parentTopic)) {
-                return true;
+
+        int count = 0;
+        // Pattern-mode: re-expand patterns against current lite-topics, then dispatch backlog per client.
+        for (String clientId : patternClientIds) {
+            liteSubscriptionRegistry.reexpandWildcardPatterns(clientId);
+            doFullDispatchForClient(clientId, group);
+        }
+
+        // Legacy-mode: dispatch over all lite-topics under the parent topic to the shared client set.
+        if (hasLegacyClient) {
+            List<ClientGroup> clients = liteSubscriptionRegistry.getWildcardSubscriber(group, parentTopic).getClients();
+            if (CollectionUtils.isNotEmpty(clients)) {
+                count += dispatchLegacyWildcardGroup(group, parentTopic, clients);
             }
+        }
+        LOGGER.info("doFullDispatchForWildcardGroup finish. {}, patternClients:{}, legacyDispatch:{}",
+            group, patternClientIds.size(), count);
+    }
+
+    /**
+     * Iterate lite-topics under {@code parentTopic} (O(M), not the full LMQ scan) and dispatch the
+     * ones with consumer lag to the shared legacy wildcard client set.
+     */
+    private int dispatchLegacyWildcardGroup(String group, String parentTopic, List<ClientGroup> clients) {
+        List<String> lmqNames = liteLifecycleManager.collectByParentTopic(parentTopic);
+        AtomicInteger count = new AtomicInteger();
+        for (String lmqName : lmqNames) {
+            long maxOffset = liteLifecycleManager.getMaxOffsetInQueue(lmqName);
             if (maxOffset <= 0) {
-                return true;
+                continue;
             }
             long consumerOffset = consumerOffsetManager.queryOffset(group, lmqName, 0);
             if (consumerOffset >= maxOffset) {
-                return true;
+                continue;
             }
             if (selectAndDispatch(lmqName, clients, null)) {
                 count.incrementAndGet();
             } else {
                 LOGGER.warn("doFullDispatchForWildcardGroup, wait another period. {}", group);
-                return false;
+                break;
             }
-            return true;
-        };
-        liteLifecycleManager.forEachLiteTopic(function);
-        LOGGER.info("doFullDispatchForWildcardGroup finish. {}, dispatch:{}", group, count);
+        }
+        return count.get();
     }
 
     /**

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

@@ -38,6 +38,54 @@ public interface LiteSubscriptionRegistry {
 
     void addCompleteSubscription(String clientId, String group, String topic, Set<String> newLmqNameSet, long version);
 
+    /**
+     * Complete-add overload carrying wildcard patterns for a wildcard group in pattern mode.
+     *
+     * <p>When {@code wildcardPatterns} is non-empty, the group is in <em>pattern mode</em>: the
+     * patterns are eagerly expanded against the existing lite-topics under {@code topic} and the
+     * matched lmqNames are registered (as a normal subscription) into {@code liteTopic2Group}.
+     * When empty, the call delegates to {@link #addCompleteSubscription(String, String, String,
+     * Set, long)} (legacy wildcard group receives all lite-topics under the parent topic).
+     *
+     * @param clientId          the client id
+     * @param group             the consumer group
+     * @param topic             the parent (lite-bind) topic
+     * @param newLmqNameSet     literal lmqNames (used only by the non-wildcard delegate path)
+     * @param wildcardPatterns  wildcard patterns for pattern mode (empty for legacy mode)
+     * @param version           the subscription version
+     */
+    void addCompleteSubscription(String clientId, String group, String topic, Set<String> newLmqNameSet,
+        Set<String> wildcardPatterns, long version);
+
+    /**
+     * Lazily re-expand a pattern-mode wildcard client's stored patterns against the current
+     * lite-topics under its parent topic, registering any newly-matched lmqNames into
+     * {@code liteTopic2Group}. This is the mechanism by which a pattern-mode client picks up
+     * lite-topics created after its initial {@code COMPLETE_ADD}.
+     *
+     * @param clientId the client id
+     * @return the number of newly-registered lmqNames (0 if the client is not in pattern mode or
+     *         nothing new matched)
+     */
+    int reexpandWildcardPatterns(String clientId);
+
+    /**
+     * Lazily register the arriving lmqName against any pattern-mode wildcard clients whose stored
+     * patterns match it, so that the current message-arriving dispatch can reach them. This is the
+     * single-lmqName counterpart to {@link #reexpandWildcardPatterns(String)}: it matches only
+     * against the one arriving lmqName (avoiding the O(M) {@code collectByParentTopic} scan on the
+     * message-arriving hot path) and lets a pattern-mode client receive a newly-created matching
+     * lite-topic without waiting for the periodic full-dispatch re-expand.
+     *
+     * <p>Only pattern-mode clients (non-empty {@code wildcardPatterns}) are considered; legacy
+     * wildcard clients are already reachable via the synthetic {@code topic@group} key in
+     * {@link #getAllSubscriber}. Registration is idempotent via {@code addTopicGroup}.
+     *
+     * @param lmqName the full lmqName of the arriving message
+     * @return the number of newly-registered (client, lmqName) pairs
+     */
+    int registerArrivingLmqForPatternClients(String lmqName);
+
     void removeCompleteSubscription(String clientId);
 
     void addListener(LiteCtlListener listener);