[Enhancement] Cache stable key strings in ConsumerOffsetManager and PullRequestHoldService, fix TransactionMetricsFlushService busy spin

[wang-jiahua]

Before Creating the Enhancement Request

• I have confirmed that this should be classified as an enhancement rather than a bug/feature.

Summary

Three independent per-message/per-request allocation and CPU optimizations in the broker layer:

1. ConsumerOffsetManager: Cache topic + "@" + group string keys in a 2-level ConcurrentHashMap to eliminate per-call String concatenation.
2. PullRequestHoldService.buildKey(): Cache topic + "@" + queueId string keys in a per-topic String[] array indexed by queueId to eliminate per-call StringBuilder + toString() allocation.
3. TransactionMetricsFlushService: Fix a busy-spin bug where waitForRunning() was only called inside the if branch, causing the loop to spin without yielding when the interval hadn't elapsed yet (~170 CPU samples on idle broker in JFR).

Motivation

JFR settings=profile on a broker under steady-state load reveals:

• ConsumerOffsetManager.commitOffset() calls topic + "@" + group on every consumer offset commit — a stable set of (topic, group) pairs generates the same key string repeatedly. Each concatenation allocates a StringBuilder + String + byte[].
• PullRequestHoldService.buildKey() creates a StringBuilder and calls toString() on every pull request hold/check — same stable key pattern.
• TransactionMetricsFlushService.run() has a correctness bug: waitForRunning() is inside the if branch, so when the interval hasn't elapsed the while loop spins without any sleep — wasting CPU on a background service thread.

Describe the Solution You'd Like

1. ConsumerOffsetManager — topicGroupKeyCache

private final ConcurrentHashMap<String, ConcurrentHashMap<String, String>> topicGroupKeyCache = new ConcurrentHashMap<>();

private String buildTopicGroupKey(String topic, String group) {
    ConcurrentHashMap<String, String> groupMap = topicGroupKeyCache.get(topic);
    if (groupMap != null) {
        String cached = groupMap.get(group);
        if (cached != null) return cached;
    }
    String key = topic + TOPIC_GROUP_SEPARATOR + group;
    topicGroupKeyCache.computeIfAbsent(topic, t -> new ConcurrentHashMap<>()).put(group, key);
    return key;
}

Replace all topic + TOPIC_GROUP_SEPARATOR + group inline concatenations with buildTopicGroupKey(topic, group).

2. PullRequestHoldService — buildKeyCache

private final ConcurrentMap<String, String[]> buildKeyCache = new ConcurrentHashMap<>();

private String buildKey(String topic, int queueId) {
    String[] keys = buildKeyCache.get(topic);
    if (keys != null && queueId >= 0 && queueId < keys.length) {
        String cached = keys[queueId];
        if (cached != null) return cached;
    }
    String key = topic + TOPIC_QUEUEID_SEPARATOR + queueId;
    // grow array if needed, cache result
    return key;
}

3. TransactionMetricsFlushService — busy spin fix

// Before (buggy): waitForRunning only inside if-branch
if (System.currentTimeMillis() - start > interval) {
    start = System.currentTimeMillis();
    persist();
    waitForRunning(interval);  // only sleeps when flushing!
}

// After (fixed): always wait, then check
long interval = brokerController.getBrokerConfig().getTransactionMetricFlushInterval();
this.waitForRunning(interval);
if (System.currentTimeMillis() - start > interval) {
    start = System.currentTimeMillis();
    persist();
}

Describe Alternatives You've Considered

• String.intern() for key caching: Rejected due to global String table contention and unpredictable GC behavior.
• Single ConcurrentHashMap<String, String> for all keys: Less cache-friendly than per-topic arrays; requires string key construction for every lookup.
• Using Thread.sleep() instead of waitForRunning(): waitForRunning() is the ServiceThread pattern that supports graceful shutdown via wakeup() — Thread.sleep() would delay shutdown.

Additional Context

Files changed:

• broker/.../offset/ConsumerOffsetManager.java — buildTopicGroupKey() + 2-level CHM cache, 10 call sites migrated
• broker/.../longpolling/PullRequestHoldService.java — buildKeyCache with String[] per topic
• broker/.../transaction/TransactionMetricsFlushService.java — waitForRunning() moved before if branch

Commercial version compatibility verified: ConsumerOffsetManagerProxy only overrides configFilePath(), no override conflicts with new private methods.

[RockteMQ-AI]

Issue Evaluation

Category: type/enhancement | Status: Evaluated

Feasibility: feasible | Scope: broker (ConsumerOffsetManager, PullRequestHoldService, TransactionMetricsFlushService) | Compatibility: no breaking changes

Three well-scoped optimizations backed by JFR profiling data:

1. ConsumerOffsetManager key caching — Valid. topic + "@" + group is called on every commitOffset() with a stable (topic, group) pair set. Caching eliminates repeated StringBuilder allocation. The proposed 2-level ConcurrentHashMap approach is sound, though consider using ConcurrentHashMap.computeIfAbsent for atomicity.

