unbridled-41 opened a new pull request, #10992:
URL: https://github.com/apache/rocketmq/pull/10992
<!-- Please make sure the target branch is right. In most case, the target
branch should be `develop`. -->
### Which Issue(s) This PR Fixes
- Fixes #10988
### Brief Description
`ConsumerOffsetManager#commitOffset` used a non-atomic check-then-act to
create the per `topic@group` map:
```java
ConcurrentMap<Integer, Long> map = this.offsetTable.get(key);
if (null == map) {
map = new ConcurrentHashMap<>(2);
map.put(queueId, offset);
this.offsetTable.put(key, map); // last put wins, the other
thread's map is dropped
} else { ... }
```
When two remoting threads commit offsets for different queues of the same
brand-new `topic@group` concurrently (the normal situation right after a
consumer group starts on a multi-queue topic, or while a pop consumer acks
several queues), both see `null`, both install their own map, and the second
`put` silently drops the first queue's offset. `queryOffset` then returns `-1`
for that queue until its next commit, so a consumer reconnect/restart in the
window re-initializes per `consumeFromWhere` (duplicate consumption or skipping
to max). The same race re-arms after `removeOffset(group)` /
`cleanOffsetByTopic` while the group is still consuming.
The fix replaces the get/put pair with `offsetTable.computeIfAbsent(key, k
-> new ConcurrentHashMap<>(2))` — the same idiom the class already uses in
`commitPullOffset` — so concurrent commits always land in the same map. The
less-than-store warn behavior is unchanged.
### How Did You Test This Change?
Added
`ConsumerOffsetManagerTest#testConcurrentCommitOffsetDoesNotLoseQueues`: 8
threads commit a distinct queue for 500 brand-new `topic@group` keys each, then
the test asserts every key retains all 8 queue offsets. It fails reproducibly
on the unfixed code and passes with this change.
`mvn -pl broker test -Dtest=ConsumerOffsetManagerTest` passes (6/6).
--
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]
