[ISSUE #10700] Make consumer offset commits concurrency-safe

[ai-yang]

Which Issue(s) This PR Fixes

• Fixes [Bug] Concurrent first offset commits can overwrite queues in consumer offset managers #10700

Brief Description

Make first offset-map initialization atomic in the classic and RocksDB v1 consumer offset managers, and serialize incremental whole-map persistence for the same topic@group.

• Publish one shared inner queue-offset map with outer putIfAbsent, then update the selected winner map.
• Preserve the RocksDB v1 LMQ-specific map construction.
• In incremental mode, order the queue update, version update, whole-map serialization, and WAL batch write under the inner map monitor.
• Keep non-incremental commits lock-free across the existing concurrent map and allow different topic@group keys to proceed independently.
• Leave RocksDB v2 unchanged because it already initializes atomically and persists each queue under an independent key.

Root Cause

Classic and RocksDB v1 used check-then-act initialization followed by unconditional outer put, so two first commits could publish different inner maps and overwrite one queue. RocksDB v1 incremental mode additionally allowed an older serialized whole-map batch to be written after a newer batch for the same RocksDB key.

Historical unmerged PR #1427 identified the classic initialization race. This PR preserves that analysis while extending coverage to RocksDB v1, incremental WAL ordering, and deterministic persistence/reload tests. Open PRs #10625, #9602, and #9877 touch related files but do not change these commit paths.

Impact

Concurrent first commits retain every queue in memory and after persistence. Incremental restart recovery cannot regress to an older same-key snapshot. Existing rollback warnings, version semantics, LMQ allocation, and v2 behavior are preserved.

How Did You Test This Change?

• Deterministic red tests on the previous implementation:
  • classic concurrent first commits: expected 2 queues, reloaded 1;
  • RocksDB v1 periodic LMQ commits: expected 2 queues, retained 1;
  • RocksDB v1 incremental delayed old batch: expected 2 queues after restart, reloaded 1.
• ConsumerOffsetManagerTest + RocksDBConsumerOffsetManagerTest: 21 tests passed.
• Full broker test suite: 755 tests passed, 0 failures, 0 errors, 4 skipped.
• Maven Checkstyle: 0 violations.
• SpotBugs: 0 findings.
• git diff --check.

[RockteMQ-AI]

Review by github-manager-bot

Summary

This PR fixes a real concurrency race in both ConsumerOffsetManager and RocksDBConsumerOffsetManager where concurrent first-offset commits for the same topic@group could overwrite each other's inner maps, losing queue offsets. The fix uses putIfAbsent for atomic map initialization and synchronized(map) for incremental WAL persistence ordering.

Findings

• [Positive] RocksDBConsumerOffsetManager.java:195-201 — The putIfAbsent + winner-takes-all pattern correctly eliminates the check-then-act race. The original get() → put() sequence could indeed lose queues under concurrent first commits.

• [Positive] RocksDBConsumerOffsetManager.java:207-216 — The synchronized(map) block correctly serializes incremental persistence for the same key, preventing an older WAL batch from being written after a newer one. This is the right granularity — per-key, not global.

• [Positive] Non-incremental path remains lock-free across different topic@group keys, preserving throughput for the common case.

• [Info] RocksDBConsumerOffsetManager.java:203-206 — In the non-incremental path, updateOffset() is called without synchronization. Two concurrent commits for the same key could both call map.put() and both evaluate the "less than store" warning. Since ConcurrentHashMap.put() is atomic, this is functionally safe, but the warning log might be slightly misleading under contention. This matches the original behavior and is not a regression.

• [Info] ConsumerOffsetManager.java — The classic manager gets the same putIfAbsent treatment. Good that both paths are fixed consistently.

• [Positive] Test coverage is thorough — deterministic red tests that demonstrate the race on the old code, plus full broker suite validation (755 tests). The reference to historical unmerged PR #1427 shows good due diligence.

Suggestions

1. Consider adding a brief comment on the synchronized(map) block explaining why we synchronize on the inner map (WAL ordering for same RocksDB key), so future maintainers don't accidentally remove it.

2. The updateDataVersionIfNeeded() helper is a nice extraction. One minor note: in the non-incremental path it's called outside any synchronization, while in the incremental path it's inside synchronized(map). This is correct for the current logic but worth a one-line comment explaining the difference.

Verdict

Solid concurrency fix with good test coverage. The approach is sound and the scope is well-contained.

---

Automated review by github-manager-bot

[RockteMQ-AI]

Summary

Defensive fix with proper validation and test coverage. LGTM.

---

Automated review by github-manager-bot

[ai-yang]

Refreshed this PR onto the current develop (e348efa66) and force-pushed one signed-off commit (03695c870). I also added an inline explanation for the per-key synchronized(map) block: it orders the in-memory update, whole-map snapshot, and WAL write so an older snapshot cannot overwrite newer offsets.

• ConsumerOffsetManagerTest + RocksDBConsumerOffsetManagerTest: 21/21 passed;
• all 10 modules in the targeted broker -am reactor succeeded;
• Checkstyle and SpotBugs reported no findings;
• git diff --check passed.

The production change is +37/-20 lines; +261/-6 lines are deterministic concurrency/persistence tests. Full CI has been retriggered.

@RongtongJin @lizhimins, could you please take a human review when convenient, especially of same-key WAL ordering and the classic/RocksDB v1 initialization paths?

[codecov-commenter]

Codecov Report

❌ Patch coverage is 79.31034% with 6 lines in your changes missing coverage. Please review.
✅ Project coverage is 48.47%. Comparing base (e348efa) to head (03695c8).

Files with missing lines	Patch %	Lines
...broker/config/v1/RocksDBConsumerOffsetManager.java	82.60%	3 Missing and 1 partial ⚠️
.../rocketmq/broker/offset/ConsumerOffsetManager.java	66.66%	1 Missing and 1 partial ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             develop   #10701      +/-   ##
=============================================
- Coverage      48.58%   48.47%   -0.11%     
+ Complexity     13678    13647      -31     
=============================================
  Files           1381     1381              
  Lines         101475   101486      +11     
  Branches       13190    13192       +2     
=============================================
- Hits           49304    49198     -106     
- Misses         46170    46266      +96     
- Partials        6001     6022      +21     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

[RockteMQ-AI]

Summary

This PR modifies 4 files (442 lines).

Key changes reviewed. Please see inline comments for specific suggestions.

---

Automated review by "github-manager-bot"