Repository navigation
Conversation
…acked Signed-off-by: zhang hongyuan <100520587+Metastarx@users.noreply.github.com>
RockteMQ-AI
left a comment
There was a problem hiding this comment.
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 onmsgsbefore creating theConsumeRequest. This prevents NPEs from upstream callers. -
[Info]
ConsumeMessagePopOrderlyService.java(newConsumeRequestinner class) — The refactoredConsumeRequestnow carriesmsgs,popTime,invisibleTime,brokerName,queueIdfrom the POP checkpoint. This is the correct approach to propagate POP metadata through the consume pipeline. -
[Warning] The
consumeMessagemethod insideConsumeRequest.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 (
mergeBatchesor 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 fromsubmitConsumeRequest(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
-
Extract outcome handling: The switch/if-else block handling
ConsumeOrderlyStatusoutcomes (SUCCESS, COMMIT, ROLLBACK, SUSPEND) could be extracted into aprocessConsumeResult(ConsumeOrderlyStatus, ...)method. This would make the code easier to follow and test. -
Logging: Consider adding a
log.warnwhen messages are dropped due to queue destruction (processQueue.isDropped()), so operators can detect this condition in production logs. -
Javadoc: The new
ConsumeRequestconstructor andrun()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 Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## develop #11480 +/- ##
=============================================
+ Coverage 49.79% 49.82% +0.03%
- Complexity 14387 14419 +32
=============================================
Files 1390 1391 +1
Lines 103264 103566 +302
Branches 13504 13566 +62
=============================================
+ Hits 51419 51604 +185
- Misses 45657 45713 +56
- Partials 6188 6249 +61 ☔ View full report in Codecov by Harness. 🚀 New features to boost your workflow:
|
RockteMQ-AI
left a comment
There was a problem hiding this comment.
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 onmsgsbefore constructing theConsumeRequest. Prevents NPE from upstream callers. - [Info]
ConsumeMessagePopOrderlyService.java(batch merging) — UsingConcurrentLinkedQueuefor merging batches that arrive while a request is running is a clean approach. The atomic removal viaconsumeRequestSet.remove(this)in thefinallyblock ensures successor requests are not accidentally unregistered. - [Info]
ConsumeMessagePopOrderlyService.java(pop timeout handling) — Checkingbatch.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
ArgumentCaptorto 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@VisibleForTestingannotation 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
| for (MessageExt msg : msgs) { | ||
| MessageAccessor.setConsumeStartTimeStamp(msg, String.valueOf(System.currentTimeMillis())); | ||
| } | ||
| status = messageListener.consumeMessage(Collections.unmodifiableList(msgs), context); |
There was a problem hiding this comment.
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.
There was a problem hiding this comment.
Done in 2f7c66c: each POP result is now consumed in consumeMessageBatchMaxSize sub-batches, in queue order, so a listener that receives one message can no longer acknowledge the rest of the result.
Split each POP result by consumeMessageBatchMaxSize and consume the sub-batches in order so a listener never acknowledges messages it did not receive. A suspended sub-batch stops later sub-batches and keeps their broker-side retry schedule, while earlier successful sub-batches stay acknowledged. Signed-off-by: zhang hongyuan <100520587+Metastarx@users.noreply.github.com>
|
|
||
| // POP mode has no local offset to advance. COMMIT and SUCCESS both mean that the broker | ||
| // can forget the popped copy, while ROLLBACK and SUSPEND keep it available for redelivery. | ||
| if (ConsumeOrderlyStatus.SUCCESS == status || ConsumeOrderlyStatus.COMMIT == status) { |
There was a problem hiding this comment.
Could we preserve messages when context.isAutoCommit() is false instead of ACKing SUCCESS here? Manual-commit listeners intentionally return SUCCESS before a later COMMIT or ROLLBACK, so this branch acknowledges their messages prematurely and a subsequent rollback or consumer failure can no longer redeliver them.
There was a problem hiding this comment.
Done in 5121f08: SUCCESS no longer acks when the listener manages commits itself, so a later ROLLBACK or a consumer failure can still redeliver; COMMIT still acks explicitly.
SUCCESS only means the listener finished; when it manages commits itself the client must not ack, otherwise a later ROLLBACK or a consumer failure can no longer redeliver the popped copy. COMMIT still acks explicitly.
18a45a5 to
5121f08
Compare
Which Issue(s) This PR Fixes
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)