[ISSUE #10713] Optimize ConsumerFilterManager register CPU consumption

[ymwneu]

• Refactor filter data index from topic-based to consumerGroup-based (SubscriptionFilterHandler)
• Add topic existence check before registering filter
• Generate BloomFilterData only when enableCalcFilterBitMap is enabled
• Fix thread safety: use ConcurrentHashMap for topicSqlFilterData
• Fix TOCTOU race conditions: replace containsKey+get with single get
• Rebuild subscriptionFilterData from filterDataByTopic in decode
• Add test for subscription shrink marking removed topics as dead

Which Issue(s) This PR Fixes

• Fixes [Enhancement] Optimize ConsumerFilterManager register performance #10713

Brief Description

How Did You Test This Change?

[RockteMQ-AI]

Review by github-manager-bot

Summary

Refactor ConsumerFilterManager to use consumerGroup-based indexing instead of topic-based, improving CPU efficiency during register operations and fixing thread safety issues.

Findings

• [Info] ConsumerFilterManager.java:54 — New subscriptionFilterData map uses ConcurrentHashMap correctly for thread safety.
• [Info] ConsumerFilterManager.java:118-136 — Replaced O(n*m) iteration with O(n) HashSet lookup for detecting illegal topics. Good performance improvement.
• [Info] ConsumerFilterManager.java — TOCTOU race condition fix (replacing containsKey+get with single get) is a good defensive improvement.
• [Warning] ConsumerFilterManager.java:54 — The subscriptionFilterData map is populated in register() and rebuilt in decode(), but there's no cleanup mechanism when consumer groups are removed. If consumer groups are dynamically created and destroyed, this could lead to memory accumulation over time. Consider adding a cleanup path in unregister() or a periodic cleanup task.
• [Info] ConsumerFilterManagerTest.java — Test coverage for subscription shrink marking removed topics as dead is good.

Suggestions

• Consider adding a method to clean up subscriptionFilterData when consumer groups are no longer active, or document the expected lifecycle.
• The SubscriptionFilterHandler class could benefit from a brief Javadoc explaining its role in the new indexing scheme.

Overall

Solid optimization with meaningful thread safety improvements. The memory lifecycle concern is worth addressing but not blocking.

---

Automated review by github-manager-bot

[codecov-commenter]

Codecov Report

❌ Patch coverage is 74.07407% with 28 lines in your changes missing coverage. Please review.
✅ Project coverage is 48.22%. Comparing base (eddb235) to head (0279949).

Files with missing lines	Patch %	Lines
.../rocketmq/broker/filter/ConsumerFilterManager.java	74.07%	18 Missing and 10 partials ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             develop   #10714      +/-   ##
=============================================
- Coverage      48.30%   48.22%   -0.08%     
+ Complexity     13514    13491      -23     
=============================================
  Files           1380     1380              
  Lines         101104   101138      +34     
  Branches       13107    13120      +13     
=============================================
- Hits           48834    48772      -62     
- Misses         46300    46373      +73     
- Partials        5970     5993      +23     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[RockteMQ-AI]

Review by github-manager-bot

Summary

Refactors ConsumerFilterManager from topic-based indexing to consumerGroup-based indexing via a new SubscriptionFilterHandler inner class. This improves CPU efficiency during register operations (O(n) HashSet lookup vs O(n*m) nested iteration) and adds thread safety with ConcurrentHashMap. Also adds topic existence validation and conditional BloomFilter generation.

Findings

• [Warning] ConsumerFilterManager.java:167-170 — Potential race condition: After subscriptionFilterHandler.register() returns, the code calls this.filterDataByTopic.putIfAbsent(topic, ...) followed by this.filterDataByTopic.get(topic).put(consumerFilterData). Between these two calls, another thread running clean() could remove the topic from filterDataByTopic, causing the get() to return null and the subsequent put() to throw NPE.

  • Suggestion: Capture the result of putIfAbsent and use it directly, or use computeIfAbsent:

    FilterDataMapByTopic mapByTopic = this.filterDataByTopic.computeIfAbsent(topic, FilterDataMapByTopic::new);
    mapByTopic.put(consumerFilterData);

• [Warning] ConsumerFilterManager.java:323-345 (clean method) — Cleanup order mismatch: clean() removes from subscriptionFilterData first, then from filterDataByTopic. Meanwhile register() adds to subscriptionFilterData first, then to filterDataByTopic. This ordering asymmetry could cause temporary inconsistency where filterDataByTopic has entries not present in subscriptionFilterData.

  • Suggestion: Consider using a single atomic operation or ensuring both maps are updated in the same order in all code paths.

• [Info] ConsumerFilterManager.java:54 — New subscriptionFilterData map uses ConcurrentHashMap correctly for thread-safe consumerGroup-based indexing.

• [Info] ConsumerFilterManager.java:152-156 — Topic existence check before registration is a good defensive measure. Prevents orphaned filter data for non-existent topics.

• [Info] ConsumerFilterManager.java:158-160 — Conditional BloomFilter generation (only when enableCalcFilterBitMap is true) is a valid optimization. The test correctly handles null BloomFilterData.

• [Info] ConsumerFilterManager.java:297-309 (decode) — Rebuild of subscriptionFilterData from filterDataByTopic during deserialization is correct and maintains backward compatibility with persisted data.

• [Info] ConsumerFilterManagerTest.java:274-310 — New test testRegister_bySubscriptionData_shrinkMakesDead correctly validates the shrink scenario. Good coverage.

Suggestions

1. Fix the potential NPE in register() by using computeIfAbsent or capturing the putIfAbsent result.
2. Document the expected ordering of map updates in both register() and clean() to prevent future inconsistencies.
3. Consider adding a test for concurrent register/cleanup scenarios to validate thread safety.

Verdict

Good refactoring with meaningful performance improvements. The race condition in register() should be addressed before merge.

---

Automated review by github-manager-bot

[RockteMQ-AI]

Summary

Defensive fix with proper validation and test coverage. LGTM.

---

Automated review by github-manager-bot