RocketMQ version
develop at a06836dd564e5e43115493f775626cf98d51d10e
Describe the bug
ConsumeQueue.putMessagePositionInfoWrapper appends a ConsumeQueueExt unit before the main ConsumeQueue checks whether the dispatch has already been applied.
The private putMessagePositionInfo method correctly treats a physical range ending at or before maxPhysicOffset as an idempotent replay and returns success without appending another CQ entry. By that point, however, the wrapper has already called consumeQueueExt.put. The new extension unit is therefore left unreferenced.
When a later valid dispatch is appended, its CQ entry points past the orphan. The orphan becomes an interior unit rather than a tail unit, so normal reload/recovery cannot remove it.
Deterministic reproduction
- Create a file-backed
ConsumeQueue with enableConsumeQueueExt=true.
- Dispatch a first message at physical offset
0 and record its raw CQExt address.
- Replay the exact same dispatch request. The main CQ correctly remains unchanged, but the wrapper allocates a second CQExt unit.
- Dispatch a second valid message at physical offset
100.
- Flush, load another
ConsumeQueue from the same directory, and call recover().
- Read the CQExt unit immediately after the first referenced unit and compare it with the second message's CQ entry.
Expected:
- Replaying a dispatch does not allocate persistent extension data.
- The second valid message uses the extension address immediately following the first message's unit.
- Reload/recovery preserves that address and tag code.
Actual:
- The replay allocates an unreferenced CQExt unit.
- The second valid message points to the following address, leaving the replay unit between two referenced units.
- After reload, the expected second address still contains the first message's tag code, while the second message points past it.
The controlled regression is named:
ConsumeQueueTest#testDuplicateDispatchDoesNotLeaveConsumeQueueExtOrphan
It uses a temporary file-backed queue and no sleeps, randomized scheduling, network access, or external service.
On the latest unmodified develop baseline under JDK 8, the test failed at the same business assertion in 5/5 isolated Maven/JUnit processes: the expected second-message tag code was 200, but the orphan at that address still contained the replayed first-message tag code 100.
Impact
CommitLog recovery and other idempotent dispatch replays can accumulate unreachable ConsumeQueueExt units even though the main ConsumeQueue remains correct. Once a later valid entry references a higher extension address, the orphan is inside the live range and tail recovery cannot reclaim it. Repeated replay therefore causes persistent CQExt space amplification.
Suggested fix
Gate CQExt allocation with the same physical-end condition used by the main CQ idempotency check. The wrapper should still execute the existing success-side checkpoint and multi-dispatch behavior for replayed requests; this should not be implemented as an early return from the wrapper.
Add a regression covering the replay, a later valid append, and reload/recovery.
Related work checked
Immediate pre-publication searches across open and closed issues and open, closed, and merged pull requests for duplicate dispatch, putMessagePositionInfoWrapper, CQExt allocation, and orphaned extension data found no equivalent report, claimed implementation, or assignee.
RocketMQ version
developata06836dd564e5e43115493f775626cf98d51d10eDescribe the bug
ConsumeQueue.putMessagePositionInfoWrapperappends aConsumeQueueExtunit before the main ConsumeQueue checks whether the dispatch has already been applied.The private
putMessagePositionInfomethod correctly treats a physical range ending at or beforemaxPhysicOffsetas an idempotent replay and returns success without appending another CQ entry. By that point, however, the wrapper has already calledconsumeQueueExt.put. The new extension unit is therefore left unreferenced.When a later valid dispatch is appended, its CQ entry points past the orphan. The orphan becomes an interior unit rather than a tail unit, so normal reload/recovery cannot remove it.
Deterministic reproduction
ConsumeQueuewithenableConsumeQueueExt=true.0and record its raw CQExt address.100.ConsumeQueuefrom the same directory, and callrecover().Expected:
Actual:
The controlled regression is named:
It uses a temporary file-backed queue and no sleeps, randomized scheduling, network access, or external service.
On the latest unmodified
developbaseline under JDK 8, the test failed at the same business assertion in 5/5 isolated Maven/JUnit processes: the expected second-message tag code was200, but the orphan at that address still contained the replayed first-message tag code100.Impact
CommitLog recovery and other idempotent dispatch replays can accumulate unreachable ConsumeQueueExt units even though the main ConsumeQueue remains correct. Once a later valid entry references a higher extension address, the orphan is inside the live range and tail recovery cannot reclaim it. Repeated replay therefore causes persistent CQExt space amplification.
Suggested fix
Gate CQExt allocation with the same physical-end condition used by the main CQ idempotency check. The wrapper should still execute the existing success-side checkpoint and multi-dispatch behavior for replayed requests; this should not be implemented as an early return from the wrapper.
Add a regression covering the replay, a later valid append, and reload/recovery.
Related work checked
confirmOffset.Immediate pre-publication searches across open and closed issues and open, closed, and merged pull requests for duplicate dispatch,
putMessagePositionInfoWrapper, CQExt allocation, and orphaned extension data found no equivalent report, claimed implementation, or assignee.