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

[ai-yang]

Before Creating the Bug Report

• I found a bug, not just asking a question, which should be created in GitHub Discussions.
• I have searched the GitHub Issues and GitHub Discussions of this repository and believe that this is not a duplicate.
• I have confirmed that this bug belongs to the current repository, not other repositories of RocketMQ.

Runtime platform environment

Linux x86_64. The reproducer is a deterministic broker unit test and does not require a running RocketMQ cluster.

RocketMQ version

Branch: develop

Commit: 00e45b8a6db23efbe756d0306f10716156cfd4dd

JDK Version

OpenJDK 8

Describe the Bug

BroadcastOffsetManager.queryInitOffset initializes a missing client entry with a non-atomic get / create / put sequence. updateOffset initializes and updates the same entry atomically with ConcurrentHashMap.compute.

If an update wins after the query observes a missing entry but before the query performs its put, the query replaces the updated BroadcastTimedOffsetStore with a new empty instance. The just-committed client offset is lost, and the query falls back to the consumer offset or queue maximum.

Steps to Reproduce

1. Create an existing BroadcastOffsetData with no entry for a client.
2. Start queryInitOffset for that client and block it after it observes the missing entry but before its map insertion.
3. Call updateOffset for the same client and queue with offset 100; this inserts a store and records 100.
4. Resume the query insertion.
5. Read the query result and the stored client offset.

A latch-controlled ConcurrentHashMap test reproduces this ordering without sleeps or random stress. On the unmodified commit above, it failed 5/5 times with:

expected:<100> but was:<10>

Here 10 is the mocked queue maximum used after the updated store has been overwritten.

What Did You Expect to See?

The winning client store and its offset 100 should remain in clientOffsetStore. A concurrent initialization must not overwrite an offset already recorded by updateOffset.

What Did You See Instead?

The query's later put replaces the store created by updateOffset. Both the initial query and a subsequent query observe the fallback offset 10, so the client update to 100 is lost.

Additional Context

The race is scoped to the per-client initialization in BroadcastOffsetManager; no public protocol change is required. I am willing to submit a focused fix with the deterministic regression test.

[RockteMQ-AI]

Issue Evaluation

Category: type/bug | Status: Confirmed

The reported issue has been verified against the current codebase.

Root Cause: BroadcastOffsetManager.queryInitOffset uses a non-atomic get → create → put sequence to initialize a missing client entry. Meanwhile, updateOffset uses ConcurrentHashMap.compute for atomic initialization and update. When updateOffset completes between the query's get (observing absence) and put (inserting a new empty store), the query's put overwrites the store that already contains the committed offset, causing the client offset to be lost and falling back to the consumer offset or queue maximum.

Impact: Broadcast consumer offset tracking can silently lose committed offsets under concurrent initialization, leading to message re-delivery or offset regression.

Severity: Medium — scoped to per-client initialization race in BroadcastOffsetManager; no public protocol change required.

Suggested Fix Direction: Replace the non-atomic get/create/put in queryInitOffset with ConcurrentHashMap.computeIfAbsent (or compute) to ensure atomic initialization, consistent with how updateOffset already operates.

An automated fix proposal will be generated. Reply /approve to proceed with PR generation.

---

Automated evaluation by github-manager