2. PullRequestHoldService buildKey caching — Valid. Per-topic String[] indexed by queueId is efficient for the typical queue count range. Ensure bounds checking for dynamic queue expansion.

3. TransactionMetricsFlushService busy-spin fix — This is a correctness bug, not just an optimization. waitForRunning() inside the if branch means the while loop spins without sleep when the interval hasn't elapsed. The fix to move it outside the if is correct and should be prioritized.

Recommendation: Item 3 (busy-spin fix) should be split into a separate PR as a bug fix. Items 1-2 can be bundled as a performance enhancement PR.

---

Automated evaluation by github-manager-bot

[qianye1001]

Thanks for the detailed profiling work — JFR analysis is always appreciated!

Item 3 (busy-spin fix) — definitely worth fixing

This is a real correctness bug: waitForRunning() inside the if branch means the loop spins without yielding when the interval hasn't elapsed. Agree with the bot's suggestion — this should be a separate bug-fix PR and prioritized.

Items 1 & 2 (string key caching) — I'd recommend not doing these

While JFR does show allocation events for StringBuilder/String from key concatenation, I don't think caching them is a net improvement. Here's why:

The "problem" is nearly free in practice. On JDK 9+, topic + "@" + group compiles to invokedynamic → StringConcatFactory, which is already well-optimized. The resulting strings are small, short-lived objects allocated in TLAB (essentially a pointer bump), and collected cheaply in Young GC. Seeing allocation events in JFR settings=profile is expected — it samples all allocations, but high event count ≠ high cost.

The cache is likely more expensive than what it replaces. The proposed 2-level ConcurrentHashMap<String, ConcurrentHashMap<String, String>> requires two CHM lookups per call (hash computation + volatile reads). Each CHM Node entry carries ~32–48 bytes of overhead (object header + hash + key/value references + next pointer). For a moderate number of topic×group pairs, the cache's memory footprint and lookup cost could exceed the cost of simply concatenating a short string.

It also introduces a memory leak risk. There's no eviction logic in the proposed code — when topics or consumer groups are removed, cached entries remain indefinitely. Over time with topic/group churn, the cache grows without bound.

The same reasoning applies to the String[] array cache in PullRequestHoldService — it adds bounds-checking complexity and dynamic resizing concerns, all to avoid a new StringBuilder(topic.length() + 10) that costs a few nanoseconds.

Suggestion: Focus this issue on the busy-spin fix (item 3), which has clear and measurable impact. For items 1 & 2, it might be worth verifying with a benchmark (e.g., JMH) whether the caching actually improves throughput or latency — my expectation is that the difference would be within noise.

[wang-jiahua]

Thanks for the review! Removed items 1 (ConsumerOffsetManager key cache) and 2 (PullRequestHoldService key cache), keeping only item 3 (TransactionMetricsFlushService busy spin fix). PR now has a single file change (+7/-2).

[RockteMQ-AI]

Issue Evaluation

Category: type/enhancement | Status: Evaluated

Feasibility: Feasible
Scope: ConsumerOffsetManager, PullRequestHoldService, TransactionMetricsFlushService
Compatibility: No breaking changes — internal caching with same external behavior.

Well-motivated with JFR profiling evidence. The three changes are independent and each is low-risk:

1. ConsumerOffsetManager key caching — Sound approach. The set of (topic, group) pairs is bounded and stable. Consider using a single ConcurrentHashMap<String, String> with a composite key instead of a 2-level map to reduce memory overhead.
2. PullRequestHoldService buildKey caching — String[] indexed by queueId is efficient. Ensure cache invalidation on topic deletion.
3. TransactionMetricsFlushService busy-spin fix — This is actually a correctness bug (CPU waste from missing waitForRunning() in the else branch). Should be straightforward to fix.

The enhancement and bug fix parts could be split into separate PRs for easier review.

---

Automated evaluation by RockteMQ-AI

[RockteMQ-AI]

Issue Evaluation

Category: type/enhancement | Status: Evaluated

Well-structured proposal with three independent optimizations:

1. ConsumerOffsetManager key caching — Valid. Caching topic + "@" + group keys eliminates per-call StringBuilder allocation for a stable (topic, group) pair set. The 2-level ConcurrentHashMap approach is reasonable.
2. PullRequestHoldService.buildKey() caching — Valid. Per-topic String[] indexed by queueId avoids StringBuilder + toString() on every pull hold/check.
3. TransactionMetricsFlushService busy-spin — This is a correctness bug, not just an enhancement. waitForRunning() inside the if branch means the loop spins without yielding when the interval has not elapsed. This should be tracked as a separate bug fix if needed.

Feasibility: Feasible. Each optimization is isolated and low-risk.
Scope: broker module (store/remoting layers)
Compatibility: No breaking changes.

---

Automated evaluation by RockteMQ-AI