[ISSUE #6529]♻️Refactor pop order consume message handling for improved error handling and clarity

[mxsm]

Which Issue(s) This PR Fixes(Closes)

• Fixes [Refactor♻️] Refactor pop order consume message handling for improved error handling and clarity #6529

Brief Description

How Did You Test This Change?

Summary by CodeRabbit

• Bug Fixes

  • Improved message retry handling with explicit metadata management.
  • Enhanced queue suspension timing logic for orderly message processing.
  • Refined rollback behavior with improved visibility recovery mechanism.

• Refactor

  • Restructured message consumption flow for better error handling and state management.

[rocketmq-rust-bot]

🔊@mxsm 🚀Thanks for your contribution🎉！

💡CodeRabbit(AI) will review your code first🔥！

Note

🚨The code review suggestions from CodeRabbit are to be used as a reference only, and the PR submitter can decide whether to make changes based on their own judgment. Ultimately, the project management personnel will conduct the final code review💥.

[coderabbitai]

Walkthrough

This refactoring improves ordered pop message consumption by replacing direct message mutation with MessageAccessor-based updates, introducing blocking tasks for message processing, implementing explicit context passing throughout the flow, and splitting commit/rollback handling into distinct branches with specialized logic paths.

Changes

Cohort / File(s)	Summary
Message Mutation & Retry Logic rocketmq-client/src/consumer/consumer_impl/consume_message_pop_orderly_service.rs	Replaced direct reconsume_times mutation with MessageAccessor; refactored requeue logic to construct new retry messages with copied properties, origin id, flags, and retry-topic metadata; explicitly sets reconsume_time and max_reconsume_times on retry messages.
Blocking Task Execution & Context Handling rocketmq-client/src/consumer/consumer_impl/consume_message_pop_orderly_service.rs	Introduced blocking_spawned tasks in orderly processing path to execute consume_message; captures MessageExt context, returns both status and updated context; unified error handling to set CMResult::CRThrowException with panic descriptors.
Suspend & Status Management rocketmq-client/src/consumer/consumer_impl/consume_message_pop_orderly_service.rs	Updated SuspendCurrentQueueAMoment handling to compute suspend_time from context; split Commit and Rollback in ConsumeOrderlyStatus into distinct branches with Commit acknowledging messages and Rollback reverting visibility with 1000ms delay; updated process_consume_result signature to actively use context parameter.
Direct Consumption Flow rocketmq-client/src/consumer/consumer_impl/consume_message_pop_orderly_service.rs	Replaced synchronous consumption in consume_message_directly with blocking task evaluation; maps returned status and context to CMResult with remarks; updated ConsumeRequest::run to rely on context from blocking invocation for suspend logic.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~60 minutes

Suggested reviewers

• TeslaRustor
• SpaceXCN
• rocketmq-rust-bot

Poem

🐰 A rabbit hops through message queues with glee,
Blocking tasks now dance in harmony,
Accessors guard mutations tight and true,
Contexts flow where retry paths are new,
Order preserved, clarity in sight! 🚀

🚥 Pre-merge checks | ✅ 2 | ❌ 3

❌ Failed checks (1 warning, 2 inconclusive)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 20.00% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.
Linked Issues check	❓ Inconclusive	The linked issue #6529 provides no specific objectives, target code, problems, or proposed changes to validate against the implementation.	The linked issue lacks concrete requirements. Verify with maintainers that the refactoring achieves the intended improvements in error handling and code clarity.
Out of Scope Changes check	❓ Inconclusive	Without specific objectives in the linked issue, it is unclear whether all changes are in scope or if some modifications extend beyond the intended refactoring.	Define clear scope in the linked issue or provide explicit documentation of the refactoring's intended boundaries and goals.

✅ Passed checks (2 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately reflects the main change: refactoring pop order consume message handling with improved error handling and clarity.

_{✏️ Tip: You can configure your own custom pre-merge checks in the settings.}

✨ Finishing Touches

• 📝 Generate docstrings (stacked PR)
• 📝 Generate docstrings (commit on current branch)

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Post copyable unit tests in a comment
• Commit unit tests in branch feat-6529

---

Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out.

❤️ Share

• X
• Mastodon
• Reddit
• LinkedIn

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[coderabbitai]

🧹 Nitpick comments (2)

rocketmq-client/src/consumer/consumer_impl/consume_message_pop_orderly_service.rs (2)

319-328: Inconsistent suspend time handling for deprecated Rollback status.

The SuspendCurrentQueueAMoment branch (lines 301-305) derives suspend_time from the context, while Rollback uses a hardcoded 1000ms. For consistency and to respect any custom suspend time the listener may have set, consider using the same pattern:

♻️ Suggested change for consistency

             #[allow(deprecated)]
             ConsumeOrderlyStatus::Rollback => {
                 warn!(
                     "Consumer group {} received deprecated Rollback status, reverting messages",
                     self.consumer_group
                 );
+                let suspend_time = if context.get_suspend_current_queue_time_millis() > 0 {
+                    context.get_suspend_current_queue_time_millis() as u64
+                } else {
+                    1000
+                };
                 for msg in msgs {
-                    self.change_invisible_time(msg.as_ref(), 1000).await;
+                    self.change_invisible_time(msg.as_ref(), suspend_time).await;
                 }
                 false
             }

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In
`@rocketmq-client/src/consumer/consumer_impl/consume_message_pop_orderly_service.rs`
around lines 319 - 328, The Rollback branch of ConsumeOrderlyStatus uses a
hardcoded 1000ms suspend time; change it to use the same suspend_time value
derived in the SuspendCurrentQueueAMoment branch (i.e., obtain suspend_time from
the listener/context where SuspendCurrentQueueAMoment does) and call
change_invisible_time(msg.as_ref(), suspend_time). Update the
ConsumeOrderlyStatus::Rollback arm to mirror SuspendCurrentQueueAMoment's
suspend_time retrieval and application so custom suspend times are respected for
self.consumer_group and msgs.

---

301-305: Consider extracting suspend time derivation to reduce duplication.

The same pattern for deriving suspend_time from context appears at lines 301-305 and 607-609. A small helper method could improve clarity and reduce duplication:

♻️ Optional helper extraction

impl ConsumeMessagePopOrderlyService {
    fn get_suspend_time(&self, context: &ConsumeOrderlyContext) -> u64 {
        if context.get_suspend_current_queue_time_millis() > 0 {
            context.get_suspend_current_queue_time_millis() as u64
        } else {
            self.consumer_config.suspend_current_queue_time_millis
        }
    }
}

Then use self.get_suspend_time(&context) in both locations.

🤖 Prompt for AI Agents

Verify each finding against the current code and only fix it if needed.

In
`@rocketmq-client/src/consumer/consumer_impl/consume_message_pop_orderly_service.rs`
around lines 301 - 305, Extract the repeated suspend_time derivation into a
helper on ConsumeMessagePopOrderlyService: create a method fn
get_suspend_time(&self, context: &ConsumeOrderlyContext) -> u64 that returns
context.get_suspend_current_queue_time_millis() as u64 when > 0 otherwise
returns self.consumer_config.suspend_current_queue_time_millis, then replace the
duplicated logic at the sites using suspend_time (the blocks referencing
context.get_suspend_current_queue_time_millis() at lines shown) with
self.get_suspend_time(&context).

🤖 Prompt for all review comments with AI agents

Verify each finding against the current code and only fix it if needed.

Nitpick comments:
In
`@rocketmq-client/src/consumer/consumer_impl/consume_message_pop_orderly_service.rs`:
- Around line 319-328: The Rollback branch of ConsumeOrderlyStatus uses a
hardcoded 1000ms suspend time; change it to use the same suspend_time value
derived in the SuspendCurrentQueueAMoment branch (i.e., obtain suspend_time from
the listener/context where SuspendCurrentQueueAMoment does) and call
change_invisible_time(msg.as_ref(), suspend_time). Update the
ConsumeOrderlyStatus::Rollback arm to mirror SuspendCurrentQueueAMoment's
suspend_time retrieval and application so custom suspend times are respected for
self.consumer_group and msgs.
- Around line 301-305: Extract the repeated suspend_time derivation into a
helper on ConsumeMessagePopOrderlyService: create a method fn
get_suspend_time(&self, context: &ConsumeOrderlyContext) -> u64 that returns
context.get_suspend_current_queue_time_millis() as u64 when > 0 otherwise
returns self.consumer_config.suspend_current_queue_time_millis, then replace the
duplicated logic at the sites using suspend_time (the blocks referencing
context.get_suspend_current_queue_time_millis() at lines shown) with
self.get_suspend_time(&context).

---

ℹ️ Review info

Configuration used: Path: .coderabbit.yaml

Review profile: CHILL

Plan: Pro

📥 Commits

Reviewing files that changed from the base of the PR and between 25d5c5b and e0c257b.

📒 Files selected for processing (1)

• rocketmq-client/src/consumer/consumer_impl/consume_message_pop_orderly_service.rs

[codecov]

Codecov Report

❌ Patch coverage is 0% with 85 lines in your changes missing coverage. Please review.
✅ Project coverage is 42.07%. Comparing base (25d5c5b) to head (e0c257b).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
...nsumer_impl/consume_message_pop_orderly_service.rs	0.00%	85 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #6530      +/-   ##
==========================================
- Coverage   42.08%   42.07%   -0.02%     
==========================================
  Files         947      947              
  Lines      132264   132311      +47     
==========================================
+ Hits        55666    55667       +1     
- Misses      76598    76644      +46     

☔ View full report in Codecov by Sentry.
📢 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.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[rocketmq-rust-bot]

LGTM - All CI checks passed ✅