Skip to content

[ISSUE #10734] Fix broadcast offset initialization race - #10735

Open
ai-yang wants to merge 1 commit into
apache:developfrom
ai-yang:agent/fix-broadcast-offset-init-race
Open

[ISSUE #10734] Fix broadcast offset initialization race#10735
ai-yang wants to merge 1 commit into
apache:developfrom
ai-yang:agent/fix-broadcast-offset-init-race

Conversation

@ai-yang

@ai-yang ai-yang commented Aug 1, 2026

Copy link
Copy Markdown

Which Issue(s) This PR Fixes

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.

Signed-off-by: Rui <1685901819@qq.com>
@ai-yang
ai-yang marked this pull request as ready for review August 1, 2026 04:11

@RockteMQ-AI RockteMQ-AI left a comment

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

[Bug] Broadcast offset initialization can overwrite a concurrent client offset update

2 participants