[ISSUE #10756] Fix duplicate dispatch ConsumeQueueExt leak

[ai-yang]

Which Issue(s) This PR Fixes

Fixes #10756

Brief Description

ConsumeQueue.putMessagePositionInfoWrapper() allocated a ConsumeQueueExt unit before the main CQ idempotency check. Replaying a duplicate dispatch therefore skipped the CQ append but still left an unreferenced Ext unit, which a later valid dispatch could seal inside the live Ext range.

This change reuses the main CQ physical-end idempotency predicate before allocating an Ext unit. Duplicate replay still follows the existing successful wrapper path, so checkpoint advancement and multi-dispatch behavior are preserved while the orphan allocation is avoided.

How Did You Test This Change?

• Unmodified develop: the deterministic duplicate-dispatch regression failed in 5/5 isolated JDK 8 Maven processes.
• Fixed targeted regression: 20 isolated Maven/JVM processes, 1/1 each (20/20 total).
• Complete ConsumeQueueTest: 11/11.
• Full store -am test: common 241/241, remoting 174/174, and store 314 tests with 4 skips, 0 failures, and 0 errors.
• Maven validate and Checkstyle: 0 violations.
• SpotBugs: 0 bugs/errors across all four reactor modules.
• git diff --check: passed.

Scope and Concurrency

The new guard reuses the physical-end idempotency predicate already applied by putMessagePositionInfo() on the same dispatch attempt. It introduces no new shared state or locking and leaves supported writes, checkpoint advancement, retries, and multi-dispatch handling unchanged. This PR is scoped to duplicate re-dispatch; cleanup after a genuine Ext-append/CQ-append failure remains a separate failure mode.

Applicability and user-visible impact

enableConsumeQueueExt defaults to false. The affected path requires CQExt writes and duplicate Broker dispatch/recovery of an already-indexed physical message. This is internal index-build idempotency, not producer retry deduplication or an exactly-once delivery feature.

The demonstrated impact is unreferenced extension allocation and extra storage retained in the live Ext range. It is not evidence of lost message bodies or permanent retention under every cleanup policy. Total disk cost and eventual reclamation depend on duplicate volume, mapped-file boundaries and normal retention cleanup; no production byte estimate or occurrence rate is claimed.

[RockteMQ-AI]

Review by github-manager-bot

Summary

Prevents ConsumeQueueExt orphan entries when duplicate dispatch requests arrive by extracting isDispatchAlreadyApplied() and applying it consistently in both the ext-write guard and the main dispatch path.

Findings

• [Info] ConsumeQueue.java:730 — The new isDispatchAlreadyApplied() extraction is clean and ensures both putMessagePositionInfoWrapper and putMessagePositionInfo use the same duplicate check. Good consolidation.
• [Info] ConsumeQueue.java:901 — The helper method is well-named and the semantics are clear (offset + size <= maxPhysicOffset means the dispatch was already applied).
• [Info] ConsumeQueueTest.java — Test testDuplicateDispatchDoesNotLeaveConsumeQueueExtOrphan directly verifies the fix scenario with ext address validation after reload. Solid coverage.

Suggestions

• Consider adding a brief Javadoc on isDispatchAlreadyApplied() explaining that it detects duplicate dispatch after broker crash recovery, to help future readers understand why the check matters in both call sites.

---

Automated review by github-manager-bot

[RockteMQ-AI]

Summary

Fix for ConsumeQueueExt orphan entries during duplicate dispatch — extracts isDispatchAlreadyApplied() for consistent use in both the ext-write guard and the main dispatch path. Test coverage is solid.

The previous suggestion about Javadoc was informational only. The code is correct and the fix is well-targeted. 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]

@RongtongJin @guyinyou, could you please take a human review when convenient? This is a focused Store/ConsumeQueueExt fix that prevents duplicate dispatch from allocating an orphan Ext unit. The regression reproduces the old behavior deterministically; the complete Store reactor, Checkstyle, and SpotBugs pass. The remaining GitHub Actions runs are currently awaiting maintainer approval.

[ai-yang]

Refreshed this PR against the latest develop (e348efa66b08eb645ee123706ea6492fa9a3ad35, RocketMQ 5.5.1).

• refreshed head: d623f0dcc (one signed-off commit);
• complete ConsumeQueueTest: 11/11 passed;
• Checkstyle, SpotBugs, and git diff --check passed in the affected reactor build.

The force-push has retriggered the full CI matrix.

[ai-yang]

CI triage update:

The two failing checks are outside this PRs changed Store/ConsumeQueue path:

• Maven: TimerMessageStoreTest#testTimerFlowControl;
• coverage: DefaultLitePullConsumerTest#testSubscribe_BroadcastPollMessageSuccess, with Mockito/async-test pollution symptoms.

The affected ConsumeQueueTest passes 11/11, while the other 8 checks passed. I found no evidence that these failures require a code change in this PR. A maintainer rerun of the failed workflows would be appreciated.

[ai-yang]

Refreshed onto develop at 1a50c6e4e and pushed signed-off commit f9247d2a6.

The rebase retains upstream #10566's mapped-file hold()/release() protection in ConsumeQueue. The duplicate-dispatch guard is unchanged (git range-diff reports the patch as equivalent).

Clean JDK 8 verification:

mvn -pl store -am -DskipITs -Dtest=ConsumeQueueTest,ConsumeQueueExtTest \
  -Dsurefire.failIfNoSpecifiedTests=false clean test

24 tests passed: store.ConsumeQueueTest 11/11, store.ConsumeQueueExtTest 6/6, and store.queue.ConsumeQueueTest 7/7. All 4 reactor modules succeeded, with Checkstyle and SpotBugs reporting no findings. git diff --check passed.

This push starts fresh CI against the current base; the earlier Timer/LitePull failures are historical results, not validation of this refreshed head.

@guyinyou @xdkxlk, could you review the early idempotency check and its interaction with CQ/Ext dispatch when convenient?

[ai-yang]

New-head CI triage for f9247d2a6:

• Bazel run https://github.com/apache/rocketmq/actions/runs/34596301031 completed 595 test targets: 594 passed, with only PopPriorityIT failing.
• The downloaded bazel-test-results XML identifies test_priority_consume_always_high_priority[kv=false, ascending=true], line 131: expected priority 7, received 6. The other parameter combinations of this method passed.
• The local CQ/Ext regression reactor passed 24/24. The same-base Bazel workflow for [ISSUE #10696] Fix async request future cleanup on synchronous send failure #10697 also passed: https://github.com/apache/rocketmq/actions/runs/34596300805.

This points to the priority-consumption integration path, but it does not by itself establish a flaky test or exclude interaction with this change: the integration fixture enables ConsumeQueueExt. The full stdout was omitted from the job log because it exceeded the output-size limit, so the available XML only establishes the assertion above.

I attempted to rerun the Bazel workflow, but GitHub rejected the author-side request with run cannot be rerun. Could a maintainer rerun this workflow on the same head? If the failure repeats, the next check should compare this exact priority case on the base and PR head before changing production code. The remaining CI jobs are still running.

[codecov-commenter]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 49.29%. Comparing base (1a50c6e) to head (f9247d2).

Additional details and impacted files

@@              Coverage Diff              @@
##             develop   #10759      +/-   ##
=============================================
- Coverage      49.36%   49.29%   -0.08%     
+ Complexity     14230    14216      -14     
=============================================
  Files           1390     1390              
  Lines         103123   103124       +1     
  Branches       13484    13484              
=============================================
- Hits           50904    50831      -73     
- Misses         46061    46117      +56     
- Partials        6158     6176      +18     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.