[ISSUE #6533]🚀Add ConsumerStatsManager and related statistics functionality for improved consumer metrics tracking#6534
Merged
rocketmq-rust-bot merged 2 commits intomainfrom Feb 27, 2026
Merged
Conversation
…nality for improved consumer metrics tracking
…nality for improved consumer metrics tracking
Collaborator
|
🔊@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💥. |
rocketmq-rust-bot
added
AI review first
rocketmq-rust-bot
requested review from
SpaceXCN,
TeslaRustor and
rocketmq-rust-bot
Contributor
WalkthroughThis pull request introduces consumer statistics management capabilities to the RocketMQ client. A new Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~22 minutes Poem
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches
🧪 Generate unit tests (beta)
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. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Actionable comments posted: 1
🧹 Nitpick comments (1)
rocketmq-client/src/stat/consumer_stats_manager.rs (1)
158-186: Consider adding tests that verify recorded values.The current smoke tests verify methods don't panic, but don't assert that incremented values are reflected in subsequent
consume_statuscalls. Adding at least one integration-style test would strengthen confidence in the end-to-end flow.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rocketmq-client/src/stat/consumer_stats_manager.rs` around lines 158 - 186, Add assertions to the smoke tests to verify the manager actually records increments: after calling inc_consume_ok_tps, inc_consume_failed_tps, inc_consume_rt, inc_pull_rt, and inc_pull_tps on the same group/topic, call the lookup method (consume_status or the manager method that returns stored metrics) and assert the returned TPS/RT values reflect the increments (e.g., that ok/failed TPS counters increased and RT values match or aggregate as expected). Update one or more tests (e.g., smoke_inc_consume_ok_tps and/or a new integration-style test) to create the manager via make_manager(), perform increments, call consume_status (or the appropriate getter) and assert on its fields to confirm end-to-end recording.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@rocketmq-client/src/stat/consumer_stats_manager.rs`:
- Around line 62-89: The four inc_* methods (inc_pull_rt, inc_pull_tps,
inc_consume_rt, inc_consume_ok_tps, inc_consume_failed_tps) currently cast u64
-> i32 with `as i32`, which can silently wrap; change each to perform a safe,
saturating conversion (e.g., clamp the u64 to i32::MAX and cast: let val = if
msgs_or_rt > i32::MAX as u64 { i32::MAX } else { msgs_or_rt as i32 }) before
calling add_value/add_rt_value, and apply the same pattern for both rt and msgs
usages to avoid truncation when forwarding values to topic_and_group_* methods.
---
Nitpick comments:
In `@rocketmq-client/src/stat/consumer_stats_manager.rs`:
- Around line 158-186: Add assertions to the smoke tests to verify the manager
actually records increments: after calling inc_consume_ok_tps,
inc_consume_failed_tps, inc_consume_rt, inc_pull_rt, and inc_pull_tps on the
same group/topic, call the lookup method (consume_status or the manager method
that returns stored metrics) and assert the returned TPS/RT values reflect the
increments (e.g., that ok/failed TPS counters increased and RT values match or
aggregate as expected). Update one or more tests (e.g., smoke_inc_consume_ok_tps
and/or a new integration-style test) to create the manager via make_manager(),
perform increments, call consume_status (or the appropriate getter) and assert
on its fields to confirm end-to-end recording.
ℹ️ Review info
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (4)
rocketmq-client/src/lib.rsrocketmq-client/src/stat.rsrocketmq-client/src/stat/consumer_stats_manager.rsrocketmq-common/src/common/stats/stats_item_set.rs
Comment on lines
+62
to
+89
| pub fn inc_pull_rt(&self, group: &str, topic: &str, rt: u64) { | ||
| self.topic_and_group_pull_rt | ||
| .add_rt_value(&stats_key(topic, group), rt as i32, 1); | ||
| } | ||
|
|
||
| /// Records `msgs` messages successfully pulled in one batch. | ||
| pub fn inc_pull_tps(&self, group: &str, topic: &str, msgs: u64) { | ||
| self.topic_and_group_pull_tps | ||
| .add_value(&stats_key(topic, group), msgs as i32, 1); | ||
| } | ||
|
|
||
| /// Records a single consume response-time observation in milliseconds. | ||
| pub fn inc_consume_rt(&self, group: &str, topic: &str, rt: u64) { | ||
| self.topic_and_group_consume_rt | ||
| .add_rt_value(&stats_key(topic, group), rt as i32, 1); | ||
| } | ||
|
|
||
| /// Records `msgs` messages consumed successfully in one batch. | ||
| pub fn inc_consume_ok_tps(&self, group: &str, topic: &str, msgs: u64) { | ||
| self.topic_and_group_consume_ok_tps | ||
| .add_value(&stats_key(topic, group), msgs as i32, 1); | ||
| } | ||
|
|
||
| /// Records `msgs` messages that failed consumption in one batch. | ||
| pub fn inc_consume_failed_tps(&self, group: &str, topic: &str, msgs: u64) { | ||
| self.topic_and_group_consume_failed_tps | ||
| .add_value(&stats_key(topic, group), msgs as i32, 1); | ||
| } |
Contributor
There was a problem hiding this comment.
Potential silent truncation when casting u64 to i32.
All inc_* methods cast u64 parameters to i32 using as i32. For values exceeding i32::MAX (2,147,483,647), this silently truncates/wraps the value, potentially causing incorrect statistics in high-throughput scenarios.
Consider using saturating conversion or validating bounds:
🛡️ Proposed fix using saturating conversion
pub fn inc_pull_rt(&self, group: &str, topic: &str, rt: u64) {
+ let rt_clamped = rt.min(i32::MAX as u64) as i32;
self.topic_and_group_pull_rt
- .add_rt_value(&stats_key(topic, group), rt as i32, 1);
+ .add_rt_value(&stats_key(topic, group), rt_clamped, 1);
}Apply similar pattern to other inc_* methods.
📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| pub fn inc_pull_rt(&self, group: &str, topic: &str, rt: u64) { | |
| self.topic_and_group_pull_rt | |
| .add_rt_value(&stats_key(topic, group), rt as i32, 1); | |
| } | |
| /// Records `msgs` messages successfully pulled in one batch. | |
| pub fn inc_pull_tps(&self, group: &str, topic: &str, msgs: u64) { | |
| self.topic_and_group_pull_tps | |
| .add_value(&stats_key(topic, group), msgs as i32, 1); | |
| } | |
| /// Records a single consume response-time observation in milliseconds. | |
| pub fn inc_consume_rt(&self, group: &str, topic: &str, rt: u64) { | |
| self.topic_and_group_consume_rt | |
| .add_rt_value(&stats_key(topic, group), rt as i32, 1); | |
| } | |
| /// Records `msgs` messages consumed successfully in one batch. | |
| pub fn inc_consume_ok_tps(&self, group: &str, topic: &str, msgs: u64) { | |
| self.topic_and_group_consume_ok_tps | |
| .add_value(&stats_key(topic, group), msgs as i32, 1); | |
| } | |
| /// Records `msgs` messages that failed consumption in one batch. | |
| pub fn inc_consume_failed_tps(&self, group: &str, topic: &str, msgs: u64) { | |
| self.topic_and_group_consume_failed_tps | |
| .add_value(&stats_key(topic, group), msgs as i32, 1); | |
| } | |
| pub fn inc_pull_rt(&self, group: &str, topic: &str, rt: u64) { | |
| let rt_clamped = rt.min(i32::MAX as u64) as i32; | |
| self.topic_and_group_pull_rt | |
| .add_rt_value(&stats_key(topic, group), rt_clamped, 1); | |
| } | |
| /// Records `msgs` messages successfully pulled in one batch. | |
| pub fn inc_pull_tps(&self, group: &str, topic: &str, msgs: u64) { | |
| let msgs_clamped = msgs.min(i32::MAX as u64) as i32; | |
| self.topic_and_group_pull_tps | |
| .add_value(&stats_key(topic, group), msgs_clamped, 1); | |
| } | |
| /// Records a single consume response-time observation in milliseconds. | |
| pub fn inc_consume_rt(&self, group: &str, topic: &str, rt: u64) { | |
| let rt_clamped = rt.min(i32::MAX as u64) as i32; | |
| self.topic_and_group_consume_rt | |
| .add_rt_value(&stats_key(topic, group), rt_clamped, 1); | |
| } | |
| /// Records `msgs` messages consumed successfully in one batch. | |
| pub fn inc_consume_ok_tps(&self, group: &str, topic: &str, msgs: u64) { | |
| let msgs_clamped = msgs.min(i32::MAX as u64) as i32; | |
| self.topic_and_group_consume_ok_tps | |
| .add_value(&stats_key(topic, group), msgs_clamped, 1); | |
| } | |
| /// Records `msgs` messages that failed consumption in one batch. | |
| pub fn inc_consume_failed_tps(&self, group: &str, topic: &str, msgs: u64) { | |
| let msgs_clamped = msgs.min(i32::MAX as u64) as i32; | |
| self.topic_and_group_consume_failed_tps | |
| .add_value(&stats_key(topic, group), msgs_clamped, 1); | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-client/src/stat/consumer_stats_manager.rs` around lines 62 - 89, The
four inc_* methods (inc_pull_rt, inc_pull_tps, inc_consume_rt,
inc_consume_ok_tps, inc_consume_failed_tps) currently cast u64 -> i32 with `as
i32`, which can silently wrap; change each to perform a safe, saturating
conversion (e.g., clamp the u64 to i32::MAX and cast: let val = if msgs_or_rt >
i32::MAX as u64 { i32::MAX } else { msgs_or_rt as i32 }) before calling
add_value/add_rt_value, and apply the same pattern for both rt and msgs usages
to avoid truncation when forwarding values to topic_and_group_* methods.
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #6534 +/- ##
==========================================
+ Coverage 42.04% 42.10% +0.05%
==========================================
Files 948 949 +1
Lines 132385 132503 +118
==========================================
+ Hits 55664 55793 +129
+ Misses 76721 76710 -11 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
rocketmq-rust-bot
approved these changes
Feb 27, 2026
Collaborator
rocketmq-rust-bot
left a comment
rocketmq-rust-bot
left a comment
There was a problem hiding this comment.
LGTM - All CI checks passed ✅
rocketmq-rust-bot
added
approved
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
3 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which Issue(s) This PR Fixes(Closes)
Brief Description
How Did You Test This Change?
Summary by CodeRabbit