[ISSUE #10929] Add LmqPrefixIndex to accelerate wildcard dispatch and refine lite subscription model - #10930
[ISSUE #10929] Add LmqPrefixIndex to accelerate wildcard dispatch and refine lite subscription model#10930f1amingo wants to merge 2 commits into
Conversation
…ch and refine lite subscription model - Add LmqPrefixIndex, a prefix-ordered in-memory index over lmq names, bootstrapped at startup and maintained via onLmqCreate/onLmqDelete hooks - Rewrite wildcard full dispatch, getLiteTopicCount, collectByParentTopic and cleanByParentTopic on index-backed forEachLiteTopicByPrefix/ByParent, turning O(total lmqs) scans into O(matched lmqs) - Refine lite subscription model: replace SubscriberWrapper with Map-based getAllSubscribers, simplify getWildcardGroupClients, rename getLiteTopicSet to getLmqSet, make LiteCtlListener callbacks default methods
d571cb1 to
11f5473
Compare
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
This PR modifies 24 file(s) (2982 lines of diff).
Changed Files
WORKSPACE
broker/BUILD.bazel
broker/pom.xml
broker/src/main/java/org/apache/rocketmq/broker/BrokerController.java
broker/src/main/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManager.java
broker/src/main/java/org/apache/rocketmq/broker/lite/LiteCtlListener.java
broker/src/main/java/org/apache/rocketmq/broker/lite/LiteEventDispatcher.java
broker/src/main/java/org/apache/rocketmq/broker/lite/LiteLifecycleManager.java
broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistry.java
broker/src/main/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImpl.java
broker/src/main/java/org/apache/rocketmq/broker/lite/LmqPrefixIndex.java
broker/src/main/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManager.java
broker/src/main/java/org/apache/rocketmq/broker/lite/SubscriberWrapper.java
broker/src/main/java/org/apache/rocketmq/broker/processor/LiteManagerProcessor.java
broker/src/test/java/org/apache/rocketmq/broker/lite/AbstractLiteLifecycleManagerTest.java
broker/src/test/java/org/apache/rocketmq/broker/lite/LiteEventDispatcherTest.java
broker/src/test/java/org/apache/rocketmq/broker/lite/LiteLifecycleManagerTest.java
broker/src/test/java/org/apache/rocketmq/broker/lite/LiteSubscriptionRegistryImplTest.java
broker/src/test/java/org/apache/rocketmq/broker/lite/LmqPrefixIndexTest.java
broker/src/test/java/org/apache/rocketmq/broker/lite/RocksDBLiteLifecycleManagerTest.java
broker/src/test/java/org/apache/rocketmq/broker/processor/LiteManagerProcessorTest.java
common/src/main/java/org/apache/rocketmq/common/lite/LiteSubscription.java
common/src/test/java/org/apache/rocketmq/common/lite/LiteSubscriptionTest.java
pom.xml
Automated review by RockteMQ-AI
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## develop #10930 +/- ##
=============================================
- Coverage 48.60% 48.57% -0.04%
- Complexity 13690 13696 +6
=============================================
Files 1381 1382 +1
Lines 101464 101461 -3
Branches 13187 13183 -4
=============================================
- Hits 49318 49285 -33
- Misses 46158 46162 +4
- Partials 5988 6014 +26 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
This PR introduces LmqPrefixIndex, a PatriciaTrie-based prefix index over lmq names to accelerate wildcard dispatch and subscription queries. The change replaces O(n) scans with O(log n) prefix lookups, which is a significant performance improvement for brokers with many lmqs.
Overall: Good design with proper thread safety and comprehensive tests. A few minor concerns noted below.
Key Changes
- LmqPrefixIndex: New PatriciaTrie-backed index with ReadWriteLock for thread-safe prefix queries
- Lifecycle hooks: Added
onLmqCreate()/onLmqDelete()to maintain the index incrementally - Bootstrap: Populates the index once at startup via
bootstrapLmqPrefixIndex() - Refactoring: Renamed
liteTopic→lmqthroughout for consistency, simplifiedSubscriberWrapperto direct Map returns
Findings
- [Warning]
LiteSubscription.java:32— Removal of automaticupdateTime()requires all callers to explicitly calltouch() - [Info]
AbstractLiteLifecycleManager.java:82— Bootstrap should be called after full initialization
Suggestions
- Consider adding a unit test for concurrent
add()/remove()/prefixLookup()operations onLmqPrefixIndexto verify the ReadWriteLock behavior under load - The
dispatch()method callsonLmqCreate()beforedoDispatch()— this is correct, but worth documenting that the subscription registry update happens asynchronously
Cross-repo Note
This change affects the lite subscription protocol. If there are corresponding changes needed in rocketmq-clients for the wildcard subscription feature, please coordinate the release.
Automated review by github-manager
| updateTime(); | ||
| this.liteTopicSet.addAll(set); | ||
| public LiteSubscription touch() { | ||
| this.updateTime = System.currentTimeMillis(); |
There was a problem hiding this comment.
[Warning] The removal of automatic updateTime() from addLmq()/removeLmq() and requiring explicit touch() is a behavioral change. Please verify all callers remember to call touch() after modifying the lmq set, otherwise the subscription may be incorrectly evicted due to stale updateTime.
| */ | ||
| public abstract long getMaxOffsetInQueue(String lmqName); | ||
| public void bootstrapLmqPrefixIndex() { | ||
| long start = System.currentTimeMillis(); |
There was a problem hiding this comment.
[Info] bootstrapLmqPrefixIndex() iterates all lmqs at startup. Ensure this is called after the broker is fully initialized and no concurrent lmq create/delete operations are in progress, otherwise the index may be inconsistent. Consider adding a comment about the expected call order.
3 participants
Which Issue(s) This PR Fixes
Brief Description
Several lite-topic hot paths (
doFullDispatchForWildcardGroup,getLiteTopicCount,collectByParentTopic,cleanByParentTopic) iterate the entire consume-queue table on every invocation, so their cost grows with the total number of lmqs on the broker instead of the lmqs under the target parent topic.This PR adds
LmqPrefixIndex, a prefix-ordered in-memory index over lmq names:bootstrapLmqPrefixIndexafter lifecycleinit), then maintained on the lmq hot path viaonLmqCreate(first message of a new lmq) andonLmqDeletehooks.forEachLiteTopicByPrefix/forEachLiteTopicByParentAPIs; wildcard full dispatch, topic count and cleanup are rewritten on top of them, turning O(total lmqs) scans into O(matched lmqs). Cleanup switches to collect-then-delete to avoid lock nesting between iteration and deletion.SubscriberWrapperwith a plainMap<String, List<ClientGroup>>return fromgetAllSubscribers(exact + wildcard sources), simplifygetWildcardGroupClientssignature, renamegetLiteTopicSettogetLmqSet, and makeLiteCtlListenercallbacks default methods.No protocol or external API change; dispatch and cleanup behavior stays equivalent.
How Did You Test This Change?
LmqPrefixIndexTestcovering prefix lookup boundaries, bootstrap and create/delete hooksAbstractLiteLifecycleManagerTest/LiteEventDispatcherTestfor index-backed dispatch, count and cleanup pathsLiteSubscriptionRegistryImplTestand newLiteSubscriptionTestfor the subscription model refactoringmvn -pl broker -am testrelated suites (128+ tests) pluscommon/proxylite suites, all green;mvn checkstyleclean