[ISSUE #6508]♻️Refactor ProcessQueue to use a dedicated store struct for improved organization and clarity

[mxsm]

Which Issue(s) This PR Fixes(Closes)

• Fixes [Refactor♻️] Refactor ProcessQueue to use a dedicated store struct for improved organization and clarity #6508

Brief Description

How Did You Test This Change?

Summary by CodeRabbit

• Refactor

  • Refactored internal message queue state management to improve encapsulation and thread safety.
  • Migrated offset span retrieval to asynchronous operations for enhanced flow control logging.

• Tests

  • Updated unit tests to align with refactored queue state management structure.

[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

The PR refactors ProcessQueue to encapsulate message tree maps and offset tracking within a new ProcessQueueStore struct behind RwLock, improving organization and thread safety. Flow-control logging in the consumer implementation is updated to use an async get_offset_span() helper instead of direct map access.

Changes

Cohort / File(s)	Summary
ProcessQueue Encapsulation rocketmq-client/src/consumer/consumer_impl/process_queue.rs	Introduced ProcessQueueStore struct to encapsulate msg_tree_map, consuming_msg_orderly_tree_map, and queue_offset_max behind RwLock. Refactored all accesses to internal maps to use store.read()/store.write() guards. Changed msg_count from AtomicU64 to AtomicI64. Added async get_offset_span() helper method. Updated unit tests to work with new store-based structure.
Flow-Control Logging Refactor rocketmq-client/src/consumer/consumer_impl/default_mq_push_consumer_impl.rs	Replaced direct msg_tree_map reads at three flow-control logging sites with async get_offset_span().await calls. Conditional logging now uses min/max offsets from get_offset_span() instead of first/last key-value pairs from the map.

Estimated code review effort

🎯 4 (Complex) | ⏱️ ~50 minutes

Poem

🐰 A warren of maps, once scattered and free,
Now nestled in stores, locked safely with glee,
RwLock guards standing tall, letting readers commune,
While writers take turns—oh, what a sweet tune!
Offsets now span through async's swift flight,
ProcessQueue dances, refactored just right! ✨

🚥 Pre-merge checks | ✅ 4 | ❌ 1

❌ Failed checks (1 warning)

Check name	Status	Explanation	Resolution
Docstring Coverage	⚠️ Warning	Docstring coverage is 18.18% which is insufficient. The required threshold is 80.00%.	Write docstrings for the functions missing them to satisfy the coverage threshold.

✅ Passed checks (4 passed)

Check name	Status	Explanation
Description Check	✅ Passed	Check skipped - CodeRabbit’s high-level summary is enabled.
Title check	✅ Passed	The title accurately describes the main change: refactoring ProcessQueue to use a dedicated ProcessQueueStore struct for better organization and encapsulation.
Linked Issues check	✅ Passed	The PR successfully implements the refactoring objective stated in issue #6508 by introducing ProcessQueueStore struct for improved code organization and clarity.
Out of Scope Changes check	✅ Passed	All changes are directly related to the ProcessQueue refactoring goal; the logging improvements in default_mq_push_consumer_impl.rs support the new store-based offset retrieval pattern.

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

---

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/process_queue.rs (1)

428-430: Consider asserting msg_count never goes negative.
If negatives are not expected, a debug_assert can catch underflow early instead of silently clamping.

💡 Possible tweak

 pub fn msg_count(&self) -> u64 {
-    self.msg_count.load(Ordering::Acquire).max(0) as u64
+    let v = self.msg_count.load(Ordering::Acquire);
+    debug_assert!(v >= 0, "msg_count should not be negative");
+    v.max(0) as u64
 }

🤖 Prompt for AI Agents

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

In `@rocketmq-client/src/consumer/consumer_impl/process_queue.rs` around lines 428
- 430, The msg_count method currently clamps a possibly-negative atomic value to
0; instead, load the atomic into a local (e.g., let v =
self.msg_count.load(Ordering::Acquire)), add a debug_assert!(v >= 0, "msg_count
underflow") to catch unexpected negatives in debug builds, then return v as u64
(or cast after verifying non-negative). Modify the msg_count function and
reference the self.msg_count atomic in your change so underflow is detected
early rather than silently clamped.

🤖 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/process_queue.rs`:
- Around line 428-430: The msg_count method currently clamps a possibly-negative
atomic value to 0; instead, load the atomic into a local (e.g., let v =
self.msg_count.load(Ordering::Acquire)), add a debug_assert!(v >= 0, "msg_count
underflow") to catch unexpected negatives in debug builds, then return v as u64
(or cast after verifying non-negative). Modify the msg_count function and
reference the self.msg_count atomic in your change so underflow is detected
early rather than silently clamped.

---

ℹ️ 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 80b0d8d and fa69f4d.

📒 Files selected for processing (2)

• rocketmq-client/src/consumer/consumer_impl/default_mq_push_consumer_impl.rs
• rocketmq-client/src/consumer/consumer_impl/process_queue.rs

[codecov]

Codecov Report

❌ Patch coverage is 63.87097% with 56 lines in your changes missing coverage. Please review.
✅ Project coverage is 42.15%. Comparing base (80b0d8d) to head (fa69f4d).
⚠️ Report is 1 commits behind head on main.

Files with missing lines	Patch %	Lines
...client/src/consumer/consumer_impl/process_queue.rs	70.21%	42 Missing ⚠️
...mer/consumer_impl/default_mq_push_consumer_impl.rs	0.00%	14 Missing ⚠️

Additional details and impacted files

@@            Coverage Diff             @@
##             main    #6509      +/-   ##
==========================================
- Coverage   42.16%   42.15%   -0.01%     
==========================================
  Files         946      946              
  Lines      132130   132145      +15     
==========================================
+ Hits        55708    55710       +2     
- Misses      76422    76435      +13     

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