[Metastarx]

Which Issue(s) This PR Fixes

• Fixes client/src/main/java/org/apache/rocketmq/client/impl/consumer/ConsumeMessagePopOrderlyService.java:315 - Pop-mode orderly consumers never consume or ack messages #11479

Brief Description

POP-mode orderly consumers were silently dropping every delivered batch: the submission path built a request without the popped messages, and the request returned immediately after taking the queue lock, so the listener was never called, no acknowledgement or invisible-time update was sent, and the request stayed registered until flow control stalled the queue. This change carries the message batches and their POP checkpoint metadata into the request, drains them in order under the existing per-queue lock, invokes the orderly listener, and handles success, commit, rollback, suspend, timeout and dropped-queue outcomes with the matching ack or invisible-time update. It also removes the request atomically when it finishes and merges batches that arrive while a request is running instead of dropping them.

How Did You Test This Change?

mvn test -pl client - passed (Tests run: 613, Failures: 0, Errors: 0, Skipped: 8)

[Copilot]

Copilot was unable to review this pull request because the user who requested the review has reached their quota limit.

[RockteMQ-AI]

Review by github-manager-bot

Summary

This PR fixes a critical bug where POP-mode orderly consumers silently dropped every delivered batch. The submission path was building requests without the popped messages, causing the listener to never be called, no ack/invisible-time update to be sent, and flow control to eventually stall the queue. The fix properly carries message batches and POP checkpoint metadata into the consume request, drains them in order under the per-queue lock, and handles all consume outcomes correctly.

Findings

• [Info] ConsumeMessagePopOrderlyService.java:170-175 — Good defensive null/empty check on msgs before creating the ConsumeRequest. This prevents NPEs from upstream callers.

• [Info] ConsumeMessagePopOrderlyService.java (new ConsumeRequest inner class) — The refactored ConsumeRequest now carries msgs, popTime, invisibleTime, brokerName, queueId from the POP checkpoint. This is the correct approach to propagate POP metadata through the consume pipeline.

• [Warning] The consumeMessage method inside ConsumeRequest.run() is quite long (~150 lines). Consider extracting the ack/invisible-time update logic into a separate helper method (e.g., handleConsumeResult()) to improve readability and testability. The current structure makes it harder to unit-test individual outcome paths.

• [Info] The batch merging logic (mergeBatches or equivalent) when a request is already running is a nice touch — it avoids dropping messages that arrive during processing instead of re-submitting a new request.

• [Warning] The submitConsumeRequest(req) call changed from submitConsumeRequest(req, false) — please verify that removing the boolean parameter does not affect the non-POP path (submitConsumeRequest(List<MessageExt>, ProcessQueue, MessageQueue)). If the old second parameter controlled retry behavior, ensure the POP path has equivalent retry semantics.

• [Info] Test coverage looks comprehensive — 263 lines of new tests covering success, commit, rollback, suspend, timeout, and dropped-queue scenarios. Good alignment with the production code paths.

Suggestions

1. Extract outcome handling: The switch/if-else block handling ConsumeOrderlyStatus outcomes (SUCCESS, COMMIT, ROLLBACK, SUSPEND) could be extracted into a processConsumeResult(ConsumeOrderlyStatus, ...) method. This would make the code easier to follow and test.

2. Logging: Consider adding a log.warn when messages are dropped due to queue destruction (processQueue.isDropped()), so operators can detect this condition in production logs.

3. Javadoc: The new ConsumeRequest constructor and run() method would benefit from class-level Javadoc explaining the POP-specific lifecycle (popTime, invisibleTime, ack).

Overall

This is a well-structured fix for a serious data-loss bug. The approach of carrying POP metadata through the existing consume pipeline is clean. The test coverage is thorough. The main suggestion is to improve readability by extracting the outcome handling logic.

---

Automated review by github-manager-bot

[codecov-commenter]

Codecov Report

❌ Patch coverage is 61.57407% with 83 lines in your changes missing coverage. Please review.
✅ Project coverage is 49.89%. Comparing base (57c2dc0) to head (2f7c66c).
⚠️ Report is 8 commits behind head on develop.

Files with missing lines	Patch %	Lines
...impl/consumer/ConsumeMessagePopOrderlyService.java	61.57%	63 Missing and 20 partials ⚠️

Additional details and impacted files

@@              Coverage Diff              @@
##             develop   #11480      +/-   ##
=============================================
+ Coverage      49.79%   49.89%   +0.10%     
- Complexity     14387    14426      +39     
=============================================
  Files           1390     1391       +1     
  Lines         103264   103545     +281     
  Branches       13504    13553      +49     
=============================================
+ Hits           51419    51663     +244     
+ Misses         45657    45646      -11     
- Partials        6188     6236      +48     

☔ 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.

[RockteMQ-AI]

Review by github-manager-bot

Summary

Fixes POP-mode orderly consumption where messages were silently dropped — the submission path built a request without carrying the popped messages, so the listener was never invoked and no ack/invisible-time update was sent. This change properly carries message batches and POP checkpoint metadata into the consume request, drains them in order, and handles all outcomes (success, commit, rollback, suspend, timeout, dropped queue) with the matching ack or invisible-time update.

Findings

• [Info] ConsumeMessagePopOrderlyService.java:170-175 — Good defensive null/empty check on msgs before constructing the ConsumeRequest. Prevents NPE from upstream callers.
• [Info] ConsumeMessagePopOrderlyService.java (batch merging) — Using ConcurrentLinkedQueue for merging batches that arrive while a request is running is a clean approach. The atomic removal via consumeRequestSet.remove(this) in the finally block ensures successor requests are not accidentally unregistered.
• [Info] ConsumeMessagePopOrderlyService.java (pop timeout handling) — Checking batch.isPopTimeout() after consumption and skipping ack is correct — timed-out messages should not be acknowledged as they will be redelivered.
• [Info] Test coverage is comprehensive — 10+ test cases covering all consume statuses (SUCCESS, COMMIT, ROLLBACK, SUSPEND), timeout, dropped queue, batch merging, late removal safety, and max reconsume times. The use of ArgumentCaptor to capture submitted runnables is a clean testing pattern.

Suggestions

• Consider adding a brief Javadoc on the ConsumeRequest.run() method explaining the batch-draining loop and the merge-on-arrival semantics, as this is the core fix and future maintainers will benefit from understanding the design intent.
• The abortRemainingBatches() method accessed via reflection in tests (testLateRemovalKeepsSuccessorRequest) suggests this is an important safety mechanism. Consider making it package-private with a @VisibleForTesting annotation instead of reflection, if the project conventions allow it.

Overall this is a solid fix with excellent test coverage. The changes follow the established patterns from the non-POP orderly service (ConsumeMessageOrderlyService), which makes the code consistent and easier to reason about.

---

Automated review by github-manager-bot

[yx9o]

Could we split the POP result by consumeMessageBatchMaxSize and consume the sub-batches in order? With the default batch size of 1, a listener may process only msgs.get(0), but returning SUCCESS here acknowledges the entire POP result.