[ISSUE #6511]♻️Refactor consume message handling and concurrency management for improved clarity and performance

[mxsm]

Which Issue(s) This PR Fixes(Closes)

• Fixes [Refactor♻️] Refactor consume message handling and concurrency management for improved clarity and performance #6511

Brief Description

How Did You Test This Change?

Summary by CodeRabbit

• Refactor
  • Optimized ordered message consumption handling with improved status and result processing.
  • Refactored lock synchronization mechanisms and adjusted locking semantics for consistency.
  • Enhanced error handling with structured fallback mechanisms during message listener execution.
  • Improved task execution patterns for better resource management and thread safety.

[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 PR refactors message consumption handling and lock management in the RocketMQ client consumer. The primary changes involve replacing direct listener invocation with spawned blocking tasks, adjusting status code mappings for auto-commit and manual-commit scenarios, updating lock types from a custom wrapper to standard Tokio RwLock, and refining concurrency synchronization semantics.

Changes

Cohort / File(s)	Summary
Consume Message Orderly Service Refactor rocketmq-client/src/consumer/consumer_impl/consume_message_orderly_service.rs	Refactored message listener invocation to use spawned blocking tasks instead of direct calls. Updated status code handling for both auto-commit and manual-commit paths: auto-commit now treats only Success as direct success; Rollback and Commit both commit after warning. Non-auto-commit remapped Commit to commit and Rollback to rollback with late re-consume. Introduced AcqRel atomic ordering for active task counter adjustments. Updated TaskGuard to hold shared reference without cloning. Optimized repeated message_model lookups and added explicit consume_lock release timing.
Lock Type Standardization rocketmq-client/src/consumer/consumer_impl/process_queue.rs, rocketmq-client/src/consumer/consumer_impl/re_balance/rebalance_push_impl.rs	Replaced custom RocketMQTokioRwLock with standard tokio::sync::RwLock for consume_lock across ProcessQueue. Updated rebalance_push_impl to use Tokio timeout wrapping write() future instead of try_write_timeout helper, changing lock acquisition from Result to Option guard pattern.

Estimated Code Review Effort

🎯 4 (Complex) | ⏱️ ~50 minutes

Poem

🐰 Hops through async tasks, oh what a sight!
Messages dance in spawned threads so bright,
Locks now align in harmony true,
AcqRel atomic guards the queue through and through! 🔒✨

🚥 Pre-merge checks | ✅ 3 | ❌ 2

❌ Failed checks (1 warning, 1 inconclusive)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 40.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	Linked issue #6511 provides no concrete objectives, requirements, or acceptance criteria, only a generic refactoring goal title, making it impossible to validate whether code changes satisfy specific requirements.	The linked issue lacks detailed requirements. Confirm that the refactoring improves code clarity and concurrency management as intended, and document the specific goals and acceptance criteria.

✅ Passed checks (3 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title directly references ISSUE #6511 and accurately describes the main refactoring changes across the codebase: improving message consumption handling, concurrency management, and lock semantics.
Out of Scope Changes check	✅ Passed	All changes are directly related to consume message handling and concurrency management: lock type replacements, timeout handling, task spawning for listeners, and active task counter synchronization.

_{✏️ 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 refactor-6511

---

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 (1)

rocketmq-client/src/consumer/consumer_impl/re_balance/rebalance_push_impl.rs (1)

125-130: Consider releasing the consume_lock guard before async I/O if it’s only a try-lock check.

consume_lock stays alive across persist/remove_offset/unlock, which can hold the queue lock during network I/O. If the intent is only to check lock availability, drop it before the awaits; otherwise, add a short comment to clarify the deliberate long hold.

♻️ Optional refactor (release guard before async I/O)

-        let consume_lock = tokio::time::timeout(Duration::from_millis(500), pq.consume_lock.write())
-            .await
-            .ok();
-        if force_unlock || consume_lock.is_some() {
+        let consume_lock = tokio::time::timeout(Duration::from_millis(500), pq.consume_lock.write())
+            .await
+            .ok();
+        let lock_acquired = consume_lock.is_some();
+        drop(consume_lock);
+        if force_unlock || lock_acquired {
             let Some(offset_store) = default_mqpush_consumer_impl.offset_store.as_mut() else {
                 error!("Offset store not initialized");
                 return 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/re_balance/rebalance_push_impl.rs`
around lines 125 - 130, The consume_lock guard returned by
pq.consume_lock.write() is held across subsequent async I/O calls
(persist/remove_offset/unlock) — either drop the guard immediately after using
it as a mere availability check or explicitly document why the guard must be
held across awaits; locate the check around pq.is_dropped(),
UNLOCK_DELAY_TIME_MILLS and the consume_lock variable in rebalance_push_impl.rs
(where consume_lock is assigned from tokio::time::timeout(...
pq.consume_lock.write() ...)) and: if the intent is only to test lock
availability, release (drop) consume_lock before calling
persist/remove_offset/unlock so the queue lock is not held during network I/O,
otherwise add a short comment above the block explaining the deliberate
long-lived lock and why awaits are safe while holding it.

🤖 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/re_balance/rebalance_push_impl.rs`:
- Around line 125-130: The consume_lock guard returned by
pq.consume_lock.write() is held across subsequent async I/O calls
(persist/remove_offset/unlock) — either drop the guard immediately after using
it as a mere availability check or explicitly document why the guard must be
held across awaits; locate the check around pq.is_dropped(),
UNLOCK_DELAY_TIME_MILLS and the consume_lock variable in rebalance_push_impl.rs
(where consume_lock is assigned from tokio::time::timeout(...
pq.consume_lock.write() ...)) and: if the intent is only to test lock
availability, release (drop) consume_lock before calling
persist/remove_offset/unlock so the queue lock is not held during network I/O,
otherwise add a short comment above the block explaining the deliberate
long-lived lock and why awaits are safe while holding it.

---

ℹ️ 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 2f92108 and ff5773b.

📒 Files selected for processing (3)

• rocketmq-client/src/consumer/consumer_impl/consume_message_orderly_service.rs
• rocketmq-client/src/consumer/consumer_impl/process_queue.rs
• rocketmq-client/src/consumer/consumer_impl/re_balance/rebalance_push_impl.rs

[codecov]

Codecov Report

❌ Patch coverage is 2.63158% with 37 lines in your changes missing coverage. Please review.
✅ Project coverage is 42.15%. Comparing base (2f92108) to head (ff5773b).
⚠️ Report is 2 commits behind head on main.

Files with missing lines	Patch %	Lines
...r/consumer_impl/consume_message_orderly_service.rs	0.00%	34 Missing ⚠️
...er/consumer_impl/re_balance/rebalance_push_impl.rs	0.00%	3 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #6512      +/-   ##
==========================================
- Coverage   42.15%   42.15%   -0.01%     
==========================================
  Files         946      946              
  Lines      132145   132158      +13     
==========================================
+ Hits        55710    55711       +1     
- Misses      76435    76447      +12     

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