Repository navigation
[ISSUE #7016]🚀Add Lite topic support and configuration options for enhanced Lite message handling - #7017
Merged
Conversation
…hanced Lite message handling
rocketmq-rust-bot
requested review from
SpaceXCN,
TeslaRustor and
rocketmq-rust-bot
March 29, 2026 10:14
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
Contributor
WalkthroughThis PR introduces comprehensive "Lite" topic support across the RocketMQ broker system, adding a new message type variant, configuration options for lite message handling, subscription group attributes, topic attributes with expiration support, and proxy integration for the new lite topic type. Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 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: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
rocketmq-broker/src/transaction/queue/default_transactional_message_check_listener.rs (1)
208-230:⚠️ Potential issue | 🟡 MinorFix self-comparison assertion and provide fmt/clippy verification results.
Line 216 contains a suspicious assertion comparing a field to itself:
assert_eq!(result.message_ext_inner.queue_id, result.message_ext_inner.queue_id);This should compare against the expected value from
msg_ext(similar to the other assertions in the test). Additionally, as per coding guidelines, run and attach results from:
cargo fmt --all -- --checkcargo clippy --workspace --no-deps --all-targets --all-features -- -D warningscargo fmt -p rocketmq-broker -- --checkcargo clippy -p rocketmq-broker --all-targets --all-features -- -D warnings🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@rocketmq-broker/src/transaction/queue/default_transactional_message_check_listener.rs` around lines 208 - 230, The test contains a self-comparison assertion: replace the bogus assert_eq!(result.message_ext_inner.queue_id, result.message_ext_inner.queue_id) with a comparison against the expected value from the source message (e.g., assert_eq!(result.message_ext_inner.queue_id, msg_ext.queue_id() or the appropriate accessor on msg_ext) so it mirrors the other assertions; then run and attach verification results for formatting and lints: run `cargo fmt --all -- --check`, `cargo clippy --workspace --no-deps --all-targets --all-features -- -D warnings`, and the crate-scoped checks `cargo fmt -p rocketmq-broker -- --check` and `cargo clippy -p rocketmq-broker --all-targets --all-features -- -D warnings`.
🤖 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-common/src/common/attribute/subscription_group_attributes.rs`:
- Around line 1-3: This file is missing the repository's standard
copyright/license header; add the project's header block at the top of
rocketmq-common/src/common/attribute/subscription_group_attributes.rs (above the
existing use statements) and ensure the copyright year is set to "Copyright
2023" exactly; do not modify the rest of the imports (use
std::collections::HashMap, use std::collections::HashSet, use std::sync::Arc) or
other symbols.
In `@rocketmq-common/src/common/broker/broker_config.rs`:
- Around line 702-737: The CI/workflow is running rustfmt/clippy for
rocketmq-proxy even though only rocketmq-common changed; update the
formatting/lint steps to target only the modified crate by replacing or
parameterizing the commands so they run cargo fmt and cargo clippy against
rocketmq-common (use the manifest path rocketmq-common/Cargo.toml) instead of
running checks for rocketmq-proxy or the whole workspace; specifically adjust
the steps that currently invoke cargo fmt --all or cargo clippy --workspace to
use the single-crate commands (cargo fmt --manifest-path
rocketmq-common/Cargo.toml --all and cargo clippy --manifest-path
rocketmq-common/Cargo.toml --all-targets --all-features -- -D warnings) and
ensure any fallback workspace-wide checks are not applied to rocketmq-proxy when
it wasn't modified.
In `@rocketmq-common/src/common/config.rs`:
- Around line 92-110: The lite-topic expiration uses i32 but the attribute range
is [−1, i64::MAX]; change set_lite_topic_expiration and
get_lite_topic_expiration to use i64 instead of i32, update
set_lite_topic_expiration to store lite_topic_expiration.to_string() as before,
and update get_lite_topic_expiration to parse with parse::<i64>().ok() and
return an i64 (still returning -1 when not Lite or on parse/absent). Update all
related signatures and any callers to match the new i64 types and ensure
TopicAttributes::lite_topic_expiration_attribute() remains the authoritative
range.
---
Outside diff comments:
In
`@rocketmq-broker/src/transaction/queue/default_transactional_message_check_listener.rs`:
- Around line 208-230: The test contains a self-comparison assertion: replace
the bogus assert_eq!(result.message_ext_inner.queue_id,
result.message_ext_inner.queue_id) with a comparison against the expected value
from the source message (e.g., assert_eq!(result.message_ext_inner.queue_id,
msg_ext.queue_id() or the appropriate accessor on msg_ext) so it mirrors the
other assertions; then run and attach verification results for formatting and
lints: run `cargo fmt --all -- --check`, `cargo clippy --workspace --no-deps
--all-targets --all-features -- -D warnings`, and the crate-scoped checks `cargo
fmt -p rocketmq-broker -- --check` and `cargo clippy -p rocketmq-broker
--all-targets --all-features -- -D warnings`.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Path: .coderabbit.yaml
Review profile: CHILL
Plan: Pro
Run ID: 0c3b0143-12d4-48f4-aa67-c6971385da12
📒 Files selected for processing (9)
rocketmq-broker/src/transaction/queue/default_transactional_message_check_listener.rsrocketmq-common/src/common/attribute/subscription_group_attributes.rsrocketmq-common/src/common/attribute/topic_attributes.rsrocketmq-common/src/common/attribute/topic_message_type.rsrocketmq-common/src/common/broker/broker_config.rsrocketmq-common/src/common/config.rsrocketmq-common/src/common/message.rsrocketmq-proxy/src/cluster.rsrocketmq-proxy/src/local.rs
Comment on lines
1
to
3
| use std::collections::HashMap; | ||
| use std::collections::HashSet; | ||
| use std::sync::Arc; |
Contributor
There was a problem hiding this comment.
Add the standard copyright/license header to this Rust source file.
Line 1 starts directly with imports, but touched Rust files should keep the repository’s standard Copyright 2023 header block.
Based on learnings: In Rust source files (*.rs) across the rocketmq-rust repository, enforce using "Copyright 2023" as the year in the header instead of the current calendar year.
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-common/src/common/attribute/subscription_group_attributes.rs` around
lines 1 - 3, This file is missing the repository's standard copyright/license
header; add the project's header block at the top of
rocketmq-common/src/common/attribute/subscription_group_attributes.rs (above the
existing use statements) and ensure the copyright year is set to "Copyright
2023" exactly; do not modify the rest of the imports (use
std::collections::HashMap, use std::collections::HashSet, use std::sync::Arc) or
other symbols.
Comment on lines
+702
to
+737
| #[serde(default = "defaults::enable_lite_event_mode")] | ||
| pub enable_lite_event_mode: bool, | ||
|
|
||
| #[serde(default = "defaults::lite_event_check_interval")] | ||
| pub lite_event_check_interval: u64, | ||
|
|
||
| #[serde(default = "defaults::lite_ttl_check_interval")] | ||
| pub lite_ttl_check_interval: u64, | ||
|
|
||
| #[serde(default = "defaults::lite_subscription_check_interval")] | ||
| pub lite_subscription_check_interval: u64, | ||
|
|
||
| #[serde(default = "defaults::lite_subscription_check_timeout_mills")] | ||
| pub lite_subscription_check_timeout_mills: u64, | ||
|
|
||
| #[serde(default = "defaults::max_lite_subscription_count")] | ||
| pub max_lite_subscription_count: u64, | ||
|
|
||
| #[serde(default = "defaults::enable_lite_pop_log")] | ||
| pub enable_lite_pop_log: bool, | ||
|
|
||
| #[serde(default = "defaults::lite_event_full_dispatch_delay_time")] | ||
| pub lite_event_full_dispatch_delay_time: u64, | ||
|
|
||
| #[serde(default = "defaults::lite_lag_latency_collect_enable")] | ||
| pub lite_lag_latency_collect_enable: bool, | ||
|
|
||
| #[serde(default = "defaults::lite_lag_latency_metrics_enable")] | ||
| pub lite_lag_latency_metrics_enable: bool, | ||
|
|
||
| #[serde(default = "defaults::lite_lag_count_metrics_enable")] | ||
| pub lite_lag_count_metrics_enable: bool, | ||
|
|
||
| #[serde(default = "defaults::lite_lag_latency_top_k")] | ||
| pub lite_lag_latency_top_k: i32, | ||
|
|
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Affected projects
cargo fmt --manifest-path rocketmq-common/Cargo.toml --all -- --check
cargo clippy --manifest-path rocketmq-common/Cargo.toml --all-targets --all-features -- -D warnings
cargo fmt --manifest-path rocketmq-proxy/Cargo.toml --all -- --check
cargo clippy --manifest-path rocketmq-proxy/Cargo.toml --all-targets --all-features -- -D warnings
# Root workspace validation
cargo fmt --all -- --check
cargo clippy --workspace --no-deps --all-targets --all-features -- -D warningsRepository: mxsm/rocketmq-rust
Length of output: 267
🏁 Script executed:
git diff --name-only HEAD~1 HEAD 2>/dev/null | head -20Repository: mxsm/rocketmq-rust
Length of output: 44
🏁 Script executed:
git status --porcelain 2>/dev/null | head -20Repository: mxsm/rocketmq-rust
Length of output: 44
🏁 Script executed:
# Check if rocketmq-proxy was actually modified
git diff --name-only HEAD~1 HEAD 2>/dev/null | grep -i "rocketmq-proxy" || echo "No rocketmq-proxy changes found"Repository: mxsm/rocketmq-rust
Length of output: 93
🏁 Script executed:
# List all modified files to understand scope
git log --oneline -1 2>/dev/null || echo "Git log not available"Repository: mxsm/rocketmq-rust
Length of output: 169
🏁 Script executed:
# Show all files modified in the current commit
git show --name-only --pretty="" HEAD 2>/dev/null || git diff --name-only 2>/dev/nullRepository: mxsm/rocketmq-rust
Length of output: 50375
🏁 Script executed:
# Get more detailed view of what changed
git show --stat HEAD 2>/dev/null | head -50Repository: mxsm/rocketmq-rust
Length of output: 3017
🏁 Script executed:
# Get files modified in rocketmq-common in this commit
git diff --name-only HEAD~1 HEAD 2>/dev/null | grep "^rocketmq-common/" | head -30Repository: mxsm/rocketmq-rust
Length of output: 44
🏁 Script executed:
# Check if rocketmq-proxy directory exists and was modified
git diff --name-only HEAD~1 HEAD 2>/dev/null | grep -c "^rocketmq-proxy/" || echo "0"Repository: mxsm/rocketmq-rust
Length of output: 65
🏁 Script executed:
# Get the specific diff for broker_config.rs to see what changed
git diff HEAD~1 HEAD -- "rocketmq-common/src/common/broker/broker_config.rs" 2>/dev/null | head -100Repository: mxsm/rocketmq-rust
Length of output: 44
🏁 Script executed:
# List all Rust files modified in this commit
git diff --name-only HEAD~1 HEAD 2>/dev/null | grep "\.rs$" | head -30Repository: mxsm/rocketmq-rust
Length of output: 44
🏁 Script executed:
# Check if broker_config.rs exists in the repo
find . -name "broker_config.rs" -type f 2>/dev/nullRepository: mxsm/rocketmq-rust
Length of output: 114
🏁 Script executed:
# Check what was actually in the commit (show the stat summary)
git show --stat HEAD 2>/dev/null | tail -30Repository: mxsm/rocketmq-rust
Length of output: 1916
🏁 Script executed:
# Check the line count of broker_config.rs
wc -l rocketmq-common/src/common/broker/broker_config.rsRepository: mxsm/rocketmq-rust
Length of output: 117
🏁 Script executed:
# Read lines around 702-737 to verify the content matches the review comment
sed -n '700,740p' rocketmq-common/src/common/broker/broker_config.rsRepository: mxsm/rocketmq-rust
Length of output: 1523
Remove rocketmq-proxy from the scope of required Rust formatting and linting checks—it was not modified in this commit.
The review comment requests validation that cargo fmt and cargo clippy were executed for both rocketmq-common and rocketmq-proxy. However, only rocketmq-common was modified in this commit. Ensure the required Rust code quality checks are run only for the affected crate:
cargo fmt --manifest-path rocketmq-common/Cargo.toml --all
cargo clippy --manifest-path rocketmq-common/Cargo.toml --all-targets --all-features -- -D warnings
cargo fmt --all
cargo clippy --workspace --no-deps --all-targets --all-features -- -D warnings
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-common/src/common/broker/broker_config.rs` around lines 702 - 737,
The CI/workflow is running rustfmt/clippy for rocketmq-proxy even though only
rocketmq-common changed; update the formatting/lint steps to target only the
modified crate by replacing or parameterizing the commands so they run cargo fmt
and cargo clippy against rocketmq-common (use the manifest path
rocketmq-common/Cargo.toml) instead of running checks for rocketmq-proxy or the
whole workspace; specifically adjust the steps that currently invoke cargo fmt
--all or cargo clippy --workspace to use the single-crate commands (cargo fmt
--manifest-path rocketmq-common/Cargo.toml --all and cargo clippy
--manifest-path rocketmq-common/Cargo.toml --all-targets --all-features -- -D
warnings) and ensure any fallback workspace-wide checks are not applied to
rocketmq-proxy when it wasn't modified.
Comment on lines
+92
to
+110
| pub fn set_lite_topic_expiration(&mut self, lite_topic_expiration: i32) { | ||
| if self.get_topic_message_type() != TopicMessageType::Lite { | ||
| return; | ||
| } | ||
| self.attributes.insert( | ||
| TopicAttributes::lite_topic_expiration_attribute().name().clone(), | ||
| CheetahString::from_string(lite_topic_expiration.to_string()), | ||
| ); | ||
| } | ||
|
|
||
| pub fn get_lite_topic_expiration(&self) -> i32 { | ||
| if self.get_topic_message_type() != TopicMessageType::Lite { | ||
| return -1; | ||
| } | ||
| self.attributes | ||
| .get(TopicAttributes::lite_topic_expiration_attribute().name()) | ||
| .and_then(|content| content.parse::<i32>().ok()) | ||
| .unwrap_or(-1) | ||
| } |
Contributor
There was a problem hiding this comment.
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Verify current signatures and all call sites before widening to i64.
rg -n -C2 '\bpub fn set_lite_topic_expiration\s*\(' --type=rust
rg -n -C2 '\bpub fn get_lite_topic_expiration\s*\(' --type=rust
rg -n -C2 '\bset_lite_topic_expiration\s*\(' --type=rust
rg -n -C2 '\bget_lite_topic_expiration\s*\(' --type=rustRepository: mxsm/rocketmq-rust
Length of output: 3752
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Find TopicAttributes definition
rg -n "lite_topic_expiration_attribute" --type=rust -A5
# Find TopicAttributes struct/impl
rg -n "impl.*TopicAttributes|struct TopicAttributes" --type=rust
# Search for LongRangeAttribute to understand the type
rg -n "LongRangeAttribute" --type=rust -B2 -A5Repository: mxsm/rocketmq-rust
Length of output: 17102
🏁 Script executed:
#!/bin/bash
set -euo pipefail
# Check the file structure to understand the module context
head -20 rocketmq-common/src/common/config.rs
# Verify the exact location and surrounding context
sed -n '90,115p' rocketmq-common/src/common/config.rsRepository: mxsm/rocketmq-rust
Length of output: 1702
Use i64 for lite-topic expiration to match the attribute's declared range.
The lite_topic_expiration_attribute() is defined with range [−1, i64::MAX], but the getter/setter methods use i32. Values exceeding i32::MAX will silently fail to parse and return −1.
Proposed fix
- pub fn set_lite_topic_expiration(&mut self, lite_topic_expiration: i32) {
+ pub fn set_lite_topic_expiration(&mut self, lite_topic_expiration: i64) {
if self.get_topic_message_type() != TopicMessageType::Lite {
return;
}
self.attributes.insert(
TopicAttributes::lite_topic_expiration_attribute().name().clone(),
CheetahString::from_string(lite_topic_expiration.to_string()),
);
}
- pub fn get_lite_topic_expiration(&self) -> i32 {
+ pub fn get_lite_topic_expiration(&self) -> i64 {
if self.get_topic_message_type() != TopicMessageType::Lite {
return -1;
}
self.attributes
.get(TopicAttributes::lite_topic_expiration_attribute().name())
- .and_then(|content| content.parse::<i32>().ok())
+ .and_then(|content| content.parse::<i64>().ok())
.unwrap_or(-1)
}📝 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 set_lite_topic_expiration(&mut self, lite_topic_expiration: i32) { | |
| if self.get_topic_message_type() != TopicMessageType::Lite { | |
| return; | |
| } | |
| self.attributes.insert( | |
| TopicAttributes::lite_topic_expiration_attribute().name().clone(), | |
| CheetahString::from_string(lite_topic_expiration.to_string()), | |
| ); | |
| } | |
| pub fn get_lite_topic_expiration(&self) -> i32 { | |
| if self.get_topic_message_type() != TopicMessageType::Lite { | |
| return -1; | |
| } | |
| self.attributes | |
| .get(TopicAttributes::lite_topic_expiration_attribute().name()) | |
| .and_then(|content| content.parse::<i32>().ok()) | |
| .unwrap_or(-1) | |
| } | |
| pub fn set_lite_topic_expiration(&mut self, lite_topic_expiration: i64) { | |
| if self.get_topic_message_type() != TopicMessageType::Lite { | |
| return; | |
| } | |
| self.attributes.insert( | |
| TopicAttributes::lite_topic_expiration_attribute().name().clone(), | |
| CheetahString::from_string(lite_topic_expiration.to_string()), | |
| ); | |
| } | |
| pub fn get_lite_topic_expiration(&self) -> i64 { | |
| if self.get_topic_message_type() != TopicMessageType::Lite { | |
| return -1; | |
| } | |
| self.attributes | |
| .get(TopicAttributes::lite_topic_expiration_attribute().name()) | |
| .and_then(|content| content.parse::<i64>().ok()) | |
| .unwrap_or(-1) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-common/src/common/config.rs` around lines 92 - 110, The lite-topic
expiration uses i32 but the attribute range is [−1, i64::MAX]; change
set_lite_topic_expiration and get_lite_topic_expiration to use i64 instead of
i32, update set_lite_topic_expiration to store lite_topic_expiration.to_string()
as before, and update get_lite_topic_expiration to parse with
parse::<i64>().ok() and return an i64 (still returning -1 when not Lite or on
parse/absent). Update all related signatures and any callers to match the new
i64 types and ensure TopicAttributes::lite_topic_expiration_attribute() remains
the authoritative range.
Codecov Report✅ All modified and coverable lines are covered by tests. Additional details and impacted files@@ Coverage Diff @@
## main #7017 +/- ##
==========================================
+ Coverage 54.53% 54.95% +0.41%
==========================================
Files 1056 1057 +1
Lines 171844 172688 +844
==========================================
+ Hits 93719 94903 +1184
+ Misses 78125 77785 -340 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
rocketmq-rust-bot
approved these changes
Mar 29, 2026
rocketmq-rust-bot
left a comment
rocketmq-rust-bot
left a comment
Collaborator
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
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
Release Notes