[Bug] Concurrent first offset commits can overwrite queues in consumer offset managers

[ai-yang]

Affected baseline

develop at 00e45b8a6db23efbe756d0306f10716156cfd4dd.

Problem

Concurrent first offset commits for the same topic@group and different queue IDs can overwrite each other in both the classic JSON ConsumerOffsetManager and the RocksDB v1 manager.

Both implementations currently follow this pattern:

1. Read offsetTable.get(topic@group) and receive null.
2. Each thread creates its own inner queue-offset map.
3. Each writes one queue to its private map.
4. Each unconditionally publishes with offsetTable.put(key, map).

The last outer write replaces the other map, so one queue disappears from memory and subsequent persistence.

RocksDB v1 has a second race in incremental mode. It serializes the entire inner queue map as one RocksDB value. A thread can serialize an older {q0} snapshot, pause, let another thread write {q0,q1}, then write the older batch last. Memory still contains both queues, but a restart reloads only q0.

Expected behavior

• Concurrent first commits for different queue IDs retain every queue in memory and after persistence/reload.
• Classic and RocksDB v1 atomically initialize the shared inner map with outer putIfAbsent.
• RocksDB v1 preserves its LMQ ConcurrentHashMap<>(1, 1.0F) initialization.
• Incremental persistence orders the update, version change, whole-map serialization, and WAL write for the same topic@group while allowing different keys to proceed concurrently.
• RocksDB v2 remains unchanged: it already uses outer putIfAbsent and persists each queue under an independent RocksDB key.

Root cause

The classic and v1 managers use check-then-act initialization rather than atomically publishing one shared map. In v1 incremental mode, whole-map read/modify/serialize/write cycles for the same RocksDB key have no shared critical section.

Deterministic test plan

• Use a barrier-backed outer map whose first two get(key) calls both return their saved null; commit two queue IDs concurrently and verify classic JSON encode/decode retains both.
• Repeat against RocksDB v1 periodic persistence and verify persist -> stop -> clear -> load retains both queues.
• In incremental mode, pause the first batchPutWithWal after its old snapshot has been serialized. Let the second commit either write ahead on current code or block on the fixed per-key monitor, then reload and verify both queues remain.

Related work and scope

Historical unmerged PR #1427 precisely identified and proposed putIfAbsent for the classic-manager first-commit race. It did not cover RocksDB v1, incremental whole-map WAL ordering, or deterministic concurrency/persistence tests, so this issue extends that valid prior analysis rather than claiming the classic root cause is new.

Current open PRs #10625, #9602, and #9877 touch related manager/test files but do not change these commit paths or solve either race. This fix will stay limited to classic initialization, RocksDB v1 initialization/incremental ordering, and their regression tests; v2 and offset-removal behavior are out of scope.

I am working on a focused fix and will submit a PR against develop.

[RockteMQ-AI]

Issue Evaluation

Category: type/bug | Status: Needs Info

Thank you for reporting this issue. To proceed with verification, please provide:

• Version/tag where this occurs
• Steps to reproduce
• Expected vs actual behavior
• Relevant logs or stack traces

---

Automated evaluation by github-manager-bot

[ai-yang]

Thanks for the evaluation. Here are the requested details in an explicit format:

• Version/tag: develop at 00e45b8a6db23efbe756d0306f10716156cfd4dd.
• Steps to reproduce:
  1. Start two concurrent first commitOffset calls for the same topic@group but different queue IDs.
  2. Use a barrier so both calls observe a missing inner offset map.
  3. Let both calls initialize and publish their maps.
  4. Persist and reload the offset manager.
• Actual behavior: One inner map can overwrite the other, so only one queue remains in memory or after reload. In RocksDB v1 incremental mode, an older serialized whole-map WAL batch can also be written after the newer snapshot and overwrite it on disk.
• Expected behavior: Both queue offsets must remain in memory and after persistence/reload.
• Logs or stack traces: This is a silent data race and does not emit an exception or stack trace. The deterministic regression tests against the previous implementation failed with expected: 2 but was: 1 for the classic JSON manager, RocksDB v1 periodic persistence, and RocksDB v1 incremental restart recovery.

The focused fix and deterministic regression tests are available in PR #10701.