RockteMQ-AI commented on code in PR #10528:
URL: https://github.com/apache/rocketmq/pull/10528#discussion_r3867553364
##########
store/src/main/java/org/apache/rocketmq/store/index/IndexService.java:
##########
@@ -215,10 +216,12 @@ public QueryOffsetResult queryOffset(String topic, String
key, int maxNum, long
}
private String buildKey(final String topic, final String key) {
- return topic + "#" + key;
+ reusableKeyBuilder.setLength(0);
Review Comment:
**[Critical] Thread-safety bug**: `reusableKeyBuilder` is an instance field
shared across threads, but `buildKey()` is called from `queryOffset()` which
can be invoked concurrently by multiple admin/consumer query threads.
StringBuilder is NOT thread-safe — concurrent calls to `setLength(0)` and
`append()` will interleave, producing corrupted keys like `topic1#key2` instead
of `topic1#key1`.
This will cause index lookups to return wrong results or miss entries
entirely.
**Fix**: Use `ThreadLocal<StringBuilder>` instead of a shared instance
field, or revert to string concatenation (the JIT compiler optimizes it well).
##########
store/src/main/java/org/apache/rocketmq/store/timer/TimerWheel.java:
##########
@@ -128,17 +130,25 @@ public void flush() {
if (mappedByteBuffer == null) {
return;
}
+ if (!dirty) {
Review Comment:
**[Critical] Thread-safety bug**: The `dirty` flag is instance-level (shared
across threads), but `localBuffer` is `ThreadLocal<ByteBuffer>` (per-thread).
This creates a race condition:
1. Thread A calls `putSlot()` → sets `dirty = true`
2. Thread B calls `flush()` → sees `dirty = true` → flushes Thread B's local
buffer
3. Thread B's buffer may be uninitialized (all zeros) or contain stale data
4. Flushing zeros over the mapped file corrupts the timer wheel
**Fix**: Make `dirty` a `ThreadLocal<Boolean>` to match the per-thread
buffer, or use a synchronized block around flush operations.
##########
store/src/main/java/org/apache/rocketmq/store/queue/QueueOffsetOperator.java:
##########
@@ -123,10 +132,18 @@ public void setLmqTopicQueueTable(ConcurrentMap<String,
Long> lmqTopicQueueTable
}
public ConcurrentMap<String, Long> getTopicQueueTable() {
- return topicQueueTable;
+ ConcurrentMap<String, Long> snapshot = new
ConcurrentHashMap<>(this.topicQueueTable.size() * 2);
Review Comment:
**[Warning] Behavioral change**: `getTopicQueueTable()` now returns a
snapshot copy instead of the live map. While this is safer for thread safety,
it's a behavioral change that could affect callers who expected to modify the
map directly.
If any code relies on modifying the returned map and seeing those changes
reflected in the internal state, it will break silently.
**Recommendation**: Document this change in the method's Javadoc, or
consider whether callers actually need live access.
--
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]
