[Enhancement] Cache stable key strings in ConsumerOffsetManager and PullRequestHoldService, fix TransactionMetricsFlushService busy spin
- Dominant language
- Java
- Stars
- 22.6k
- Forks
- 12k
- Avg merge
- 3d 1h
- Merged PRs (30d)
- 27
Description
### Before Creating the Enhancement Request
- [x] 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**
```java
private final ConcurrentHashMap> topicGroupKeyCache = new ConcurrentHashMap<>();
private String buildTopicGroupKey(String topic, String group) {
ConcurrentHashMap 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**
```java
private final ConcurrentMap 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**
```java
// 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` 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.
Contributor guide
Research direction
Start with broker/.../offset/ConsumerOffsetManager.java, broker/.../longpolling/PullRequestHoldService.java, and broker/.../transaction/TransactionMetricsFlushService.java. Review the listed key-building call sites and the run loop, then verify that repeated keys are cached and the service waits before checking its interval without changing shutdown behavior.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- backend, distributed-systems
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 52/100