[ISSUE #10988] Fix concurrent commitOffset race losing queue offsets for a new topic@group

[unbridled-41]

Which Issue(s) This PR Fixes

• Fixes [Bug] Concurrent commitOffset for a new topic@group can drop queue offsets (check-then-act race) #10988

Brief Description

ConsumerOffsetManager#commitOffset used a non-atomic check-then-act to create the per topic@group map:

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).

[RockteMQ-AI]

Summary

This PR fixes a race condition in ConsumerOffsetManager.commitOffset where concurrent commits for different queues of a new topic@group could overwrite each other's entries. The original check-then-act pattern (if null, create and put) is replaced with the atomic computeIfAbsent, ensuring only one map is created per key. A comprehensive test simulates concurrent commits from 8 threads across 500 keys to verify the fix.

LGTM — excellent fix for the concurrency bug!

---

Automated review by github-manager-bot

[unbridled-41]

Test evidence (before → after)

ConsumerOffsetManagerTest#testConcurrentCommitOffsetDoesNotLoseQueues was run repeatedly on JDK 8 with the fix reverted and with the fix applied (5 runs each, mvn -pl broker test -Dtest='ConsumerOffsetManagerTest#testConcurrentCommitOffsetDoesNotLoseQueues'):

On the unfixed code: 5/5 runs fail. Example failure (queues 0 and 7 of topic1 lost — the loser's map was overwritten):

java.lang.AssertionError:
[offsets of topic1]
Expected size: 8 but was: 6 in:
{1=1L, 2=1L, 3=1L, 4=1L, 5=1L, 6=1L}

The race window is the very first offsetTable.get(key)/offsetTable.put(key, map) pair per fresh topic@group key, so it reproduces reliably, not stochastically.

With this PR: 5/5 runs pass (Tests run: 1, Failures: 0, Errors: 0, Skipped: 0 each run; the whole ConsumerOffsetManagerTest is 6/6 green).

The loss is not just a test artifact: every commit the losing thread made for that queue is gone until that queue's next commit (5s flush cadence, but queryOffset returns -1 immediately), so a consumer reconnect in that window re-initializes the queue per consumeFromWhere.

Side note on CI: the workflow runs for this PR are in action_required state (first-time contributor) and will start once a maintainer approves them.