[ISSUE #10734] Fix broadcast offset initialization race

[ai-yang]

Which Issue(s) This PR Fixes

• Fixes [Bug] Broadcast offset initialization can overwrite a concurrent client offset update #10734

Brief Description

BroadcastOffsetManager.queryInitOffset initialized clientOffsetStore with a non-atomic get / create / put sequence. If updateOffset inserted and updated the same client's store between the get and put, the query thread replaced that store and discarded the newly committed offset.

Use computeIfAbsent so initialization and concurrent updates retain the same winning BroadcastTimedOffsetStore instance. The change is internal and does not alter the public protocol.

How Did You Test This Change?

• Added a deterministic latch-controlled concurrency regression test with no sleeps or random scheduling.
• Reverted to the previous implementation: the regression test failed 5/5 times (expected 100 but was 10).
• Applied this fix: the regression test passed 20/20 times.
• Ran the complete affected JDK 8 reactor: all modules passed; broker ran 753 tests with 0 failures, 0 errors, and 4 skipped tests.
• Ran Maven validate: Checkstyle reported 0 violations across the 10 affected modules.
• Ran broker SpotBugs: BugInstance=0, Error=0.
• Ran git diff --check successfully.

Concurrency and Test Design

computeIfAbsent only makes publication of the per-client store atomic; it does not change the existing update semantics once a store exists. The regression uses latches to force the exact missing-key initialization/update interleaving, with no sleeps or random scheduling: the previous get/create/put sequence loses the updated offset, while the atomic initialization retains it.

[RockteMQ-AI]

Review by github-manager-bot

Summary

Fixes a TOCTOU race condition in BroadcastOffsetManager.queryInitOffset where a non-atomic get/create/put sequence could discard a concurrently committed broadcast offset.

Findings

• [Info] BroadcastOffsetManager.java:88-89 — Replacing the manual get/if null/put with computeIfAbsent is the correct and idiomatic fix. ConcurrentHashMap.computeIfAbsent guarantees atomic initialization, preventing the race where updateOffset inserts a store between the get and put.
• [Info] BroadcastOffsetManagerConcurrencyTest.java — The BlockingInitOffsetStore design is excellent. By overriding both put and computeIfAbsent with CountDownLatch-controlled blocking, the test deterministically reproduces the race without relying on sleeps or random scheduling. The test verifies:
  1. With the old code: the race causes offset loss (test fails 5/5)
  2. With the fix: the offset is correctly preserved (test passes 20/20)

Suggestions

No issues found. The fix is a single-line change that eliminates the race, and the test is a model for deterministic concurrency testing.

• Correctness: computeIfAbsent is the standard pattern for atomic get-or-create. No edge cases missed.
• Performance: computeIfAbsent is actually more efficient than the previous get+put (one atomic operation vs two).
• Compatibility: Internal change only, no public API or protocol change.

LGTM.

---

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]

@lizhimins @xdkxlk, could you please take a human review when convenient? This is a focused BroadcastOffsetManager atomic-initialization race fix with a deterministic latch-controlled regression. The complete affected reactor, Checkstyle, and SpotBugs pass. The remaining GitHub Actions runs are currently awaiting maintainer approval.