From 4d51ad9997a2168571fab8a294c17dedd816066e Mon Sep 17 00:00:00 2001 From: liuhy Date: Fri, 10 Jul 2026 12:50:39 +0800 Subject: [PATCH 1/4] [ISSUE #10559] Support wildcard pattern matching for LiteTopic subscriptions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Add NATS-style wildcard pattern matching so a wildcard LiteTopic consumer group can subscribe to a subset of lite-topics under its parent topic, using "__" as the segment separator. - LitePatternMatcher (new, common): validate/matches/matchesAny/expand. "*" matches one segment; "**" matches one or more trailing segments. Recursive, no regex; internal pre-split for the expand hot path. - LiteSubscription: add in-memory wildcardPatterns set. - LiteSubscriptionRegistry[Impl]: a wildcard group with non-empty validated patterns enters pattern mode — eagerly expand against collectByParentTopic and register real lmqNames. Empty patterns keep legacy receive-all behavior (synthetic topic@group key) unchanged. getAllSubscriber always merges the synthetic-key clients for any wildcard group so mixed pattern/legacy groups deliver to both; reexpandWildcardPatterns picks up lite-topics created after the initial subscription. - LiteSubscriptionCtlProcessor: route wildcard COMPLETE_ADD through toWildcardPatterns (patterns ride the existing liteTopicSet — no wire change). Non-empty but all-invalid pattern input is rejected instead of silently widening to receive-all. - LiteEventDispatcher.doFullDispatchForWildcardGroup: replace the O(N) forEachLiteTopic scan with O(M) collectByParentTopic; re-expand patterns for new LMQs and clear backlog per pattern-mode client. Full backward compatibility: wildcard groups without patterns behave identically to before. Affected lite tests pass; checkstyle clean. Co-Authored-By: Claude --- .../broker/lite/LiteEventDispatcher.java | 70 ++++-- .../broker/lite/LiteSubscriptionRegistry.java | 31 +++ .../lite/LiteSubscriptionRegistryImpl.java | 116 ++++++++- .../LiteSubscriptionCtlProcessor.java | 51 +++- .../broker/lite/LiteEventDispatcherTest.java | 74 ++++++ .../LiteSubscriptionRegistryImplTest.java | 223 ++++++++++++++++++ .../LiteSubscriptionCtlProcessorTest.java | 73 ++++++ .../common/lite/LitePatternMatcher.java | 218 +++++++++++++++++ .../common/lite/LiteSubscription.java | 25 ++ .../common/lite/LitePatternMatcherTest.java | 198 ++++++++++++++++ 10 files changed, 1052 insertions(+), 27 deletions(-) create mode 100644 common/src/main/java/org/apache/rocketmq/common/lite/LitePatternMatcher.java create mode 100644 common/src/test/java/org/apache/rocketmq/common/lite/LitePatternMatcherTest.java diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java index 2bd1f36186b..afa39ba8dad 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java +++ b/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 { @@ -253,41 +251,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. + * + *

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 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 clientIds = liteSubscriptionRegistry.getAllClientIdByGroup(group); + List 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, 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 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 clients) { + List 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(); } /** 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 965ed180fc6..2baf840da6b 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 @@ -38,6 +38,37 @@ public interface LiteSubscriptionRegistry { void addCompleteSubscription(String clientId, String group, String topic, Set newLmqNameSet, long version); + /** + * Complete-add overload carrying wildcard patterns for a wildcard group in pattern mode. + * + *

When {@code wildcardPatterns} is non-empty, the group is in pattern mode: 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 newLmqNameSet, + Set 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); + void removeCompleteSubscription(String clientId); void addListener(LiteCtlListener listener); 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 b487b8757f4..d61ef4bf3ad 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 @@ -23,6 +23,7 @@ import io.netty.channel.Channel; import java.util.ArrayList; import java.util.Collections; +import java.util.HashSet; import java.util.List; import java.util.Map; import java.util.Objects; @@ -39,6 +40,7 @@ import org.apache.rocketmq.common.ServiceThread; import org.apache.rocketmq.common.constant.LoggerName; import org.apache.rocketmq.common.entity.ClientGroup; +import org.apache.rocketmq.common.lite.LitePatternMatcher; import org.apache.rocketmq.common.lite.LiteSubscription; import org.apache.rocketmq.common.lite.LiteUtil; import org.apache.rocketmq.common.lite.OffsetOption; @@ -124,8 +126,27 @@ public void removePartialSubscription(String clientId, String group, String topi @Override public void addCompleteSubscription(String clientId, String group, String topic, Set lmqNameAll, long version) { + addCompleteSubscription(clientId, group, topic, lmqNameAll, Collections.emptySet(), version); + } + + @Override + public void addCompleteSubscription(String clientId, String group, String topic, Set lmqNameAll, + Set wildcardPatterns, long version) { Set lmqNameNew; - if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) { + boolean isWildcardGroup = LiteMetadataUtil.isWildcardGroup(group, brokerController); + boolean isPatternMode = isWildcardGroup && wildcardPatterns != null && !wildcardPatterns.isEmpty(); + + LiteSubscription thisSub = getOrCreateLiteSubscription(clientId, group, topic); + if (isPatternMode) { + // Persist the patterns (authoritative intent) and eagerly expand against existing + // lite-topics under the parent topic. The expanded lmqNames are registered as a normal + // subscription; the topic@group synthetic key is NOT used for pattern-mode groups. + thisSub.setWildcardPatterns(wildcardPatterns); + lmqNameNew = expandWildcardPatterns(topic, wildcardPatterns); + markWildcardGroup(topic, group); + } else if (isWildcardGroup) { + // Legacy wildcard group: receive all lite-topics under the parent topic via the + // synthetic topic@group key. lmqNameNew = Collections.singleton(mockLmqNameForWildcardGroup(topic, group)); markWildcardGroup(topic, group); } else { @@ -134,7 +155,6 @@ public void addCompleteSubscription(String clientId, String group, String topic, .collect(Collectors.toSet()); } - LiteSubscription thisSub = getOrCreateLiteSubscription(clientId, group, topic); Set lmqNamePrev = thisSub.getLiteTopicSet(); // Find topics to remove (in current set but not in new set) Set lmqNameRemove = lmqNamePrev.stream() @@ -167,6 +187,60 @@ public void addCompleteSubscription(String clientId, String group, String topic, } } + /** + * Eagerly expand wildcard patterns against the existing lite-topics under {@code parentTopic}. + * Returns the set of matched full lmqNames that this broker is responsible for + * (sharding-aware via {@link AbstractLiteLifecycleManager#isSubscriptionActive}). + */ + private Set expandWildcardPatterns(String parentTopic, Set wildcardPatterns) { + List candidates = liteLifecycleManager.collectByParentTopic(parentTopic); + if (candidates.isEmpty()) { + return Collections.emptySet(); + } + Set matchedLmqNames = new HashSet<>(); + for (String lmqName : candidates) { + String child = LiteUtil.getLiteTopic(lmqName); + if (child == null) { + continue; + } + if (LitePatternMatcher.matchesAny(wildcardPatterns, child) + && liteLifecycleManager.isSubscriptionActive(parentTopic, lmqName)) { + matchedLmqNames.add(lmqName); + } + } + return matchedLmqNames; + } + + @Override + public int reexpandWildcardPatterns(String clientId) { + LiteSubscription thisSub = client2Subscription.get(clientId); + if (thisSub == null || CollectionUtils.isEmpty(thisSub.getWildcardPatterns())) { + return 0; + } + String topic = thisSub.getTopic(); + String group = thisSub.getGroup(); + Set matched = expandWildcardPatterns(topic, thisSub.getWildcardPatterns()); + Set current = thisSub.getLiteTopicSet(); + // Register only the newly-matched lmqNames; existing ones are already subscribed. + Set toAdd = matched.stream().filter(lmqName -> !current.contains(lmqName)).collect(Collectors.toSet()); + if (toAdd.isEmpty()) { + return 0; + } + ClientGroup clientGroup = new ClientGroup(clientId, group); + int added = 0; + for (String lmqName : toAdd) { + thisSub.addLiteTopic(lmqName); + if (addTopicGroup(clientGroup, lmqName)) { + added++; + } + } + if (added > 0) { + LOGGER.info("reexpandWildcardPatterns, clientId:{}, group:{}, topic:{}, newly matched:{}", + clientId, group, topic, added); + } + return added; + } + @Override public void removeCompleteSubscription(String clientId) { clientChannels.remove(clientId); @@ -183,6 +257,13 @@ public void removeCompleteSubscription(String clientId) { thisSub.getLiteTopicSet().forEach(lmqName -> { removeTopicGroup(clientGroup, lmqName, false); }); + // Pattern-mode wildcard groups are marked in wildcardGroupMap for fan-out enumeration but + // register real lmqNames (so unmarkWildcardGroupIfNecessary, which parses the synthetic + // topic@group key, never fires for them). Clean up explicitly when the last client leaves. + if (CollectionUtils.isNotEmpty(thisSub.getWildcardPatterns()) + && getAllClientIdByGroup(thisSub.getGroup()).isEmpty()) { + unmarkWildcardGroup(thisSub.getTopic(), thisSub.getGroup()); + } for (LiteCtlListener listener : listeners) { listener.onRemoveAll(clientId, thisSub.getGroup()); } @@ -204,18 +285,28 @@ public void addListener(LiteCtlListener listener) { @Override public SubscriberWrapper getAllSubscriber(String group, String lmqName) { String topic = LiteUtil.getParentTopic(lmqName); + boolean isWildcardGroup = group != null && LiteMetadataUtil.isWildcardGroup(group, brokerController); if (group != null) { - if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) { - return getWildcardSubscriber(group, topic); - } SubscriberWrapper.ListWrapper wrapper = new SubscriberWrapper.ListWrapper(); + // Pattern-mode wildcard groups register real lmqNames into liteTopic2Group (same as a + // normal subscription), so the normal lookup finds their clients here. Set subscribers = liteTopic2Group.get(lmqName); if (subscribers != null) { wrapper.getClients().addAll(subscribers.stream() .filter(clientGroup -> group.equals(clientGroup.group)) .collect(Collectors.toSet())); } + // For a wildcard group, always merge clients from the synthetic topic@group key. Legacy + // wildcard clients (receive-all) live there; pattern-mode clients never do (they register + // real lmqNames, already merged above). This keeps delivery correct in mixed groups where + // some clients use patterns and others are legacy — both sets are returned. + if (isWildcardGroup) { + List wildcardClients = getWildcardGroupClients(topic, group); + if (CollectionUtils.isNotEmpty(wildcardClients)) { + wrapper.getClients().addAll(wildcardClients); + } + } return wrapper; } else { SubscriberWrapper.MapWrapper wrapper = new SubscriberWrapper.MapWrapper(); @@ -225,6 +316,10 @@ public SubscriberWrapper getAllSubscriber(String group, String lmqName) { wrapper.getGroupMap().computeIfAbsent(clientGroup.group, k -> new ArrayList<>()).add(clientGroup); } } + // Fan out wildcard groups via the synthetic topic@group key. Legacy wildcard clients are + // only reachable there; pattern-mode wildcard clients are already in liteTopic2Group + // (merged above) and absent from the synthetic key, so they are not double-counted. A + // group may contain both kinds, so enumerate every wildcard group without skipping. Set wildcardGroups = wildcardGroupMap.get(topic); if (wildcardGroups != null) { for (String wildcardGroup : wildcardGroups) { @@ -269,7 +364,7 @@ public void cleanSubscription(String lmqName, boolean notifyClient) { } } - protected void addTopicGroup(ClientGroup clientGroup, String lmqName) { + protected boolean addTopicGroup(ClientGroup clientGroup, String lmqName) { Set topicGroupSet = liteTopic2Group .computeIfAbsent(lmqName, k -> ConcurrentHashMap.newKeySet()); if (topicGroupSet.add(clientGroup)) { @@ -278,7 +373,9 @@ protected void addTopicGroup(ClientGroup clientGroup, String lmqName) { for (LiteCtlListener listener : listeners) { listener.onRegister(clientGroup.clientId, clientGroup.group, lmqName); } + return true; } + return false; } protected void removeTopicGroup(ClientGroup clientGroup, String lmqName, boolean resetOffset) { @@ -424,6 +521,13 @@ private void markWildcardGroup(String topic, String group) { wildcardGroupMap.computeIfAbsent(topic, k -> ConcurrentHashMap.newKeySet()).add(group); } + private void unmarkWildcardGroup(String topic, String group) { + wildcardGroupMap.computeIfPresent(topic, (k, v) -> { + v.remove(group); + return v.isEmpty() ? null : v; + }); + } + private void unmarkWildcardGroupIfNecessary(String lmqName) { if (!LiteUtil.isLiteTopicQueue(lmqName)) { // must be topic@group String[] topicAtGroup = StringUtils.split(lmqName, "@"); diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteSubscriptionCtlProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteSubscriptionCtlProcessor.java index bcf0df41270..d13dc061d70 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteSubscriptionCtlProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteSubscriptionCtlProcessor.java @@ -28,6 +28,7 @@ import org.apache.rocketmq.broker.lite.LiteQuotaException; import org.apache.rocketmq.broker.lite.LiteMetadataUtil; import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.common.lite.LitePatternMatcher; import org.apache.rocketmq.common.lite.LiteSubscriptionDTO; import org.apache.rocketmq.common.lite.LiteUtil; import org.apache.rocketmq.remoting.netty.NettyRequestProcessor; @@ -94,8 +95,16 @@ public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand case COMPLETE_ADD: checkConsumeEnable(group); this.liteSubscriptionRegistry.updateClientChannel(clientId, ctx.channel()); - this.liteSubscriptionRegistry.addCompleteSubscription(clientId, group, topic, lmqNameSet, - entry.getVersion()); + if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) { + // For a wildcard group, liteTopicSet carries pattern strings (not literal + // lmqNames); pass them through verbatim and let the registry expand them. + Set wildcardPatterns = toWildcardPatterns(entry); + this.liteSubscriptionRegistry.addCompleteSubscription(clientId, group, topic, lmqNameSet, + wildcardPatterns, entry.getVersion()); + } else { + this.liteSubscriptionRegistry.addCompleteSubscription(clientId, group, topic, lmqNameSet, + entry.getVersion()); + } break; case COMPLETE_REMOVE: this.liteSubscriptionRegistry.removeCompleteSubscription(clientId); @@ -128,4 +137,42 @@ private Set toLmqNameSet(LiteSubscriptionDTO liteSubscriptionDTO) { .collect(Collectors.toSet()); } + /** + * Extract wildcard patterns from a wildcard group's {@code liteTopicSet}. Pattern strings are + * passed through verbatim (they contain {@code *} / {@code **} and must not be converted to + * lmqNames); entries that fail {@link LitePatternMatcher#validate(String)} are dropped with a + * warning rather than aborting the whole request. + * + *

A genuinely empty {@code liteTopicSet} means the group runs in legacy wildcard mode + * (receive all lite-topics under the parent topic). But a non-empty set in which every + * pattern is invalid must NOT silently widen into legacy/full-wildcard consumption — that would + * turn a malformed restrictive subscription into receiving everything. Such input is rejected + * with an {@link IllegalStateException} (mapped to {@code ILLEGAL_OPERATION}). + * + * @return the validated patterns; empty only when the input {@code liteTopicSet} was empty + * @throws IllegalStateException if the input was non-empty but contained no valid patterns + */ + private Set toWildcardPatterns(LiteSubscriptionDTO liteSubscriptionDTO) { + Set raw = liteSubscriptionDTO.getLiteTopicSet(); + if (CollectionUtils.isEmpty(raw)) { + return Collections.emptySet(); + } + Set patterns = raw.stream() + .filter(pattern -> { + if (LitePatternMatcher.validate(pattern)) { + return true; + } + log.warn("drop invalid wildcard pattern, group:{}, topic:{}, pattern:{}", + liteSubscriptionDTO.getGroup(), liteSubscriptionDTO.getTopic(), pattern); + return false; + }) + .collect(Collectors.toSet()); + if (patterns.isEmpty()) { + // Non-empty input but no valid pattern: reject rather than widen to receive-all. + throw new IllegalStateException("all wildcard patterns are invalid, group:" + + liteSubscriptionDTO.getGroup() + ", topic:" + liteSubscriptionDTO.getTopic()); + } + return patterns; + } + } diff --git a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java index 84aec24829f..3c90e85733f 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java @@ -57,6 +57,7 @@ import static org.mockito.ArgumentMatchers.anyList; 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.mock; import static org.mockito.Mockito.never; @@ -541,4 +542,77 @@ public void testScan_FullDispatch() { liteEventDispatcher.scan(); assertTrue(liteEventDispatcher.fullDispatchSet.isEmpty()); } + + // ==================== doFullDispatchForWildcardGroup ==================== + + private SubscriptionGroupConfig wildcardGroupConfig(String group, String bindTopic) { + SubscriptionGroupConfig config = new SubscriptionGroupConfig(); + config.setGroupName(group); + config.setWildcardLiteGroup(true); + config.setLiteBindTopic(bindTopic); + return config; + } + + /** + * Legacy wildcard group: dispatch iterates collectByParentTopic (O(M)) instead of the full + * forEachLiteTopic scan, dispatching lagged lite-topics to the shared client set. + */ + @Test + public void testDoFullDispatchForWildcardGroup_LegacyMode_UsesCollectByParentTopic() { + String group = "wildcardGroup"; + String parentTopic = "order_events"; + String lmqName = LiteUtil.toLmqName(parentTopic, "pay__refund"); + when(subscriptionGroupManager.findSubscriptionGroupConfig(group)) + .thenReturn(wildcardGroupConfig(group, parentTopic)); + when(liteLifecycleManager.collectByParentTopic(parentTopic)).thenReturn(Collections.singletonList(lmqName)); + when(liteLifecycleManager.getMaxOffsetInQueue(lmqName)).thenReturn(100L); + when(consumerOffsetManager.queryOffset(group, lmqName, 0)).thenReturn(50L); + // no pattern-mode clients: getAllClientIdByGroup returns a legacy client id whose + // subscription has empty wildcardPatterns + when(liteSubscriptionRegistry.getAllClientIdByGroup(group)).thenReturn(Collections.singletonList("legacyClient")); + LiteSubscription legacySub = new LiteSubscription().setGroup(group).setTopic(parentTopic); + when(liteSubscriptionRegistry.getLiteSubscription("legacyClient")).thenReturn(legacySub); + List clients = Collections.singletonList(new ClientGroup("legacyClient", group)); + SubscriberWrapper.ListWrapper wrapper = mock(SubscriberWrapper.ListWrapper.class); + when(wrapper.getClients()).thenReturn(clients); + when(liteSubscriptionRegistry.getWildcardSubscriber(group, parentTopic)).thenReturn(wrapper); + + LiteEventDispatcher spyDispatcher = spy(liteEventDispatcher); + doReturn(true).when(spyDispatcher).selectAndDispatch(eq(lmqName), eq(clients), eq(null)); + + spyDispatcher.doFullDispatchForWildcardGroup(group); + + // collectByParentTopic is used, not forEachLiteTopic + verify(liteLifecycleManager).collectByParentTopic(parentTopic); + verify(liteLifecycleManager, never()).forEachLiteTopic(any()); + verify(spyDispatcher).selectAndDispatch(lmqName, clients, null); + } + + /** + * Pattern-mode wildcard group: patterns are re-expanded and backlog is dispatched per client + * via doFullDispatchForClient. + */ + @Test + public void testDoFullDispatchForWildcardGroup_PatternMode_ReexpandsAndDispatches() { + String group = "wildcardGroup"; + String parentTopic = "order_events"; + String clientId = "patternClient"; + when(subscriptionGroupManager.findSubscriptionGroupConfig(group)) + .thenReturn(wildcardGroupConfig(group, parentTopic)); + when(liteSubscriptionRegistry.getAllClientIdByGroup(group)).thenReturn(Collections.singletonList(clientId)); + LiteSubscription patternSub = new LiteSubscription().setGroup(group).setTopic(parentTopic); + patternSub.setWildcardPatterns(Collections.singleton("pay__*")); + when(liteSubscriptionRegistry.getLiteSubscription(clientId)).thenReturn(patternSub); + + LiteEventDispatcher spyDispatcher = spy(liteEventDispatcher); + // re-expansion delegates to the registry; per-client backlog dispatch delegates to doFullDispatchForClient + doNothing().when(spyDispatcher).doFullDispatchForClient(clientId, group); + + spyDispatcher.doFullDispatchForWildcardGroup(group); + + verify(liteSubscriptionRegistry).reexpandWildcardPatterns(clientId); + verify(spyDispatcher).doFullDispatchForClient(clientId, group); + // legacy path (getWildcardSubscriber) must NOT be taken for a pure pattern-mode group + verify(liteSubscriptionRegistry, never()).getWildcardSubscriber(anyString(), anyString()); + } } \ No newline at end of file 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 7645a470962..026a65bc385 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 @@ -18,6 +18,7 @@ package org.apache.rocketmq.broker.lite; import io.netty.channel.Channel; +import java.util.Arrays; import java.util.Collections; import java.util.HashSet; import java.util.List; @@ -335,6 +336,228 @@ public void testRemoveCompleteSubscription_WildcardGroupMetadataCleanup() { assertEquals(0, registry.getActiveSubscriptionNum()); } + // ==================== Pattern-mode Wildcard Group Tests ==================== + + private SubscriptionGroupConfig wildcardGroupConfig(String group) { + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + groupConfig.setWildcardLiteGroup(true); + return groupConfig; + } + + /** + * Pattern-mode wildcard group: patterns are eagerly expanded against existing lite-topics + * under the parent topic, and matched lmqNames are registered as a normal subscription. + */ + @Test + public void testAddCompleteSubscription_PatternWildcardGroup_ExpandsAgainstExistingLmq() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "order_events"; + // candidates returned by collectByParentTopic (full lmqNames) + String payRefund = LiteUtil.toLmqName(topic, "pay__refund"); + String paySuccess = LiteUtil.toLmqName(topic, "pay__success"); + String payRefundNotify = LiteUtil.toLmqName(topic, "pay__refund__notify"); + String orderCreated = LiteUtil.toLmqName(topic, "order__created"); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn( + Arrays.asList(payRefund, paySuccess, payRefundNotify, orderCreated)); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + Set patterns = new HashSet<>(Collections.singletonList("pay__*")); + registry.addCompleteSubscription(clientId, group, topic, Collections.emptySet(), patterns, 1L); + + LiteSubscription subscription = registry.getLiteSubscription(clientId); + assertNotNull(subscription); + assertEquals(patterns, subscription.getWildcardPatterns()); + // pay__* matches pay__refund and pay__success only (single-level) + assertTrue(subscription.getLiteTopicSet().contains(payRefund)); + assertTrue(subscription.getLiteTopicSet().contains(paySuccess)); + assertFalse(subscription.getLiteTopicSet().contains(payRefundNotify)); + assertFalse(subscription.getLiteTopicSet().contains(orderCreated)); + // No synthetic topic@group key for pattern-mode groups + assertFalse(subscription.getLiteTopicSet().contains(topic + "@" + group)); + // marked in wildcardGroupMap for fan-out enumeration + assertTrue(registry.wildcardGroupMap.containsKey(topic)); + assertTrue(registry.wildcardGroupMap.get(topic).contains(group)); + assertEquals(2, registry.getActiveSubscriptionNum()); + } + + /** + * Re-subscribing with changed patterns removes old matched lmqNames and adds new ones. + */ + @Test + public void testAddCompleteSubscription_PatternWildcardGroup_DiffOnResubscribe() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "order_events"; + String payRefund = LiteUtil.toLmqName(topic, "pay__refund"); + String notifyRefundSms = LiteUtil.toLmqName(topic, "notify__refund__sms"); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn(Arrays.asList(payRefund, notifyRefundSms)); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + // first subscribe: pay__* + registry.addCompleteSubscription(clientId, group, topic, Collections.emptySet(), + new HashSet<>(Collections.singletonList("pay__*")), 1L); + LiteSubscription subscription = registry.getLiteSubscription(clientId); + assertTrue(subscription.getLiteTopicSet().contains(payRefund)); + assertFalse(subscription.getLiteTopicSet().contains(notifyRefundSms)); + + // re-subscribe: notify__** + registry.addCompleteSubscription(clientId, group, topic, Collections.emptySet(), + new HashSet<>(Collections.singletonList("notify__**")), 2L); + subscription = registry.getLiteSubscription(clientId); + assertFalse(subscription.getLiteTopicSet().contains(payRefund)); + assertTrue(subscription.getLiteTopicSet().contains(notifyRefundSms)); + assertEquals(1, registry.getActiveSubscriptionNum()); + } + + /** + * getAllSubscriber for a pattern-mode wildcard group uses the normal liteTopic2Group path. + */ + @Test + public void testGetAllSubscriber_PatternWildcardGroup_UsesNormalPath() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "order_events"; + String payRefund = LiteUtil.toLmqName(topic, "pay__refund"); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn(Collections.singletonList(payRefund)); + when(mockLifecycleManager.isSubscriptionActive(topic, payRefund)).thenReturn(true); + + registry.addCompleteSubscription(clientId, group, topic, Collections.emptySet(), + new HashSet<>(Collections.singletonList("pay__*")), 1L); + + SubscriberWrapper result = registry.getAllSubscriber(group, payRefund); + assertNotNull(result); + assertInstanceOf(SubscriberWrapper.ListWrapper.class, result); + SubscriberWrapper.ListWrapper listWrapper = (SubscriberWrapper.ListWrapper) result; + assertEquals(1, listWrapper.getClients().size()); + assertEquals(clientId, listWrapper.getClients().get(0).clientId); + assertEquals(group, listWrapper.getClients().get(0).group); + } + + /** + * removeCompleteSubscription cleans up wildcardGroupMap for pattern-mode groups when the last + * client leaves (the synthetic-key path does not fire for them). + */ + @Test + public void testRemoveCompleteSubscription_PatternWildcardGroupMetadataCleanup() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "order_events"; + String payRefund = LiteUtil.toLmqName(topic, "pay__refund"); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn(Collections.singletonList(payRefund)); + when(mockLifecycleManager.isSubscriptionActive(topic, payRefund)).thenReturn(true); + + registry.addCompleteSubscription(clientId, group, topic, Collections.emptySet(), + new HashSet<>(Collections.singletonList("pay__*")), 1L); + assertTrue(registry.wildcardGroupMap.containsKey(topic)); + + registry.removeCompleteSubscription(clientId); + + assertFalse(registry.wildcardGroupMap.containsKey(topic)); + assertNull(registry.getLiteSubscription(clientId)); + assertEquals(0, registry.getActiveSubscriptionNum()); + } + + /** + * Backward compat: a legacy wildcard group (empty patterns) still uses the synthetic topic@group + * key and receives all lite-topics under the parent topic. + */ + @Test + public void testAddCompleteSubscription_LegacyWildcardGroup_StillUsesSyntheticKey() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "testTopic"; + Set lmqNameAll = new HashSet<>(Arrays.asList("lmq1", "lmq2")); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + + // 5-arg overload delegates with empty patterns -> legacy mode + registry.addCompleteSubscription(clientId, group, topic, lmqNameAll, 1L); + + assertTrue(registry.wildcardGroupMap.containsKey(topic)); + LiteSubscription subscription = registry.getLiteSubscription(clientId); + assertNotNull(subscription); + assertTrue(subscription.getWildcardPatterns().isEmpty()); + assertTrue(subscription.getLiteTopicSet().contains(topic + "@" + group)); + assertEquals(1, registry.getActiveSubscriptionNum()); + } + + /** + * reexpandWildcardPatterns picks up lite-topics created after the initial subscription. + */ + @Test + public void testReexpandWildcardPatterns_PicksUpNewlyCreatedLmq() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "order_events"; + String payRefund = LiteUtil.toLmqName(topic, "pay__refund"); + String paySuccess = LiteUtil.toLmqName(topic, "pay__success"); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + // initial subscription: only pay__refund exists + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn(Collections.singletonList(payRefund)); + registry.addCompleteSubscription(clientId, group, topic, Collections.emptySet(), + new HashSet<>(Collections.singletonList("pay__*")), 1L); + assertEquals(1, registry.getActiveSubscriptionNum()); + + // a new matching lite-topic pay__success is created + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn(Arrays.asList(payRefund, paySuccess)); + int added = registry.reexpandWildcardPatterns(clientId); + + assertEquals(1, added); + LiteSubscription subscription = registry.getLiteSubscription(clientId); + assertTrue(subscription.getLiteTopicSet().contains(payRefund)); + assertTrue(subscription.getLiteTopicSet().contains(paySuccess)); + assertEquals(2, registry.getActiveSubscriptionNum()); + } + + /** + * A mixed wildcard group (one pattern-mode client + one legacy client) must deliver to BOTH: + * the pattern-mode client (via its real lmqName in liteTopic2Group) and the legacy client (via + * the synthetic topic@group key). getAllSubscriber must merge the two sets. + */ + @Test + public void testGetAllSubscriber_MixedWildcardGroup_DeliversToBothPatternAndLegacy() { + String group = "testGroup"; + String topic = "order_events"; + String payRefund = LiteUtil.toLmqName(topic, "pay__refund"); + String unmatched = LiteUtil.toLmqName(topic, "order__created"); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn(Arrays.asList(payRefund, unmatched)); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + // pattern-mode client subscribes pay__* + registry.addCompleteSubscription("patternClient", group, topic, Collections.emptySet(), + new HashSet<>(Collections.singletonList("pay__*")), 1L); + // legacy client subscribes (empty patterns -> synthetic key, receives all) + registry.addCompleteSubscription("legacyClient", group, topic, Collections.emptySet(), 1L); + + // For a topic the pattern client matched: both clients should be returned. + SubscriberWrapper result = registry.getAllSubscriber(group, payRefund); + assertNotNull(result); + assertInstanceOf(SubscriberWrapper.ListWrapper.class, result); + Set deliveredClientIds = new HashSet<>(); + for (ClientGroup cg : ((SubscriberWrapper.ListWrapper) result).getClients()) { + deliveredClientIds.add(cg.clientId); + } + assertTrue("pattern-mode client must be delivered", deliveredClientIds.contains("patternClient")); + assertTrue("legacy client must be delivered", deliveredClientIds.contains("legacyClient")); + + // For a topic the pattern client did NOT match: only the legacy client (receive-all). + SubscriberWrapper unmatchedResult = registry.getAllSubscriber(group, unmatched); + Set unmatchedIds = new HashSet<>(); + for (ClientGroup cg : ((SubscriberWrapper.ListWrapper) unmatchedResult).getClients()) { + unmatchedIds.add(cg.clientId); + } + assertFalse("pattern client must not receive unmatched topic", unmatchedIds.contains("patternClient")); + assertTrue("legacy client must still receive unmatched topic", unmatchedIds.contains("legacyClient")); + } + /** * Test addCompleteSubscription updates complete subscription */ diff --git a/broker/src/test/java/org/apache/rocketmq/broker/processor/LiteSubscriptionCtlProcessorTest.java b/broker/src/test/java/org/apache/rocketmq/broker/processor/LiteSubscriptionCtlProcessorTest.java index cc4692955fa..29be4bbfe7e 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/processor/LiteSubscriptionCtlProcessorTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/processor/LiteSubscriptionCtlProcessorTest.java @@ -159,6 +159,79 @@ public void testProcessRequest_ActionIsAllAdd() throws Exception { verify(liteSubscriptionRegistry).addCompleteSubscription(eq(clientId), eq(group), eq(topic), anySet(), eq(1L)); } + @Test + public void testProcessRequest_WildcardCompleteAdd_RoutesToPatternOverload() throws Exception { + String clientId = "clientId"; + String group = "group"; + String topic = "topic"; + Set patterns = new HashSet<>(); + patterns.add("pay__*"); + patterns.add("notify__**"); + + LiteSubscriptionDTO dto = new LiteSubscriptionDTO(); + dto.setClientId(clientId); + dto.setGroup(group); + dto.setTopic(topic); + dto.setLiteTopicSet(patterns); + dto.setAction(LiteSubscriptionAction.COMPLETE_ADD); + dto.setVersion(1L); + + LiteSubscriptionCtlRequestBody requestBody = new LiteSubscriptionCtlRequestBody(); + requestBody.setSubscriptionSet(Collections.singleton(dto)); + RemotingCommand request = RemotingCommand.createRequestCommand(0, null); + request.setBody(requestBody.encode()); + + when(ctx.channel()).thenReturn(channel); + when(brokerController.getSubscriptionGroupManager()).thenReturn(subscriptionGroupManager); + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setConsumeEnable(true); + groupConfig.setWildcardLiteGroup(true); + when(subscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + + RemotingCommand response = processor.processRequest(ctx, request); + + assertEquals(ResponseCode.SUCCESS, response.getCode()); + // wildcard group -> 6-arg overload carrying the validated patterns + verify(liteSubscriptionRegistry).addCompleteSubscription(eq(clientId), eq(group), eq(topic), + anySet(), anySet(), eq(1L)); + } + + @Test + public void testProcessRequest_WildcardCompleteAdd_AllInvalidPatternsRejected() throws Exception { + String clientId = "clientId"; + String group = "group"; + String topic = "topic"; + // every pattern is invalid (mixed * in a segment / ** not at end) + Set patterns = new HashSet<>(); + patterns.add("pay*"); + patterns.add("**__pay"); + + LiteSubscriptionDTO dto = new LiteSubscriptionDTO(); + dto.setClientId(clientId); + dto.setGroup(group); + dto.setTopic(topic); + dto.setLiteTopicSet(patterns); + dto.setAction(LiteSubscriptionAction.COMPLETE_ADD); + dto.setVersion(1L); + + LiteSubscriptionCtlRequestBody requestBody = new LiteSubscriptionCtlRequestBody(); + requestBody.setSubscriptionSet(Collections.singleton(dto)); + RemotingCommand request = RemotingCommand.createRequestCommand(0, null); + request.setBody(requestBody.encode()); + + when(ctx.channel()).thenReturn(channel); + when(brokerController.getSubscriptionGroupManager()).thenReturn(subscriptionGroupManager); + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setConsumeEnable(true); + groupConfig.setWildcardLiteGroup(true); + when(subscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + + RemotingCommand response = processor.processRequest(ctx, request); + + // Must be rejected, NOT silently widened to receive-all legacy wildcard mode. + assertEquals(ResponseCode.ILLEGAL_OPERATION, response.getCode()); + } + @Test public void testProcessRequest_ActionIsIncrementalRemove() throws Exception { String clientId = "clientId"; diff --git a/common/src/main/java/org/apache/rocketmq/common/lite/LitePatternMatcher.java b/common/src/main/java/org/apache/rocketmq/common/lite/LitePatternMatcher.java new file mode 100644 index 00000000000..fc672dd9a85 --- /dev/null +++ b/common/src/main/java/org/apache/rocketmq/common/lite/LitePatternMatcher.java @@ -0,0 +1,218 @@ +/* + * 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.common.lite; + +import java.util.Collection; +import java.util.HashSet; +import java.util.Set; +import org.apache.rocketmq.common.constant.LoggerName; +import org.apache.rocketmq.logging.org.slf4j.Logger; +import org.apache.rocketmq.logging.org.slf4j.LoggerFactory; + +/** + * NATS-style wildcard pattern matcher for LiteTopic consumer subscriptions, operating in + * child lite-topic name space (i.e. the {@code $liteTopic} segment of {@code %LMQ%$parent$liteTopic}). + * + *

Separator: {@code __} (double underscore). {@code $} is reserved as the LMQ separator + * (see {@link LiteUtil#SEPARATOR}) and {@code .} is used in topic names/paths, so {@code __} + * is used here to avoid conflicts. + * + *

Pattern tokens: + *

+ * + *

Examples (parent topic = {@code order_events}): + * 

+ *   pay__refund        matches pay__refund
+ *   pay__*             matches pay__refund, pay__success
+ *   *__refund          matches pay__refund, notify__refund
+ *   pay__*__notify     matches pay__refund__notify, pay__success__notify
+ *   **                 matches all
+ *   pay__**            matches pay__refund, pay__refund__notify
+ *
+ * + *

Matching is recursive segment-by-segment with no regex, so it is fast enough for the + * dispatch hot path. + */ +public final class LitePatternMatcher { + + private static final Logger LOGGER = LoggerFactory.getLogger(LoggerName.ROCKETMQ_POP_LITE_LOGGER_NAME); + + /** + * Segment separator for patterns and lite-topic names. Distinct from {@link LiteUtil#SEPARATOR} + * (which is the LMQ {@code $} separator and must not appear in lite-topic child names). + */ + public static final String SEPARATOR = "__"; + + private static final String SINGLE = "*"; + private static final String DOUBLE = "**"; + + private LitePatternMatcher() { + } + + /** + * Validate the syntax of a pattern. + * + *

Rules: + *

+ * + * @param pattern the pattern to validate + * @return true if the pattern is syntactically valid + */ + public static boolean validate(String pattern) { + if (pattern == null || pattern.isEmpty()) { + return false; + } + String[] segments = pattern.split(SEPARATOR, -1); + for (int i = 0; i < segments.length; i++) { + String segment = segments[i]; + if (segment.isEmpty()) { + return false; + } + if (DOUBLE.equals(segment)) { + // ** only allowed as the last segment + if (i != segments.length - 1) { + return false; + } + } else if (SINGLE.equals(segment)) { + // * is allowed anywhere + } else if (segment.indexOf('*') >= 0) { + // a segment mixing * with other chars is invalid + return false; + } + // otherwise a literal segment, always valid + } + return true; + } + + /** + * Test whether a lite-topic name matches a pattern. Both are split on {@code __}. + * + * @param pattern the pattern, should be {@link #validate(String)} valid + * @param liteTopic the candidate lite-topic child name + * @return true if the lite-topic matches the pattern + */ + public static boolean matches(String pattern, String liteTopic) { + if (!validate(pattern) || liteTopic == null || liteTopic.isEmpty()) { + return false; + } + String[] p = pattern.split(SEPARATOR, -1); + String[] t = liteTopic.split(SEPARATOR, -1); + return matchSeg(p, 0, t, 0); + } + + /** + * Recursive segment matcher. + * + *

Invariants enforced by {@link #validate(String)}: no empty segments, {@code **} only at + * the last index, {@code *} is a whole segment. + */ + private static boolean matchSeg(String[] p, int pi, String[] t, int ti) { + if (pi == p.length) { + // pattern exhausted: topic must also be exhausted + return ti == t.length; + } + String segment = p[pi]; + if (DOUBLE.equals(segment)) { + // ** matches one or more remaining segments to the end (validated to be last) + return ti < t.length; + } + if (SINGLE.equals(segment)) { + // * matches exactly one segment + return ti < t.length && matchSeg(p, pi + 1, t, ti + 1); + } + // literal segment + return ti < t.length && segment.equals(t[ti]) && matchSeg(p, pi + 1, t, ti + 1); + } + + /** + * Test whether a lite-topic name matches any of the given patterns. Invalid patterns are + * skipped. This is a convenience over {@link #matches(String, String)} for the common + * "does this candidate match any of the client's patterns" check on the dispatch hot path. + * + * @param patterns the patterns to test + * @param liteTopic the candidate lite-topic child name + * @return true if the lite-topic matches at least one valid pattern + */ + public static boolean matchesAny(Collection patterns, String liteTopic) { + if (patterns == null || patterns.isEmpty() || liteTopic == null || liteTopic.isEmpty()) { + return false; + } + for (String pattern : patterns) { + if (matches(pattern, liteTopic)) { + return true; + } + } + return false; + } + + /** + * Expand a set of patterns against a collection of candidate lite-topic names, returning the + * union of candidates matched by any valid pattern. Invalid patterns are skipped (with a debug + * log) rather than aborting the whole expansion, so one bad pattern does not kill a subscription. + * + *

Patterns and candidates are each pre-split once and reused across the inner loop to keep + * the cost under the {@code <10ms / 1000 candidates} target. + * + * @param patterns the patterns to expand + * @param candidates the candidate lite-topic child names + * @return the union of matched candidate names (never null) + */ + public static Set expand(Collection patterns, Collection candidates) { + Set matched = new HashSet<>(); + if (patterns == null || patterns.isEmpty() || candidates == null || candidates.isEmpty()) { + return matched; + } + // Pre-split valid patterns once + int count = 0; + String[][] compiledPatterns = new String[patterns.size()][]; + for (String pattern : patterns) { + if (!validate(pattern)) { + LOGGER.debug("skip invalid wildcard pattern during expand: {}", pattern); + continue; + } + compiledPatterns[count++] = pattern.split(SEPARATOR, -1); + } + if (count == 0) { + return matched; + } + // Pre-split each candidate once and test against all compiled patterns + for (String candidate : candidates) { + if (candidate == null || candidate.isEmpty()) { + continue; + } + String[] t = candidate.split(SEPARATOR, -1); + for (int i = 0; i < count; i++) { + if (matchSeg(compiledPatterns[i], 0, t, 0)) { + matched.add(candidate); + break; + } + } + } + return matched; + } +} diff --git a/common/src/main/java/org/apache/rocketmq/common/lite/LiteSubscription.java b/common/src/main/java/org/apache/rocketmq/common/lite/LiteSubscription.java index abf7c9ee3af..0a54bde848c 100644 --- a/common/src/main/java/org/apache/rocketmq/common/lite/LiteSubscription.java +++ b/common/src/main/java/org/apache/rocketmq/common/lite/LiteSubscription.java @@ -25,6 +25,13 @@ public class LiteSubscription { private String group; private String topic; private final Set liteTopicSet = ConcurrentHashMap.newKeySet(); + /** + * Wildcard patterns for a wildcard group in pattern mode. Empty means the group is either a + * non-wildcard subscription or a legacy wildcard group (receives all lite-topics under the + * parent topic). When non-empty, {@link #liteTopicSet} holds the expanded (matched) lmqNames. + * Not serialized over the wire; rebuilt on client reconnect, same as {@link #liteTopicSet}. + */ + private final Set wildcardPatterns = ConcurrentHashMap.newKeySet(); private volatile long updateTime = System.currentTimeMillis(); public boolean addLiteTopic(String liteTopic) { @@ -74,6 +81,23 @@ public LiteSubscription setLiteTopicSet(Set liteTopicSet) { return this; } + public Set getWildcardPatterns() { + return wildcardPatterns; + } + + /** + * Full replace of the wildcard patterns. Existing patterns not in the given collection are + * removed and the new ones are added. + */ + public LiteSubscription setWildcardPatterns(Collection patterns) { + updateTime(); + this.wildcardPatterns.clear(); + if (patterns != null) { + this.wildcardPatterns.addAll(patterns); + } + return this; + } + public long getUpdateTime() { return updateTime; } @@ -92,6 +116,7 @@ public String toString() { "group='" + group + '\'' + ", topic='" + topic + '\'' + ", liteTopicSet=" + liteTopicSet + + ", wildcardPatterns=" + wildcardPatterns + ", updateTime=" + updateTime + '}'; } diff --git a/common/src/test/java/org/apache/rocketmq/common/lite/LitePatternMatcherTest.java b/common/src/test/java/org/apache/rocketmq/common/lite/LitePatternMatcherTest.java new file mode 100644 index 00000000000..fe8443f25c9 --- /dev/null +++ b/common/src/test/java/org/apache/rocketmq/common/lite/LitePatternMatcherTest.java @@ -0,0 +1,198 @@ +/* + * 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.common.lite; + +import java.util.Arrays; +import java.util.Collections; +import java.util.HashSet; +import java.util.Set; +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +public class LitePatternMatcherTest { + + // ==================== validate ==================== + + @Test + public void validate_acceptsValidPatterns() { + assertTrue(LitePatternMatcher.validate("pay__refund")); + assertTrue(LitePatternMatcher.validate("pay__*")); + assertTrue(LitePatternMatcher.validate("*__refund")); + assertTrue(LitePatternMatcher.validate("pay__*__notify")); + assertTrue(LitePatternMatcher.validate("**")); + assertTrue(LitePatternMatcher.validate("pay__**")); + assertTrue(LitePatternMatcher.validate("*")); + assertTrue(LitePatternMatcher.validate("a__b__c")); + } + + @Test + public void validate_rejectsInvalidPatterns() { + assertFalse(LitePatternMatcher.validate(null)); + assertFalse(LitePatternMatcher.validate("")); + // empty segments + assertFalse(LitePatternMatcher.validate("pay__")); + assertFalse(LitePatternMatcher.validate("__pay")); + assertFalse(LitePatternMatcher.validate("pay____refund")); + // ** not at the end + assertFalse(LitePatternMatcher.validate("**__pay")); + assertFalse(LitePatternMatcher.validate("a__**__b")); + // * mixed with other chars in a segment + assertFalse(LitePatternMatcher.validate("pay*")); + assertFalse(LitePatternMatcher.validate("pa*y")); + } + + // ==================== matches — issue example rows ==================== + + @Test + public void matches_literalExact() { + assertTrue(LitePatternMatcher.matches("pay__refund", "pay__refund")); + assertFalse(LitePatternMatcher.matches("pay__refund", "pay__refund__notify")); + assertFalse(LitePatternMatcher.matches("pay__refund", "notify__refund")); + } + + @Test + public void matches_singleLevelWildcard() { + assertTrue(LitePatternMatcher.matches("pay__*", "pay__refund")); + assertTrue(LitePatternMatcher.matches("pay__*", "pay__success")); + assertFalse(LitePatternMatcher.matches("pay__*", "pay__refund__notify")); + } + + @Test + public void matches_singleLevelWildcardPrefix() { + assertTrue(LitePatternMatcher.matches("*__refund", "pay__refund")); + assertTrue(LitePatternMatcher.matches("*__refund", "notify__refund")); + assertFalse(LitePatternMatcher.matches("*__refund", "pay__refund__notify")); + } + + @Test + public void matches_wildcardInMiddle() { + assertTrue(LitePatternMatcher.matches("pay__*__notify", "pay__refund__notify")); + assertTrue(LitePatternMatcher.matches("pay__*__notify", "pay__success__notify")); + assertFalse(LitePatternMatcher.matches("pay__*__notify", "pay__refund")); + } + + @Test + public void matches_multiLevelAll() { + assertTrue(LitePatternMatcher.matches("**", "pay")); + assertTrue(LitePatternMatcher.matches("**", "pay__refund")); + assertTrue(LitePatternMatcher.matches("**", "pay__refund__notify")); + assertFalse(LitePatternMatcher.matches("**", "")); + } + + @Test + public void matches_multiLevelPrefix() { + assertTrue(LitePatternMatcher.matches("pay__**", "pay__refund")); + assertTrue(LitePatternMatcher.matches("pay__**", "pay__refund__notify")); + assertFalse(LitePatternMatcher.matches("pay__**", "notify__refund")); + // ** consumes one-or-more: pay__** must NOT match bare "pay" + assertFalse(LitePatternMatcher.matches("pay__**", "pay")); + } + + // ==================== matches — edge cases ==================== + + @Test + public void matches_invalidPatternNeverMatches() { + assertFalse(LitePatternMatcher.matches("pay__", "pay__refund")); + assertFalse(LitePatternMatcher.matches("**__pay", "pay")); + assertFalse(LitePatternMatcher.matches("pay*", "payx")); + } + + @Test + public void matches_nullAndEmpty() { + assertFalse(LitePatternMatcher.matches(null, "pay")); + assertFalse(LitePatternMatcher.matches("pay", null)); + assertFalse(LitePatternMatcher.matches("pay", "")); + } + + @Test + public void matches_singleSegmentLiteralAndWildcard() { + assertTrue(LitePatternMatcher.matches("pay", "pay")); + assertFalse(LitePatternMatcher.matches("pay", "refund")); + assertTrue(LitePatternMatcher.matches("*", "pay")); + assertFalse(LitePatternMatcher.matches("*", "pay__refund")); + } + + // ==================== matchesAny ==================== + + @Test + public void matchesAny_unionSemantics() { + Set patterns = new HashSet<>(Arrays.asList("pay__*", "notify__**")); + assertTrue(LitePatternMatcher.matchesAny(patterns, "pay__refund")); + assertTrue(LitePatternMatcher.matchesAny(patterns, "notify__refund__sms")); + assertFalse(LitePatternMatcher.matchesAny(patterns, "order__created")); + } + + @Test + public void matchesAny_emptyAndNull() { + assertFalse(LitePatternMatcher.matchesAny(null, "pay")); + assertFalse(LitePatternMatcher.matchesAny(Collections.emptySet(), "pay")); + assertFalse(LitePatternMatcher.matchesAny(Collections.singleton("pay__*"), null)); + assertFalse(LitePatternMatcher.matchesAny(Collections.singleton("pay__*"), "")); + } + + @Test + public void matchesAny_skipsInvalidPattern() { + // an invalid pattern among valid ones must not break matching + Set patterns = new HashSet<>(Arrays.asList("pay__", "pay__*")); + assertTrue(LitePatternMatcher.matchesAny(patterns, "pay__refund")); + assertFalse(LitePatternMatcher.matchesAny(patterns, "order__x")); + } + + // ==================== expand ==================== + + @Test + public void expand_unionOfMatched() { + Set patterns = new HashSet<>(Arrays.asList("pay__*", "notify__**")); + Set candidates = new HashSet<>(Arrays.asList( + "pay__refund", "pay__success", "pay__refund__notify", + "notify__refund", "notify__refund__sms", "order__created")); + Set matched = LitePatternMatcher.expand(patterns, candidates); + // pay__* -> pay__refund, pay__success (NOT pay__refund__notify) + // notify__** -> notify__refund, notify__refund__sms + Set expected = new HashSet<>(Arrays.asList( + "pay__refund", "pay__success", "notify__refund", "notify__refund__sms")); + assertEquals(expected, matched); + } + + @Test + public void expand_emptyInputs() { + assertEquals(Collections.emptySet(), + LitePatternMatcher.expand(Collections.emptySet(), Collections.singleton("pay__refund"))); + assertEquals(Collections.emptySet(), + LitePatternMatcher.expand(Collections.singleton("pay__*"), Collections.emptySet())); + assertEquals(Collections.emptySet(), LitePatternMatcher.expand(null, null)); + } + + @Test + public void expand_skipsInvalidPattern() { + Set patterns = new HashSet<>(Arrays.asList("pay__", "pay__*")); + Set candidates = new HashSet<>(Arrays.asList("pay__refund", "order__x")); + Set matched = LitePatternMatcher.expand(patterns, candidates); + assertEquals(Collections.singleton("pay__refund"), matched); + } + + @Test + public void expand_doubleStarMatchesAll() { + Set candidates = new HashSet<>(Arrays.asList("a", "a__b", "a__b__c")); + Set matched = LitePatternMatcher.expand(Collections.singleton("**"), candidates); + assertEquals(candidates, matched); + } +} From a26ffbb552eea3def59836c49844b16b16d5891a Mon Sep 17 00:00:00 2001 From: liuhy Date: Fri, 10 Jul 2026 18:04:45 +0800 Subject: [PATCH 2/4] [ISSUE #10607] Fix wildcard pattern dispatch gaps for new lite-topics MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Address review findings on the wildcard pattern matching feature (#10559): 1. Pattern→legacy/non-wildcard transition left stale wildcardPatterns on the LiteSubscription, so doFullDispatchForWildcardGroup / reexpandWildcardPatterns (which key off a non-empty wildcardPatterns set) kept misclassifying the client as pattern-mode. Now clear wildcardPatterns in both the legacy-wildcard and non-wildcard branches of addCompleteSubscription. 2. A pattern-mode client was not delivered a newly-created matching lite-topic on the normal message-arriving dispatch path — it only got picked up by the periodic doFullDispatchForWildcardGroup re-expand, causing delivery delay/loss. Add LiteSubscriptionRegistry#registerArrivingLmqForPatternClients(lmqName): on the message-arriving path (group == null) it enumerates the parent topic's pattern-mode clients, matches their stored patterns against the single arriving lmqName's child (no O(M) collectByParentTopic scan), and registers matches via addTopicGroup so the immediately-following getAllSubscriber finds them. The periodic re-expand remains as a backstop. 3. Nit: wildcard COMPLETE_ADD now passes Collections.emptySet() for lmqNameSet (pattern-mode registration ignores it; lmqNames are derived by expanding patterns). CI note: bazel-compile BrokerShutdownTest TIMEOUT and macOS ServiceThreadTest stress flake are pre-existing and unrelated (neither test is in this diff nor touches lite/wildcard code). Co-Authored-By: Claude --- .../broker/lite/LiteEventDispatcher.java | 7 + .../broker/lite/LiteSubscriptionRegistry.java | 17 ++ .../lite/LiteSubscriptionRegistryImpl.java | 58 ++++++- .../LiteSubscriptionCtlProcessor.java | 7 +- .../broker/lite/LiteEventDispatcherTest.java | 27 ++++ .../LiteSubscriptionRegistryImplTest.java | 150 ++++++++++++++++++ 6 files changed, 263 insertions(+), 3 deletions(-) diff --git a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java index afa39ba8dad..a6858a143eb 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java @@ -94,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); } 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 2baf840da6b..6f20d74a831 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 @@ -69,6 +69,23 @@ void addCompleteSubscription(String clientId, String group, String topic, SetOnly 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); 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 d61ef4bf3ad..08b2ff93347 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 @@ -146,10 +146,17 @@ public void addCompleteSubscription(String clientId, String group, String topic, markWildcardGroup(topic, group); } else if (isWildcardGroup) { // Legacy wildcard group: receive all lite-topics under the parent topic via the - // synthetic topic@group key. + // synthetic topic@group key. Clear any previously stored patterns so a client that + // transitions from pattern mode back to legacy mode is no longer classified as + // pattern-mode (doFullDispatchForWildcardGroup / reexpandWildcardPatterns key off a + // non-empty wildcardPatterns set). + thisSub.setWildcardPatterns(Collections.emptySet()); lmqNameNew = Collections.singleton(mockLmqNameForWildcardGroup(topic, group)); markWildcardGroup(topic, group); } else { + // Non-wildcard group: a normal subscription. Clear stale patterns from any prior + // wildcard incarnation so the group is not misclassified downstream. + thisSub.setWildcardPatterns(Collections.emptySet()); lmqNameNew = lmqNameAll.stream() .filter(lmqName -> liteLifecycleManager.isSubscriptionActive(topic, lmqName)) .collect(Collectors.toSet()); @@ -241,6 +248,55 @@ public int reexpandWildcardPatterns(String clientId) { return added; } + /** + * Single-lmqName counterpart to {@link #reexpandWildcardPatterns(String)} for the + * message-arriving dispatch path. When a message arrives on a lite-topic that no pattern-mode + * client has been registered for yet (e.g. a newly-created lite-topic), enumerate the parent + * topic's pattern-mode wildcard clients and register the ones whose stored patterns match this + * lmqName's child. This avoids the O(M) {@code collectByParentTopic} scan that + * {@code reexpandWildcardPatterns} performs, matching only against the single arriving lmqName. + * + *

Legacy wildcard clients are NOT touched here: they are already reachable via the synthetic + * {@code topic@group} key in {@link #getAllSubscriber}. The periodic + * {@code doFullDispatchForWildcardGroup} re-expand remains as a backstop. + */ + @Override + public int registerArrivingLmqForPatternClients(String lmqName) { + String parentTopic = LiteUtil.getParentTopic(lmqName); + String child = LiteUtil.getLiteTopic(lmqName); + if (parentTopic == null || child == null) { + return 0; + } + Set wildcardGroups = wildcardGroupMap.get(parentTopic); + if (wildcardGroups == null || wildcardGroups.isEmpty()) { + return 0; + } + int added = 0; + for (String group : wildcardGroups) { + for (String clientId : getAllClientIdByGroup(group)) { + LiteSubscription thisSub = client2Subscription.get(clientId); + if (thisSub == null || CollectionUtils.isEmpty(thisSub.getWildcardPatterns())) { + continue; // legacy or absent client + } + if (!LitePatternMatcher.matchesAny(thisSub.getWildcardPatterns(), child)) { + continue; + } + if (!liteLifecycleManager.isSubscriptionActive(parentTopic, lmqName)) { + continue; // mirror expandWildcardPatterns sharding guard + } + thisSub.addLiteTopic(lmqName); + if (addTopicGroup(new ClientGroup(clientId, thisSub.getGroup()), lmqName)) { + added++; + } + } + } + if (added > 0) { + LOGGER.info("registerArrivingLmqForPatternClients, topic:{}, lmqName:{}, newly registered:{}", + parentTopic, lmqName, added); + } + return added; + } + @Override public void removeCompleteSubscription(String clientId) { clientChannels.remove(clientId); diff --git a/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteSubscriptionCtlProcessor.java b/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteSubscriptionCtlProcessor.java index d13dc061d70..cfe1dc1d056 100644 --- a/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteSubscriptionCtlProcessor.java +++ b/broker/src/main/java/org/apache/rocketmq/broker/processor/LiteSubscriptionCtlProcessor.java @@ -98,9 +98,12 @@ public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand if (LiteMetadataUtil.isWildcardGroup(group, brokerController)) { // For a wildcard group, liteTopicSet carries pattern strings (not literal // lmqNames); pass them through verbatim and let the registry expand them. + // lmqNameSet is intentionally empty: pattern-mode registration ignores it + // (lmqNames are derived by expanding the patterns), and passing a set of + // pattern-derived pseudo-lmqNames here would be misleading. Set wildcardPatterns = toWildcardPatterns(entry); - this.liteSubscriptionRegistry.addCompleteSubscription(clientId, group, topic, lmqNameSet, - wildcardPatterns, entry.getVersion()); + this.liteSubscriptionRegistry.addCompleteSubscription(clientId, group, topic, + Collections.emptySet(), wildcardPatterns, entry.getVersion()); } else { this.liteSubscriptionRegistry.addCompleteSubscription(clientId, group, topic, lmqNameSet, entry.getVersion()); diff --git a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java index 3c90e85733f..5ae34cf1ca4 100644 --- a/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java +++ b/broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java @@ -119,6 +119,33 @@ public void testDispatchCallsDoDispatch() { verify(spyDispatcher).doDispatch("group", lmqName, null); } + /** + * On the message-arriving path (group == null), dispatch lazily registers the arriving lmqName + * against pattern-mode wildcard clients before fan-out, so a newly-created matching lite-topic + * is delivered without waiting for the periodic full dispatch. + */ + @Test + public void testDispatch_MessageArrival_RegistersArrivingLmqForPatternClients() { + String lmqName = LiteUtil.toLmqName("parentTopic", "child"); + LiteEventDispatcher spyDispatcher = Mockito.spy(liteEventDispatcher); + spyDispatcher.dispatch(null, lmqName, 0, 0L, 0L); + verify(liteSubscriptionRegistry).registerArrivingLmqForPatternClients(lmqName); + verify(spyDispatcher).doDispatch(null, lmqName, null); + } + + /** + * On a group-specific dispatch (e.g. ack re-dispatch), the pattern re-expand must NOT run — + * registration state is already settled for an explicit group. + */ + @Test + public void testDispatch_GroupSpecified_DoesNotRegisterArrivingLmq() { + String lmqName = LiteUtil.toLmqName("parentTopic", "child"); + LiteEventDispatcher spyDispatcher = Mockito.spy(liteEventDispatcher); + spyDispatcher.dispatch("someGroup", lmqName, 0, 0L, 0L); + verify(liteSubscriptionRegistry, never()).registerArrivingLmqForPatternClients(anyString()); + verify(spyDispatcher).doDispatch("someGroup", lmqName, null); + } + @Test public void testDoDispatchWhenWrapperIsNull() { when(liteSubscriptionRegistry.getAllSubscriber("group", "lmqName")).thenReturn(null); 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 026a65bc385..91698161e1a 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 @@ -413,6 +413,40 @@ public void testAddCompleteSubscription_PatternWildcardGroup_DiffOnResubscribe() assertEquals(1, registry.getActiveSubscriptionNum()); } + /** + * Re-subscribing from pattern mode back to legacy mode (empty patterns) must clear the stored + * patterns. Otherwise the client is misclassified as pattern-mode by + * doFullDispatchForWildcardGroup / reexpandWildcardPatterns, which key off a non-empty + * wildcardPatterns set, and legacy receive-all behavior is broken. + */ + @Test + public void testAddCompleteSubscription_PatternToLegacyTransition_ClearsPatterns() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "order_events"; + String payRefund = LiteUtil.toLmqName(topic, "pay__refund"); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn(Collections.singletonList(payRefund)); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + // first subscribe in pattern mode + registry.addCompleteSubscription(clientId, group, topic, Collections.emptySet(), + new HashSet<>(Collections.singletonList("pay__*")), 1L); + LiteSubscription subscription = registry.getLiteSubscription(clientId); + assertFalse(subscription.getWildcardPatterns().isEmpty()); + assertTrue(subscription.getLiteTopicSet().contains(payRefund)); + + // re-subscribe in legacy mode (empty patterns): must drop patterns and revert to the + // synthetic topic@group key. + registry.addCompleteSubscription(clientId, group, topic, + new HashSet<>(Collections.singletonList(payRefund)), Collections.emptySet(), 2L); + subscription = registry.getLiteSubscription(clientId); + assertTrue(subscription.getWildcardPatterns().isEmpty()); + assertFalse(subscription.getLiteTopicSet().contains(payRefund)); + assertTrue(subscription.getLiteTopicSet().contains(topic + "@" + group)); + assertEquals(1, registry.getActiveSubscriptionNum()); + } + /** * getAllSubscriber for a pattern-mode wildcard group uses the normal liteTopic2Group path. */ @@ -516,6 +550,122 @@ public void testReexpandWildcardPatterns_PicksUpNewlyCreatedLmq() { assertEquals(2, registry.getActiveSubscriptionNum()); } + /** + * registerArrivingLmqForPatternClients is the dispatch-time single-lmqName counterpart to + * reexpandWildcardPatterns. When a message arrives on a newly-created lite-topic that a + * pattern-mode client matches but was not expanded to at subscribe time, it registers the + * (client, lmqName) pair so the dispatch can reach the client immediately — without the O(M) + * collectByParentTopic scan. + */ + @Test + public void testRegisterArrivingLmqForPatternClients_RegistersNewlyCreatedMatchingLmq() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "order_events"; + String payRefund = LiteUtil.toLmqName(topic, "pay__refund"); + String paySuccess = LiteUtil.toLmqName(topic, "pay__success"); // created after subscribe + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn(Collections.singletonList(payRefund)); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + // initial subscription: only pay__refund exists at expand time + registry.addCompleteSubscription(clientId, group, topic, Collections.emptySet(), + new HashSet<>(Collections.singletonList("pay__*")), 1L); + assertEquals(1, registry.getActiveSubscriptionNum()); + + // a new matching lite-topic pay__success arrives — single-lmqName match, no re-scan + int added = registry.registerArrivingLmqForPatternClients(paySuccess); + + assertEquals(1, added); + LiteSubscription subscription = registry.getLiteSubscription(clientId); + assertTrue(subscription.getLiteTopicSet().contains(payRefund)); + assertTrue(subscription.getLiteTopicSet().contains(paySuccess)); + assertEquals(2, registry.getActiveSubscriptionNum()); + // the new lmqName is now reachable via the normal message-arriving (group == null) path + SubscriberWrapper result = registry.getAllSubscriber(null, paySuccess); + assertNotNull(result); + assertInstanceOf(SubscriberWrapper.MapWrapper.class, result); + assertTrue(((SubscriberWrapper.MapWrapper) result).getGroupMap().containsKey(group)); + } + + /** + * registerArrivingLmqForPatternClients does not register a lmqName whose child does not match + * any of the pattern client's patterns. + */ + @Test + public void testRegisterArrivingLmqForPatternClients_SkipsNonMatchingLmq() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "order_events"; + String payRefund = LiteUtil.toLmqName(topic, "pay__refund"); + String orderCreated = LiteUtil.toLmqName(topic, "order__created"); // does not match pay__* + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn(Collections.singletonList(payRefund)); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + registry.addCompleteSubscription(clientId, group, topic, Collections.emptySet(), + new HashSet<>(Collections.singletonList("pay__*")), 1L); + + int added = registry.registerArrivingLmqForPatternClients(orderCreated); + + assertEquals(0, added); + assertFalse(registry.getLiteSubscription(clientId).getLiteTopicSet().contains(orderCreated)); + assertEquals(1, registry.getActiveSubscriptionNum()); + } + + /** + * registerArrivingLmqForPatternClients is idempotent: re-arrival on an lmqName already + * registered at subscribe time registers nothing new. + */ + @Test + public void testRegisterArrivingLmqForPatternClients_IdempotentOnRepeatArrival() { + String clientId = "testClient"; + String group = "testGroup"; + String topic = "order_events"; + String payRefund = LiteUtil.toLmqName(topic, "pay__refund"); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn(Collections.singletonList(payRefund)); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + registry.addCompleteSubscription(clientId, group, topic, Collections.emptySet(), + new HashSet<>(Collections.singletonList("pay__*")), 1L); + + int added = registry.registerArrivingLmqForPatternClients(payRefund); + assertEquals(0, added); // already registered at subscribe time + assertEquals(1, registry.getActiveSubscriptionNum()); + } + + /** + * registerArrivingLmqForPatternClients ignores legacy wildcard clients (empty patterns); they + * are reachable via the synthetic topic@group key, not the pattern path, and must not be + * double-registered. + */ + @Test + public void testRegisterArrivingLmqForPatternClients_IgnoresLegacyClients() { + String legacyClientId = "legacyClient"; + String group = "testGroup"; + String topic = "order_events"; + String payRefund = LiteUtil.toLmqName(topic, "pay__refund"); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + // legacy wildcard client (empty patterns -> synthetic key, receives all) + registry.addCompleteSubscription(legacyClientId, group, topic, Collections.emptySet(), 1L); + + int added = registry.registerArrivingLmqForPatternClients(payRefund); + assertEquals(0, added); // legacy client has empty wildcardPatterns, skipped + // legacy client still reachable via the synthetic key, unaffected + assertEquals(1, registry.getActiveSubscriptionNum()); + SubscriberWrapper result = registry.getAllSubscriber(group, payRefund); + assertNotNull(result); + assertInstanceOf(SubscriberWrapper.ListWrapper.class, result); + Set deliveredClientIds = new HashSet<>(); + for (ClientGroup cg : ((SubscriberWrapper.ListWrapper) result).getClients()) { + deliveredClientIds.add(cg.clientId); + } + assertTrue(deliveredClientIds.contains(legacyClientId)); + } + /** * A mixed wildcard group (one pattern-mode client + one legacy client) must deliver to BOTH: * the pattern-mode client (via its real lmqName in liteTopic2Group) and the legacy client (via From 0e92110ee4fb8e213f7b7d3769a8778eb0de02ed Mon Sep 17 00:00:00 2001 From: liuhy Date: Fri, 10 Jul 2026 21:53:22 +0800 Subject: [PATCH 3/4] [ISSUE #10607] Enforce lite subscription quota on wildcard pattern expansion paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The maxLiteSubscriptionCount quota gate existed only in addPartialSubscription. The wildcard pattern paths added in #10559 — addCompleteSubscription (pattern-mode and non-wildcard), the periodic reexpandWildcardPatterns, and the message-arriving registerArrivingLmqForPatternClients — all call addTopicGroup and increment activeNum with no quota check, so a wide pattern could expand subscription references without bound at runtime, bypassing the broker quota. Close all three paths with context-appropriate enforcement: - addCompleteSubscription: pre-flight throw (LiteQuotaException, already mapped to LITE_SUBSCRIPTION_QUOTA_EXCEEDED by the processor's existing catch) projecting net-new references against the limit, before any mutation. - reexpandWildcardPatterns / registerArrivingLmqForPatternClients: cap and log (no throw — these run on the dispatcher background thread / message-arriving hot path with no client to respond to); register up to the limit, then stop. The guard precedes addLiteTopic so the subscription's LiteTopicSet stays consistent with liteTopic2Group on early exit. Added 4 tests covering both throw paths and both cap paths. Co-Authored-By: Claude --- .../lite/LiteSubscriptionRegistryImpl.java | 55 ++++++++ .../LiteSubscriptionRegistryImplTest.java | 132 ++++++++++++++++++ 2 files changed, 187 insertions(+) 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 08b2ff93347..f4c7d41b7e2 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 @@ -163,6 +163,17 @@ public void addCompleteSubscription(String clientId, String group, String topic, } Set lmqNamePrev = thisSub.getLiteTopicSet(); + // Pre-flight quota check: project the net-new (client, lmqName) references this operation + // would register (lmqNames in the new set not already held) against the broker-wide limit. + // Like addPartialSubscription, the budget is checked before any mutation; pending removals + // are NOT credited, so a set-replacing client at the limit may be denied (conservative). A + // re-sync of an identical set has wouldAdd == 0 and passes trivially. Throwing here maps to + // LITE_SUBSCRIPTION_QUOTA_EXCEEDED via the processor's existing catch. + int wouldAdd = (int) lmqNameNew.stream() + .filter(lmqName -> !lmqNamePrev.contains(lmqName)) + .count(); + checkQuotaOrThrow(wouldAdd); + // Find topics to remove (in current set but not in new set) Set lmqNameRemove = lmqNamePrev.stream() .filter(lmqName -> !lmqNameNew.contains(lmqName)) @@ -194,6 +205,22 @@ public void addCompleteSubscription(String clientId, String group, String topic, } } + /** + * Pre-flight quota check for the request/response registration paths. Throws + * {@link LiteQuotaException} (mapped to {@code LITE_SUBSCRIPTION_QUOTA_EXCEEDED} by the + * processor) if the projected active count would exceed the configured broker-wide limit. + * + * @param wouldAdd the number of net-new (client, lmqName) pairs this operation would register; + * 0 for a pure "is the broker already at the limit" check + */ + private void checkQuotaOrThrow(int wouldAdd) { + long maxCount = brokerController.getBrokerConfig().getMaxLiteSubscriptionCount(); + if ((long) getActiveSubscriptionNum() + wouldAdd > maxCount) { + throw new LiteQuotaException("lite subscription quota exceeded " + maxCount + + ", current: " + getActiveSubscriptionNum() + ", would add: " + wouldAdd); + } + } + /** * Eagerly expand wildcard patterns against the existing lite-topics under {@code parentTopic}. * Returns the set of matched full lmqNames that this broker is responsible for @@ -235,7 +262,18 @@ public int reexpandWildcardPatterns(String clientId) { } ClientGroup clientGroup = new ClientGroup(clientId, group); int added = 0; + // Cap-and-log, do not throw: this runs on the LiteEventDispatcher background thread with no + // client to respond to. Once the broker-wide quota is reached, stop auto-registering new + // matches (existing subscriptions keep serving); the periodic re-expand retries next cycle + // and picks up registration once room frees up. The guard precedes addLiteTopic so the + // subscription's LiteTopicSet stays consistent with liteTopic2Group on early exit. + long maxCount = brokerController.getBrokerConfig().getMaxLiteSubscriptionCount(); for (String lmqName : toAdd) { + if (getActiveSubscriptionNum() >= maxCount) { + LOGGER.warn("reexpandWildcardPatterns capped at lite subscription quota {}, clientId:{}, group:{}, " + + "topic:{}, added:{}, skipped:{}", maxCount, clientId, group, topic, added, toAdd.size() - added); + break; + } thisSub.addLiteTopic(lmqName); if (addTopicGroup(clientGroup, lmqName)) { added++; @@ -272,7 +310,19 @@ public int registerArrivingLmqForPatternClients(String lmqName) { return 0; } int added = 0; + // Cap-and-log, do not throw: this runs on the message-arriving dispatch hot path with no + // client to respond to (an exception here would abort per-message dispatch). Once the + // broker-wide quota is reached, stop auto-registering the arriving lmqName; the periodic + // doFullDispatchForWildcardGroup re-expand retries and picks it up once room frees up. The + // guard precedes addLiteTopic so the subscription's LiteTopicSet stays consistent with + // liteTopic2Group on early exit. + long maxCount = brokerController.getBrokerConfig().getMaxLiteSubscriptionCount(); for (String group : wildcardGroups) { + if (getActiveSubscriptionNum() >= maxCount) { + LOGGER.warn("registerArrivingLmqForPatternClients capped at lite subscription quota {}, " + + "topic:{}, lmqName:{}, added:{}", maxCount, parentTopic, lmqName, added); + break; + } for (String clientId : getAllClientIdByGroup(group)) { LiteSubscription thisSub = client2Subscription.get(clientId); if (thisSub == null || CollectionUtils.isEmpty(thisSub.getWildcardPatterns())) { @@ -284,6 +334,11 @@ public int registerArrivingLmqForPatternClients(String lmqName) { if (!liteLifecycleManager.isSubscriptionActive(parentTopic, lmqName)) { continue; // mirror expandWildcardPatterns sharding guard } + if (getActiveSubscriptionNum() >= maxCount) { + LOGGER.warn("registerArrivingLmqForPatternClients capped at lite subscription quota {}, " + + "topic:{}, lmqName:{}, added:{}", maxCount, parentTopic, lmqName, added); + break; + } thisSub.addLiteTopic(lmqName); if (addTopicGroup(new ClientGroup(clientId, thisSub.getGroup()), lmqName)) { added++; 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 91698161e1a..175060d5a73 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 @@ -520,6 +520,138 @@ public void testAddCompleteSubscription_LegacyWildcardGroup_StillUsesSyntheticKe assertEquals(1, registry.getActiveSubscriptionNum()); } + // ==================== Subscription Quota on All Registration Paths ==================== + + /** + * addCompleteSubscription (non-wildcard group) must respect maxLiteSubscriptionCount: once the + * broker is at the limit, a second client's complete-add is rejected with LiteQuotaException + * (mapped to LITE_SUBSCRIPTION_QUOTA_EXCEEDED by the processor) and mutates nothing. + */ + @Test + public void testAddCompleteSubscription_NonWildcard_QuotaExceeded() { + when(mockBrokerConfig.getMaxLiteSubscriptionCount()).thenReturn(1L); + String group = "testGroup"; + String topic = "testTopic"; + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + String clientA = "clientA"; + String lmq1 = "lmq1"; + registry.addCompleteSubscription(clientA, group, topic, new HashSet<>(Collections.singletonList(lmq1)), 1L); + assertEquals(1, registry.getActiveSubscriptionNum()); + assertTrue(registry.getLiteSubscription(clientA).getLiteTopicSet().contains(lmq1)); + + // clientB's complete-add would push activeNum to 2 (> maxCount=1) -> rejected, no mutation. + // (addCompleteSubscription creates a placeholder LiteSubscription for clientB before the + // pre-flight check fires, but registers no lmqNames and never increments activeNum.) + String clientB = "clientB"; + String lmq2 = "lmq2"; + assertThrows(LiteQuotaException.class, () -> + registry.addCompleteSubscription(clientB, group, topic, new HashSet<>(Collections.singletonList(lmq2)), 1L)); + assertEquals(1, registry.getActiveSubscriptionNum()); + assertTrue(registry.getLiteSubscription(clientB).getLiteTopicSet().isEmpty()); + // clientA's subscription is intact + assertTrue(registry.getLiteSubscription(clientA).getLiteTopicSet().contains(lmq1)); + } + + /** + * addCompleteSubscription (pattern-mode wildcard group) must respect maxLiteSubscriptionCount: + * when eager pattern expansion would register more lmqNames than the quota allows, the whole + * operation is rejected up front (pre-mutation) with LiteQuotaException and registers nothing. + */ + @Test + public void testAddCompleteSubscription_PatternWildcard_QuotaExceeded() { + when(mockBrokerConfig.getMaxLiteSubscriptionCount()).thenReturn(1L); + String clientId = "testClient"; + String group = "testGroup"; + String topic = "order_events"; + String payRefund = LiteUtil.toLmqName(topic, "pay__refund"); + String paySuccess = LiteUtil.toLmqName(topic, "pay__success"); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn(Arrays.asList(payRefund, paySuccess)); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + Set patterns = new HashSet<>(Collections.singletonList("pay__*")); + // Both pay__refund and pay__success match -> wouldAdd=2 > maxCount=1 -> rejected pre-mutation. + assertThrows(LiteQuotaException.class, () -> + registry.addCompleteSubscription(clientId, group, topic, Collections.emptySet(), patterns, 1L)); + + assertEquals(0, registry.getActiveSubscriptionNum()); + LiteSubscription subscription = registry.getLiteSubscription(clientId); + assertNotNull(subscription); + assertTrue(subscription.getLiteTopicSet().isEmpty()); + } + + /** + * reexpandWildcardPatterns caps at the quota instead of throwing: it registers newly-matched + * lmqNames up to the limit, then stops and logs. Runs on the dispatcher background thread with + * no client to respond to, so it must not throw. The cap guard precedes addLiteTopic so the + * subscription's LiteTopicSet stays consistent with liteTopic2Group on early exit. + */ + @Test + public void testReexpandWildcardPatterns_CapsAtQuota() { + when(mockBrokerConfig.getMaxLiteSubscriptionCount()).thenReturn(2L); + String clientId = "testClient"; + String group = "testGroup"; + String topic = "order_events"; + String payRefund = LiteUtil.toLmqName(topic, "pay__refund"); + String paySuccess = LiteUtil.toLmqName(topic, "pay__success"); + String payCancelled = LiteUtil.toLmqName(topic, "pay__cancelled"); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + // initial subscription: only pay__refund exists -> activeNum=1 (room for 1 more) + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn(Collections.singletonList(payRefund)); + registry.addCompleteSubscription(clientId, group, topic, Collections.emptySet(), + new HashSet<>(Collections.singletonList("pay__*")), 1L); + assertEquals(1, registry.getActiveSubscriptionNum()); + + // two new matching lite-topics now exist; re-expand has room for only one more + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn( + Arrays.asList(payRefund, paySuccess, payCancelled)); + int added = registry.reexpandWildcardPatterns(clientId); + + assertEquals(1, added); // added paySuccess, then capped before payCancelled + assertEquals(2, registry.getActiveSubscriptionNum()); + LiteSubscription subscription = registry.getLiteSubscription(clientId); + assertTrue(subscription.getLiteTopicSet().contains(payRefund)); + assertTrue(subscription.getLiteTopicSet().contains(paySuccess)); + assertFalse(subscription.getLiteTopicSet().contains(payCancelled)); + } + + /** + * registerArrivingLmqForPatternClients caps at the quota instead of throwing: when the broker is + * already at the limit on the message-arriving hot path, it registers nothing and logs rather than + * throwing (an exception would abort per-message dispatch). Existing subscriptions are unaffected. + */ + @Test + public void testRegisterArrivingLmqForPatternClients_CapsAtQuota() { + when(mockBrokerConfig.getMaxLiteSubscriptionCount()).thenReturn(2L); + String group = "testGroup"; + String topic = "order_events"; + String payRefund = LiteUtil.toLmqName(topic, "pay__refund"); + String paySuccess = LiteUtil.toLmqName(topic, "pay__success"); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + // two pattern clients subscribed when only pay__refund exists -> activeNum=2 (at limit) + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn(Collections.singletonList(payRefund)); + Set patterns = new HashSet<>(Collections.singletonList("pay__*")); + registry.addCompleteSubscription("clientA", group, topic, Collections.emptySet(), patterns, 1L); + registry.addCompleteSubscription("clientB", group, topic, Collections.emptySet(), patterns, 1L); + assertEquals(2, registry.getActiveSubscriptionNum()); + + // a new matching lite-topic arrives while at the limit -> nothing registered, no throw + int added = registry.registerArrivingLmqForPatternClients(paySuccess); + + assertEquals(0, added); + assertEquals(2, registry.getActiveSubscriptionNum()); + assertFalse(registry.getLiteSubscription("clientA").getLiteTopicSet().contains(paySuccess)); + assertFalse(registry.getLiteSubscription("clientB").getLiteTopicSet().contains(paySuccess)); + } + /** * reexpandWildcardPatterns picks up lite-topics created after the initial subscription. */ From 2c5976415c657a8d97ddf25e379256e897be5778 Mon Sep 17 00:00:00 2001 From: liuhy Date: Fri, 10 Jul 2026 22:37:18 +0800 Subject: [PATCH 4/4] [ISSUE #10607] Move quota projection before mutation and credit removals MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two gaps in addCompleteSubscription's quota gate (follow-up #3): 1. State leak on quota failure. The check ran AFTER setWildcardPatterns and markWildcardGroup, so a quota-exceeded throw left non-empty wildcardPatterns plus a wildcardGroupMap entry. reexpandWildcardPatterns and registerArrivingLmqForPatternClients (which key off non-empty patterns / the group map) then picked up the failed client and turned it into a deferred-effect subscription. 2. wouldAdd only counted gross additions, not crediting the lmqNames the same operation would remove. A net-zero replace at the quota limit (drop one lmq, add another) was falsely rejected. Fix: compute lmqNameNew (expand is read-only) BEFORE any mutation, project wouldAdd as max(0, added - removed), run checkQuotaOrThrow first, then apply setWildcardPatterns / markWildcardGroup / add-remove. A quota failure now leaves only the harmless empty placeholder LiteSubscription — no patterns, no group mark, no active refs. Tests: strengthened the pattern quota test to assert no pattern/group state leaks; added net-zero-replace-at-quota (must pass) and a pattern-quota-failure deferred-state test asserting later re-expand / arriving-lmq register nothing for the rejected client. Co-Authored-By: Claude --- .../lite/LiteSubscriptionRegistryImpl.java | 61 +++++++++----- .../LiteSubscriptionRegistryImplTest.java | 80 +++++++++++++++++++ 2 files changed, 119 insertions(+), 22 deletions(-) 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 f4c7d41b7e2..0fbb1ebb2eb 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 @@ -136,13 +136,45 @@ public void addCompleteSubscription(String clientId, String group, String topic, boolean isWildcardGroup = LiteMetadataUtil.isWildcardGroup(group, brokerController); boolean isPatternMode = isWildcardGroup && wildcardPatterns != null && !wildcardPatterns.isEmpty(); + // Compute the target lmqName set BEFORE any mutation. expandWildcardPatterns is read-only + // (it scans existing lite-topics via the lifecycle manager and matches patterns), so it is + // safe to call pre-flight. This lets us project the net reference delta and enforce the + // broker-wide quota BEFORE touching subscription state — a quota failure then leaves no + // patterns / wildcard-group marks behind that later re-expansion would mistake for a live + // (deferred-effect) subscription. + if (isPatternMode) { + lmqNameNew = expandWildcardPatterns(topic, wildcardPatterns); + } else if (isWildcardGroup) { + lmqNameNew = Collections.singleton(mockLmqNameForWildcardGroup(topic, group)); + } else { + lmqNameNew = lmqNameAll.stream() + .filter(lmqName -> liteLifecycleManager.isSubscriptionActive(topic, lmqName)) + .collect(Collectors.toSet()); + } + LiteSubscription thisSub = getOrCreateLiteSubscription(clientId, group, topic); + Set lmqNamePrev = thisSub.getLiteTopicSet(); + + // Pre-flight quota check: project the NET change in (client, lmqName) references this + // operation would register — lmqNames added that are not already held, MINUS lmqNames held + // that this operation will remove. Crediting removals means a set-replacing client at the + // limit whose net delta is zero (or negative) is not falsely rejected. A re-sync of an + // identical set has wouldAdd == 0 and passes trivially. The check runs before any mutation + // (setWildcardPatterns / markWildcardGroup / add-remove below), so a throw leaves only the + // harmless empty placeholder LiteSubscription created by getOrCreateLiteSubscription — no + // patterns, no wildcard-group mark, no active references. The throw maps to + // LITE_SUBSCRIPTION_QUOTA_EXCEEDED via the processor's existing catch. + int wouldAdd = Math.max(0, + (int) lmqNameNew.stream().filter(lmqName -> !lmqNamePrev.contains(lmqName)).count() + - (int) lmqNamePrev.stream().filter(lmqName -> !lmqNameNew.contains(lmqName)).count()); + checkQuotaOrThrow(wouldAdd); + + // Quota passed — now apply the authoritative intent and group marks. if (isPatternMode) { - // Persist the patterns (authoritative intent) and eagerly expand against existing - // lite-topics under the parent topic. The expanded lmqNames are registered as a normal - // subscription; the topic@group synthetic key is NOT used for pattern-mode groups. + // Persist the patterns (authoritative intent) and register the eagerly-expanded + // lmqNames as a normal subscription; the topic@group synthetic key is NOT used for + // pattern-mode groups. thisSub.setWildcardPatterns(wildcardPatterns); - lmqNameNew = expandWildcardPatterns(topic, wildcardPatterns); markWildcardGroup(topic, group); } else if (isWildcardGroup) { // Legacy wildcard group: receive all lite-topics under the parent topic via the @@ -151,29 +183,13 @@ public void addCompleteSubscription(String clientId, String group, String topic, // pattern-mode (doFullDispatchForWildcardGroup / reexpandWildcardPatterns key off a // non-empty wildcardPatterns set). thisSub.setWildcardPatterns(Collections.emptySet()); - lmqNameNew = Collections.singleton(mockLmqNameForWildcardGroup(topic, group)); markWildcardGroup(topic, group); } else { // Non-wildcard group: a normal subscription. Clear stale patterns from any prior // wildcard incarnation so the group is not misclassified downstream. thisSub.setWildcardPatterns(Collections.emptySet()); - lmqNameNew = lmqNameAll.stream() - .filter(lmqName -> liteLifecycleManager.isSubscriptionActive(topic, lmqName)) - .collect(Collectors.toSet()); } - Set lmqNamePrev = thisSub.getLiteTopicSet(); - // Pre-flight quota check: project the net-new (client, lmqName) references this operation - // would register (lmqNames in the new set not already held) against the broker-wide limit. - // Like addPartialSubscription, the budget is checked before any mutation; pending removals - // are NOT credited, so a set-replacing client at the limit may be denied (conservative). A - // re-sync of an identical set has wouldAdd == 0 and passes trivially. Throwing here maps to - // LITE_SUBSCRIPTION_QUOTA_EXCEEDED via the processor's existing catch. - int wouldAdd = (int) lmqNameNew.stream() - .filter(lmqName -> !lmqNamePrev.contains(lmqName)) - .count(); - checkQuotaOrThrow(wouldAdd); - // Find topics to remove (in current set but not in new set) Set lmqNameRemove = lmqNamePrev.stream() .filter(lmqName -> !lmqNameNew.contains(lmqName)) @@ -210,8 +226,9 @@ public void addCompleteSubscription(String clientId, String group, String topic, * {@link LiteQuotaException} (mapped to {@code LITE_SUBSCRIPTION_QUOTA_EXCEEDED} by the * processor) if the projected active count would exceed the configured broker-wide limit. * - * @param wouldAdd the number of net-new (client, lmqName) pairs this operation would register; - * 0 for a pure "is the broker already at the limit" check + * @param wouldAdd the NET number of (client, lmqName) references this operation would add + * (new additions minus pending removals, floored at 0); 0 for a pure + * "is the broker already at the limit" check */ private void checkQuotaOrThrow(int wouldAdd) { long maxCount = brokerController.getBrokerConfig().getMaxLiteSubscriptionCount(); 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 175060d5a73..c456e82404a 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 @@ -582,6 +582,11 @@ public void testAddCompleteSubscription_PatternWildcard_QuotaExceeded() { LiteSubscription subscription = registry.getLiteSubscription(clientId); assertNotNull(subscription); assertTrue(subscription.getLiteTopicSet().isEmpty()); + // The quota check runs before any state mutation, so a rejected complete-add must NOT leave + // behind wildcard patterns or a wildcard-group mark that later re-expansion would mistake + // for a live (deferred-effect) subscription. + assertTrue(subscription.getWildcardPatterns().isEmpty()); + assertFalse(registry.wildcardGroupMap.containsKey(topic)); } /** @@ -1429,4 +1434,79 @@ public void testExclusiveEviction_CompleteSyncReNotifiesForTombstonedLmq() { verify(mockBroker2Client, org.mockito.Mockito.atLeast(2)) .notifyUnsubscribeLite(eq(clientAChannel), captor.capture()); } + + // ==================== Quota projection: net delta & no state leak on failure ==================== + + /** + * wouldAdd must credit pending removals: a complete-add that removes one lmqName and adds a + * different one has a NET delta of zero, so it must NOT be rejected at the quota limit even + * though the gross "add" count is 1. Before crediting removals this net-zero replace was + * falsely rejected when the broker was full. + */ + @Test + public void testAddCompleteSubscription_NetZeroReplaceAtQuotaNotRejected() { + when(mockBrokerConfig.getMaxLiteSubscriptionCount()).thenReturn(1L); + String clientId = "testClient"; + String group = "testGroup"; + String topic = "testTopic"; + SubscriptionGroupConfig groupConfig = new SubscriptionGroupConfig(); + groupConfig.setGroupName(group); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(groupConfig); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + // client holds lmq1 -> activeNum=1 (at the limit) + String lmq1 = "lmq1"; + String lmq2 = "lmq2"; + registry.addCompleteSubscription(clientId, group, topic, new HashSet<>(Collections.singletonList(lmq1)), 1L); + assertEquals(1, registry.getActiveSubscriptionNum()); + + // Full re-sync swaps lmq1 -> lmq2. Gross add = 1, but net delta = 0 (lmq1 removed). Must pass. + registry.addCompleteSubscription(clientId, group, topic, new HashSet<>(Collections.singletonList(lmq2)), 2L); + + // Still at the limit, now holding lmq2 (lmq1 dropped). + assertEquals(1, registry.getActiveSubscriptionNum()); + LiteSubscription subscription = registry.getLiteSubscription(clientId); + assertFalse(subscription.getLiteTopicSet().contains(lmq1)); + assertTrue(subscription.getLiteTopicSet().contains(lmq2)); + } + + /** + * A rejected pattern-mode complete-add must leave NO live subscription state behind — not just + * an empty liteTopicSet, but empty wildcardPatterns and no wildcard-group mark — so the periodic + * reexpandWildcardPatterns / message-arriving registerArrivingLmqForPatternClients paths do not + * pick up the failed client and turn it into a deferred-effect subscription. This is the core of + * the review's "reexpand may turn a failed subscription into a delayed-effect subscription". + */ + @Test + public void testAddCompleteSubscription_PatternQuotaFailure_NoDeferredState() { + when(mockBrokerConfig.getMaxLiteSubscriptionCount()).thenReturn(1L); + String clientId = "testClient"; + String group = "testGroup"; + String topic = "order_events"; + String payRefund = LiteUtil.toLmqName(topic, "pay__refund"); + String paySuccess = LiteUtil.toLmqName(topic, "pay__success"); + when(mockSubscriptionGroupManager.findSubscriptionGroupConfig(group)).thenReturn(wildcardGroupConfig(group)); + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn(Arrays.asList(payRefund, paySuccess)); + when(mockLifecycleManager.isSubscriptionActive(eq(topic), anyString())).thenReturn(true); + + Set patterns = new HashSet<>(Collections.singletonList("pay__*")); + // wouldAdd=2 > maxCount=1 -> rejected. Critically, this runs before setWildcardPatterns / + // markWildcardGroup, so no patterns or group mark are persisted. + assertThrows(LiteQuotaException.class, () -> + registry.addCompleteSubscription(clientId, group, topic, Collections.emptySet(), patterns, 1L)); + + // No active references, and no pattern/group state a later re-expand could act on. + assertEquals(0, registry.getActiveSubscriptionNum()); + LiteSubscription subscription = registry.getLiteSubscription(clientId); + assertNotNull(subscription); + assertTrue(subscription.getWildcardPatterns().isEmpty()); + assertFalse(registry.wildcardGroupMap.containsKey(topic)); + + // Consequence: a later re-expand of this client registers nothing (it short-circuits on + // empty wildcardPatterns), and an arriving matching lmqName does not pick the client up. + when(mockLifecycleManager.collectByParentTopic(topic)).thenReturn(Arrays.asList(payRefund, paySuccess)); + assertEquals(0, registry.reexpandWildcardPatterns(clientId)); + assertEquals(0, registry.registerArrivingLmqForPatternClients(paySuccess)); + assertEquals(0, registry.getActiveSubscriptionNum()); + } }