fix: validate Lite subscription topics and enforce quota - #10897
Conversation
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## develop #10897 +/- ##
=============================================
- Coverage 48.60% 48.51% -0.10%
+ Complexity 13690 13664 -26
=============================================
Files 1381 1381
Lines 101464 101480 +16
Branches 13187 13191 +4
=============================================
- Hits 49318 49231 -87
- Misses 46158 46237 +79
- Partials 5988 6012 +24 ☔ 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 fixes a validation gap in Lite subscription topic checking. Previously, only the first subscription's topic was validated, meaning mismatched topics in non-first positions would silently pass. The fix iterates through all subscriptions to validate each one.
Verdict: LGTM — clean, minimal fix with proper test coverage using LinkedHashSet to ensure the mismatching entry isn't accidentally first.
Automated review by github-manager-bot
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
Validates all Lite subscriptions against the bound topic, not just the first. Clean validation fix with test coverage.
LGTM.
Automated review by github-manager-bot
24215e2 to
a46ac33
Compare
Aias00
changed the title
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
This PR modifies 4 files (221 lines changed). Some observations noted below.
Findings
- [INFO] multiple:1 — Concurrency-related changes detected. Please verify thread safety under concurrent message processing scenarios.
Automated review by github-manager-bot
Additional notes (not anchored to a changed line)
- [INFO]
multiple:1— Concurrency-related changes detected. Please verify thread safety under concurrent message processing scenarios. (line outside diff)
RockteMQ-AI
left a comment
RockteMQ-AI
left a comment
There was a problem hiding this comment.
Summary
Consolidates Lite subscription validation and quota enforcement. Key improvements:
synchronizedon mutation methods prevents race conditions under concurrent subscription updates- Quota check moved before state mutation (was after) — prevents partial state on rejection
ensureQuotaproperly accounts for removed subscriptions when checking capacity- Good test coverage for batch quota overflow scenarios
The refactoring is clean and the thread-safety improvement is important for production brokers. LGTM.
Automated review by github-manager-bot
Summary\n- validate every Lite subscription topic before processing client subscription updates\n- enforce Lite subscription quota in the broker registry\n\nThis consolidates the adjacent quota change from #10896 into one Lite subscription integrity fix.\n\nCloses #10895\n\n## Validation\n- [java] JVM args ignored when same JVM is used.
T E S T S
Running org.apache.rocketmq.broker.lite.LiteSubscriptionRegistryImplTest
05:50:55,860 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.classic.LoggerContext[default] - This is logback-classic version 1.3.5-rocketmq
05:50:55,874 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.classic.LoggerContext[default] - Found resource [rmq.logback-test.xml] at [file:/private/tmp/rmq-lite-subscription-topic-validation/broker/target/test-classes/rmq.logback-test.xml]
05:50:55,954 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.core.model.processor.AppenderModelHandler - Processing appender named [CONSOLE]
05:50:55,954 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.core.model.processor.AppenderModelHandler - About to instantiate appender of type [org.apache.rocketmq.logging.ch.qos.logback.core.ConsoleAppender]
05:50:55,980 |-WARN in org.apache.rocketmq.logging.ch.qos.logback.core.ConsoleAppender[CONSOLE] - This appender no longer admits a layout as a sub-component, set an encoder instead.
05:50:55,980 |-WARN in org.apache.rocketmq.logging.ch.qos.logback.core.ConsoleAppender[CONSOLE] - To ensure compatibility, wrapping your layout in LayoutWrappingEncoder.
05:50:55,980 |-WARN in org.apache.rocketmq.logging.ch.qos.logback.core.ConsoleAppender[CONSOLE] - See also http://logback.qos.ch/codes.html#layoutInsteadOfEncoder for details
05:50:55,980 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.classic.model.processor.LoggerModelHandler - Setting level of logger [org.apache.rocketmq] to ERROR
05:50:55,980 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.classic.model.processor.LoggerModelHandler - Setting additivity of logger [org.apache.rocketmq] to false
05:50:55,980 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.core.model.processor.AppenderRefModelHandler - Attaching appender named [CONSOLE] to Logger[org.apache.rocketmq]
05:50:55,980 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.classic.model.processor.RootLoggerModelHandler - Setting level of ROOT logger to INFO
05:50:55,981 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.core.model.processor.AppenderRefModelHandler - Attaching appender named [CONSOLE] to Logger[ROOT]
05:50:55,981 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.core.model.processor.DefaultProcessor@671d97bc - End of configuration.
05:50:55,981 |-INFO in org.apache.rocketmq.common.logging.JoranConfiguratorExt@774f2992 - Registering current configuration as safe fallback point
2026-08-14 20:50:56.890 INFO [ RocketmqPopLite] [ main] notifyUnsubscribeLite liteTopic:null, group:exclusiveGroup, topic:null, clientId:clientA
2026-08-14 20:50:56.892 INFO [ RocketmqPopLite] [ main] excludeClientByLmqName group:exclusiveGroup, lmqName:lmq1, resetOffset:false, clientId:clientA -> clientB
2026-08-14 20:50:56.899 INFO [ RocketmqPopLite] [ main] re-notify unsubscribe for tombstoned lmqName, clientId:clientA, group:exclusiveGroup, lmqName:lmq1
2026-08-14 20:50:56.899 INFO [ RocketmqPopLite] [ main] notifyUnsubscribeLite liteTopic:null, group:exclusiveGroup, topic:null, clientId:clientA
2026-08-14 20:50:56.906 INFO [ RocketmqPopLite] [ main] try to reset lite offset. testGroup, lmq1, testClient, OffsetOption{type=POLICY, value=2}, current:100, target:500
2026-08-14 20:50:56.917 INFO [ RocketmqPopLite] [ main] notifyUnsubscribeLite liteTopic:lmq1, group:testGroup, topic:testTopic, clientId:testClient
2026-08-14 20:50:56.924 INFO [ RocketmqPopLite] [ main] try to reset lite offset. testGroup, lmq1, testClient, OffsetOption{type=POLICY, value=1}, current:100, target:0
2026-08-14 20:50:56.931 INFO [ RocketmqPopLite] [ main] try to reset lite offset. testGroup, lmq1, testClient, OffsetOption{type=OFFSET, value=250}, current:100, target:250
2026-08-14 20:50:56.933 INFO [ RocketmqPopLite] [ main] notifyUnsubscribeLite liteTopic:null, group:exclusiveGroup, topic:null, clientId:clientA
2026-08-14 20:50:56.933 INFO [ RocketmqPopLite] [ main] excludeClientByLmqName group:exclusiveGroup, lmqName:lmq1, resetOffset:false, clientId:clientA -> clientB
2026-08-14 20:50:56.935 INFO [ RocketmqPopLite] [ main] notifyUnsubscribeLite liteTopic:null, group:exclusiveGroup, topic:null, clientId:clientA
2026-08-14 20:50:56.935 INFO [ RocketmqPopLite] [ main] excludeClientByLmqName group:exclusiveGroup, lmqName:lmq1, resetOffset:false, clientId:clientA -> clientB
2026-08-14 20:50:56.937 INFO [ RocketmqPopLite] [ main] removeCompleteSubscription, topic:testTopic, group:exclusiveGroup, clientId:clientA
2026-08-14 20:50:56.937 INFO [ RocketmqPopLite] [ main] Remove expired LiteSubscription, topic: testTopic, group: exclusiveGroup, clientId: clientA, timeout: 10000ms, expired: 60002ms
2026-08-14 20:50:56.940 INFO [ RocketmqPopLite] [ main] removeCompleteSubscription, topic:testTopic, group:testGroup, clientId:testClient
2026-08-14 20:50:56.940 INFO [ RocketmqPopLite] [ main] Remove expired LiteSubscription, topic: testTopic, group: testGroup, clientId: testClient, timeout: 10000ms, expired: 20000ms
2026-08-14 20:50:56.941 INFO [ RocketmqPopLite] [ main] removeCompleteSubscription, topic:testTopic, group:testGroup, clientId:testClient
2026-08-14 20:50:56.944 INFO [ RocketmqPopLite] [ main] notifyUnsubscribeLite liteTopic:null, group:exclusiveGroup, topic:null, clientId:clientA
2026-08-14 20:50:56.945 INFO [ RocketmqPopLite] [ main] excludeClientByLmqName group:exclusiveGroup, lmqName:lmq1, resetOffset:false, clientId:clientA -> clientB
2026-08-14 20:50:56.945 INFO [ RocketmqPopLite] [ main] notifyUnsubscribeLite liteTopic:null, group:exclusiveGroup, topic:null, clientId:clientA
2026-08-14 20:50:56.945 INFO [ RocketmqPopLite] [ main] excludeClientByLmqName group:exclusiveGroup, lmqName:lmq2, resetOffset:false, clientId:clientA -> clientB
2026-08-14 20:50:56.946 INFO [ RocketmqPopLite] [ main] notifyUnsubscribeLite liteTopic:null, group:exclusiveGroup, topic:null, clientId:clientA
2026-08-14 20:50:56.946 INFO [ RocketmqPopLite] [ main] excludeClientByLmqName group:exclusiveGroup, lmqName:lmq1, resetOffset:false, clientId:clientA -> clientB
2026-08-14 20:50:56.947 INFO [ RocketmqPopLite] [ main] notifyUnsubscribeLite liteTopic:null, group:exclusiveGroup, topic:null, clientId:clientB
2026-08-14 20:50:56.947 INFO [ RocketmqPopLite] [ main] excludeClientByLmqName group:exclusiveGroup, lmqName:lmq1, resetOffset:false, clientId:clientB -> clientA
2026-08-14 20:50:56.949 WARN [ RocketmqPopLite] [ main] notifyUnsubscribeLite but channel is null, liteTopic:null, group:testGroup, topic:null, clientId:testClient1,
2026-08-14 20:50:56.949 INFO [ RocketmqPopLite] [ main] excludeClientByLmqName group:testGroup, lmqName:lmq1, resetOffset:false, clientId:testClient1 -> testClient2
2026-08-14 20:50:56.951 INFO [ RocketmqPopLite] [ main] notifyUnsubscribeLite liteTopic:null, group:exclusiveGroup, topic:null, clientId:clientA
2026-08-14 20:50:56.951 INFO [ RocketmqPopLite] [ main] excludeClientByLmqName group:exclusiveGroup, lmqName:lmq1, resetOffset:false, clientId:clientA -> clientB
2026-08-14 20:50:56.952 INFO [ RocketmqPopLite] [ main] notifyUnsubscribeLite liteTopic:null, group:exclusiveGroup, topic:null, clientId:clientA
2026-08-14 20:50:56.952 INFO [ RocketmqPopLite] [ main] excludeClientByLmqName group:exclusiveGroup, lmqName:lmq2, resetOffset:false, clientId:clientA -> clientB
2026-08-14 20:50:56.952 INFO [ RocketmqPopLite] [ main] re-notify unsubscribe for tombstoned lmqName, clientId:clientA, group:exclusiveGroup, lmqName:lmq2
2026-08-14 20:50:56.952 INFO [ RocketmqPopLite] [ main] notifyUnsubscribeLite liteTopic:null, group:exclusiveGroup, topic:null, clientId:clientA
2026-08-14 20:50:56.956 INFO [ RocketmqPopLite] [ main] removeCompleteSubscription, topic:testTopic, group:testGroup, clientId:testClient
Tests run: 33, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 2.113 sec - in org.apache.rocketmq.broker.lite.LiteSubscriptionRegistryImplTest
Results :
Tests run: 33, Failures: 0, Errors: 0, Skipped: 0
T E S T S
Running org.apache.rocketmq.proxy.processor.ClientProcessorTest
05:51:07,637 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.classic.LoggerContext[default] - This is logback-classic version 1.3.5-rocketmq
05:51:07,661 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.classic.LoggerContext[default] - Found resource [rmq.logback-test.xml] at [file:/private/tmp/rmq-lite-subscription-topic-validation/proxy/target/test-classes/rmq.logback-test.xml]
05:51:07,807 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.core.model.processor.AppenderModelHandler - Processing appender named [CONSOLE]
05:51:07,807 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.core.model.processor.AppenderModelHandler - About to instantiate appender of type [org.apache.rocketmq.logging.ch.qos.logback.core.ConsoleAppender]
05:51:07,857 |-WARN in org.apache.rocketmq.logging.ch.qos.logback.core.ConsoleAppender[CONSOLE] - This appender no longer admits a layout as a sub-component, set an encoder instead.
05:51:07,857 |-WARN in org.apache.rocketmq.logging.ch.qos.logback.core.ConsoleAppender[CONSOLE] - To ensure compatibility, wrapping your layout in LayoutWrappingEncoder.
05:51:07,857 |-WARN in org.apache.rocketmq.logging.ch.qos.logback.core.ConsoleAppender[CONSOLE] - See also http://logback.qos.ch/codes.html#layoutInsteadOfEncoder for details
05:51:07,857 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.classic.model.processor.LoggerModelHandler - Setting level of logger [org.apache.rocketmq] to ERROR
05:51:07,857 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.classic.model.processor.LoggerModelHandler - Setting additivity of logger [org.apache.rocketmq] to false
05:51:07,857 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.core.model.processor.AppenderRefModelHandler - Attaching appender named [CONSOLE] to Logger[org.apache.rocketmq]
05:51:07,858 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.classic.model.processor.RootLoggerModelHandler - Setting level of ROOT logger to ERROR
05:51:07,858 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.core.model.processor.AppenderRefModelHandler - Attaching appender named [CONSOLE] to Logger[ROOT]
05:51:07,858 |-INFO in org.apache.rocketmq.logging.ch.qos.logback.core.model.processor.DefaultProcessor@61191222 - End of configuration.
05:51:07,859 |-INFO in org.apache.rocketmq.common.logging.JoranConfiguratorExt@58833798 - Registering current configuration as safe fallback point
Tests run: 13, Failures: 0, Errors: 0, Skipped: 0, Time elapsed: 2.114 sec - in org.apache.rocketmq.proxy.processor.ClientProcessorTest
Results :
Tests run: 13, Failures: 0, Errors: 0, Skipped: 0