[Bug] Broadcast offset initialization can overwrite a concurrent client offset update
- Dominant language
- Java
- Stars
- 22.6k
- Forks
- 12k
- Avg merge
- 3d 1h
- Merged PRs (30d)
- 27
Description
### Before Creating the Bug Report
- [x] I found a bug, not just asking a question, which should be created in [GitHub Discussions](https://github.com/apache/rocketmq/discussions).
- [x] I have searched the [GitHub Issues](https://github.com/apache/rocketmq/issues) and [GitHub Discussions](https://github.com/apache/rocketmq/discussions) of this repository and believe that this is not a duplicate.
- [x] 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:
```text
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.
Contributor guide
Research direction
Start by locating BroadcastOffsetManager and read queryInitOffset alongside updateOffset, focusing on their per-client initialization and offset storage. Run or recreate the latch-controlled broker unit test described in the issue, then verify that a concurrent update to offset 100 remains stored and returned instead of falling back to the queue maximum 10.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- java
- Domain
- distributed-systems
- Issue type
- Bug
- Difficulty
- 3/5
- Estimated time
- 1-2 days
- Activity status
- Quiet
- Clarity
- Clearly specified
- Newbie friendliness
- 72/100