ymwneu opened a new issue, #10713:
URL: https://github.com/apache/rocketmq/issues/10713
### Before Creating the Enhancement Request
- [x] I have confirmed that this should be classified as an enhancement
rather than a bug/feature.
### Summary
The current ConsumerFilterManager uses a topic-based index
(filterDataByTopic) as the sole data structure for managing SQL92 consumer
filter data. When register(consumerGroup, subList) is called during heartbeat
processing, it invokes getByGroup(consumerGroup) which performs a full scan
across all topics and all consumer groups — an O(T × G) operation. In clusters
with a large number of topics and SQL92 filter consumers, this becomes a
significant CPU bottleneck. Additionally, there are several thread safety
issues in the concurrent access patterns.
### Motivation
During production operations in a large-scale cluster, we observed high CPU
consumption on the broker caused by ConsumerFilterManager.register(). The root
cause is the getByGroup() method, which is called on every consumer heartbeat
to detect unsubscribed topics. It iterates through the entire filterDataByTopic
map (all topics × all consumer groups per topic) just to collect entries
belonging to one consumer group. This results in O(T × G) time complexity per
heartbeat.
For example, with 10,000 topics and 100 SQL92 filter consumer groups, each
heartbeat triggers a scan of up to 1,000,000 entries. Under high consumer count
scenarios, this becomes a serious CPU hotspot.
Beyond the performance issue, the existing code also has:
1. TOCTOU race conditions: Multiple containsKey() + get() sequences on
ConcurrentHashMap where the entry can be removed between the two calls, leading
to NullPointerException.
2. Unconditional BloomFilter generation: bloomFilter.generate() is called
on every register even when enableCalcFilterBitMap is disabled, wasting CPU
cycles.
3. No topic existence validation: Filter data is registered for topics
that may not exist on the broker, accumulating stale entries.
### Describe the Solution You'd Like
Introduce a consumerGroup-based index (SubscriptionFilterHandler) alongside
the existing topic-based index to enable O(1) lookups by consumer group:
1. Add subscriptionFilterData map (ConcurrentMap<String/*consumerGroup*/,
SubscriptionFilterHandler>): Each SubscriptionFilterHandler holds a
ConcurrentHashMap<String/*topic*/,ConsumerFilterData> for that consumer group.
This eliminates the need for full-scan getByGroup() in the hot path.
2. Dual-index consistency: Both filterDataByTopic (for topic-based queries
during message dispatch) and subscriptionFilterData (for consumerGroup-based
operations during heartbeat) are maintained in sync. Writes go through
subscriptionFilterData first, then update filterDataByTopic.
3. Conditional BloomFilter generation: Only call bloomFilter.generate()
when BrokerConfig.isEnableCalcFilterBitMap() is true, consistent with
CommitLogDispatcherCalcBitMap.
4. Topic existence check: Validate that the topic exists in
TopicConfigManager before registering filter data, preventing accumulation of
stale entries.
5. Fix TOCTOU race conditions: Replace all containsKey() + get() sequences
with a single get() assigned to a local variable, then null-check.
6. Thread safety for SubscriptionFilterHandler: Use ConcurrentHashMap
instead of HashMap for topicSqlFilterData since it is accessed concurrently.
7. Rebuild on decode: After deserializing filterDataByTopic from disk,
rebuild subscriptionFilterData to ensure consistency after broker restart.
### Describe Alternatives You've Considered
Describe Alternatives You've Considered
1. Cache getByGroup() results: Maintain a separate cache of consumerGroup
→ topics mapping that is invalidated on register/unregister. This would reduce
the scan frequency but adds cache invalidation complexity and still requires
the full scan to rebuild the cache. The dual-index approach is simpler and
provides O(1) at all times.
2. Replace filterDataByTopic entirely with a consumerGroup-based
structure: This would simplify the code but would degrade the topic-based
lookup path used by CommitLogDispatcherCalcBitMap during message dispatch
(called on every message), which is performance-critical.
3. Add synchronization instead of using ConcurrentHashMap: Using
synchronized blocks would fix thread safety but introduce lock contention on
the heartbeat path.
ConcurrentHashMap provides better concurrent throughput with no contention
on reads.
### Additional Context
_No response_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
