Repository navigation
Conversation
…iption requests and responses
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💥. |
Contributor
WalkthroughThis pull request implements a new Changes
Sequence Diagram(s)sequenceDiagram
participant Client
participant BrokerRuntime
participant LiteManagerProcessor
participant TopicConfigManager
participant SubscriptionRegistry
participant QueueStore
Client->>BrokerRuntime: RequestCode::{GetBrokerLiteInfo | GetLiteTopicInfo | ...}
BrokerRuntime->>LiteManagerProcessor: process_request(command)
alt GetBrokerLiteInfo
LiteManagerProcessor->>SubscriptionRegistry: all_subscriptions()
SubscriptionRegistry-->>LiteManagerProcessor: Vec<LiteSubscriptionRecord>
LiteManagerProcessor->>QueueStore: get stats (lmq counts, offsets)
QueueStore-->>LiteManagerProcessor: metrics
LiteManagerProcessor-->>BrokerRuntime: RemotingCommand (BrokerLiteInfo)
else GetLiteTopicInfo
LiteManagerProcessor->>TopicConfigManager: validate_topic(topic)
TopicConfigManager-->>LiteManagerProcessor: config or error
LiteManagerProcessor->>SubscriptionRegistry: query lite subscriptions
SubscriptionRegistry-->>LiteManagerProcessor: subscriber set, offsets
LiteManagerProcessor-->>BrokerRuntime: RemotingCommand (TopicLiteInfo)
else GetLiteClientInfo
LiteManagerProcessor->>TopicConfigManager: validate consumer group
TopicConfigManager-->>LiteManagerProcessor: group config
LiteManagerProcessor->>SubscriptionRegistry: resolve client subscription
SubscriptionRegistry-->>LiteManagerProcessor: lite_topic_set, timestamps
LiteManagerProcessor-->>BrokerRuntime: RemotingCommand (ClientLiteInfo)
end
BrokerRuntime-->>Client: RemotingCommand (response)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~45 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
🤖 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-broker/src/broker_runtime.rs`:
- Line 4215: The code deletes runtime.message_store_config().store_path_root_dir
via std::fs::remove_dir_all while the runtime is still alive; change the
sequence to first cleanly stop/shutdown the runtime (e.g., call the runtime's
shutdown/stop/close method and await it if async, or explicitly drop the runtime
variable) so all file handles are released, then perform std::fs::remove_dir_all
and handle its Result instead of ignoring it; apply the same change at the other
occurrences that call remove_dir_all (the blocks around the current calls at
lines referenced by the review) so temp dirs are always removed only after the
runtime is stopped and failures are logged or propagated.
In `@rocketmq-broker/src/processor/lite_manager_processor.rs`:
- Around line 244-283: After extracting parent_topic, also verify the topic
actually exists before proceeding: call the broker's topic registry/manager via
self.broker_runtime_inner (e.g., a topic_config_manager or topic_registry method
— whichever API provides a topic_exists/has_topic check) and if the topic is
missing return Ok(Some(self.response_with_code(request,
ResponseCode::TopicNotExist, format!("Parent topic [{}] does not exist.",
parent_topic)))) instead of continuing; keep this existence check before
validate_consumer_group and before looking up lite_subscription (so
GetLiteClientInfo mirrors other lite-manager handlers).
- Around line 129-205: Both handlers validate topic existence before checking
for blank/malformed input, causing empty topics to produce TopicNotExist instead
of InvalidParameter; update get_parent_topic_info to validate
request_header.topic is non-empty (return ResponseCode::InvalidParameter with an
explanatory message) before calling topic_config_manager().contains_topic, and
update get_lite_topic_info to call to_lmq_name (and/or check that
request_header.parent_topic and request_header.lite_topic are non-blank) before
the contains_topic check so empty parent_topic/lite_topic return
InvalidParameter rather than TopicNotExist; apply the same response construction
style (response_with_code) used elsewhere for consistency.
🪄 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: bb0ee3b8-b6ee-458b-a923-bc20e2b544d2
📒 Files selected for processing (4)
rocketmq-broker/src/broker_runtime.rsrocketmq-broker/src/processor.rsrocketmq-broker/src/processor/lite_manager_processor.rsrocketmq-broker/src/subscription/lite_subscription_registry.rs
| ])) | ||
| ); | ||
|
|
||
| let _ = std::fs::remove_dir_all(runtime.message_store_config().store_path_root_dir.as_str()); |
Contributor
There was a problem hiding this comment.
Shut the runtime down before deleting the temp store.
These tests remove store_path_root_dir while runtime is still alive. That makes cleanup platform-dependent and the ignored remove_dir_all result can silently hide leaked temp directories.
Suggested cleanup pattern
- let _ = std::fs::remove_dir_all(runtime.message_store_config().store_path_root_dir.as_str());
+ let temp_root = runtime.message_store_config().store_path_root_dir.as_str().to_owned();
+ runtime.shutdown().await;
+ std::fs::remove_dir_all(temp_root).expect("remove lite test root");Also applies to: 4258-4258, 4302-4302, 4352-4352
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-broker/src/broker_runtime.rs` at line 4215, The code deletes
runtime.message_store_config().store_path_root_dir via std::fs::remove_dir_all
while the runtime is still alive; change the sequence to first cleanly
stop/shutdown the runtime (e.g., call the runtime's shutdown/stop/close method
and await it if async, or explicitly drop the runtime variable) so all file
handles are released, then perform std::fs::remove_dir_all and handle its Result
instead of ignoring it; apply the same change at the other occurrences that call
remove_dir_all (the blocks around the current calls at lines referenced by the
review) so temp dirs are always removed only after the runtime is stopped and
failures are logged or propagated.
Comment on lines
+129
to
+205
| fn get_parent_topic_info( | ||
| &self, | ||
| request: &RemotingCommand, | ||
| ) -> rocketmq_error::RocketMQResult<Option<RemotingCommand>> { | ||
| let request_header = request.decode_command_custom_header::<GetParentTopicInfoRequestHeader>()?; | ||
| if !self | ||
| .broker_runtime_inner | ||
| .topic_config_manager() | ||
| .contains_topic(&request_header.topic) | ||
| { | ||
| return Ok(Some(self.response_with_code( | ||
| request, | ||
| ResponseCode::TopicNotExist, | ||
| format!("Topic [{}] not exist.", request_header.topic), | ||
| ))); | ||
| } | ||
|
|
||
| let subscriptions = self | ||
| .broker_runtime_inner | ||
| .lite_subscription_registry() | ||
| .all_subscriptions() | ||
| .into_iter() | ||
| .filter(|subscription| subscription.topic == request_header.topic) | ||
| .collect::<Vec<_>>(); | ||
| if subscriptions.is_empty() { | ||
| return Ok(Some(self.response_with_code( | ||
| request, | ||
| ResponseCode::QueryNotFound, | ||
| format!("Topic [{}] has no lite subscriptions.", request_header.topic), | ||
| ))); | ||
| } | ||
|
|
||
| let groups = subscriptions | ||
| .iter() | ||
| .map(|subscription| subscription.group.clone()) | ||
| .collect::<HashSet<_>>(); | ||
| let lite_topic_set = subscriptions | ||
| .iter() | ||
| .flat_map(|subscription| subscription.lite_topic_set.iter()) | ||
| .filter_map(|lmq_name| get_lite_topic(lmq_name.as_str()).map(CheetahString::from_string)) | ||
| .collect::<HashSet<_>>(); | ||
|
|
||
| let mut body = GetParentTopicInfoResponseBody::new(); | ||
| body.set_topic(request_header.topic); | ||
| body.set_ttl(0); | ||
| body.set_groups(groups); | ||
| body.set_lmq_num(lite_topic_set.len() as i32); | ||
| body.set_lite_topic_count(lite_topic_set.len() as i32); | ||
|
|
||
| Ok(Some(self.response_with_body(request, &body)?)) | ||
| } | ||
|
|
||
| fn get_lite_topic_info( | ||
| &self, | ||
| request: &RemotingCommand, | ||
| ) -> rocketmq_error::RocketMQResult<Option<RemotingCommand>> { | ||
| let request_header = request.decode_command_custom_header::<GetLiteTopicInfoRequestHeader>()?; | ||
| if !self | ||
| .broker_runtime_inner | ||
| .topic_config_manager() | ||
| .contains_topic(&request_header.parent_topic) | ||
| { | ||
| return Ok(Some(self.response_with_code( | ||
| request, | ||
| ResponseCode::TopicNotExist, | ||
| format!("Topic [{}] not exist.", request_header.parent_topic), | ||
| ))); | ||
| } | ||
|
|
||
| let Some(lmq_name) = to_lmq_name(request_header.parent_topic.as_str(), request_header.lite_topic.as_str()) | ||
| else { | ||
| return Ok(Some(self.response_with_code( | ||
| request, | ||
| ResponseCode::InvalidParameter, | ||
| "parentTopic or liteTopic is blank.", | ||
| ))); | ||
| }; |
Contributor
There was a problem hiding this comment.
Reject blank topic fields before the existence lookup.
GetParentTopicInfo currently turns topic="" into TopicNotExist, and GetLiteTopicInfo checks contains_topic before the to_lmq_name validation, so an empty parent_topic never reaches the InvalidParameter branch at Line 200. That makes malformed requests indistinguishable from missing topics.
Possible fix
let request_header = request.decode_command_custom_header::<GetParentTopicInfoRequestHeader>()?;
+ if request_header.topic.is_empty() {
+ return Ok(Some(self.response_with_code(
+ request,
+ ResponseCode::InvalidParameter,
+ "topic is blank.",
+ )));
+ }
if !self
.broker_runtime_inner
.topic_config_manager()
.contains_topic(&request_header.topic)
{
let request_header = request.decode_command_custom_header::<GetLiteTopicInfoRequestHeader>()?;
+ if request_header.parent_topic.is_empty() || request_header.lite_topic.is_empty() {
+ return Ok(Some(self.response_with_code(
+ request,
+ ResponseCode::InvalidParameter,
+ "parentTopic or liteTopic is blank.",
+ )));
+ }
if !self
.broker_runtime_inner
.topic_config_manager()
.contains_topic(&request_header.parent_topic)🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-broker/src/processor/lite_manager_processor.rs` around lines 129 -
205, Both handlers validate topic existence before checking for blank/malformed
input, causing empty topics to produce TopicNotExist instead of
InvalidParameter; update get_parent_topic_info to validate request_header.topic
is non-empty (return ResponseCode::InvalidParameter with an explanatory message)
before calling topic_config_manager().contains_topic, and update
get_lite_topic_info to call to_lmq_name (and/or check that
request_header.parent_topic and request_header.lite_topic are non-blank) before
the contains_topic check so empty parent_topic/lite_topic return
InvalidParameter rather than TopicNotExist; apply the same response construction
style (response_with_code) used elsewhere for consistency.
Comment on lines
+244
to
+283
| let request_header = request.decode_command_custom_header::<GetLiteClientInfoRequestHeader>()?; | ||
| let Some(parent_topic) = request_header.parent_topic.filter(|topic| !topic.is_empty()) else { | ||
| return Ok(Some(self.response_with_code( | ||
| request, | ||
| ResponseCode::InvalidParameter, | ||
| "parentTopic is blank.", | ||
| ))); | ||
| }; | ||
| let Some(group) = request_header.group.filter(|group| !group.is_empty()) else { | ||
| return Ok(Some(self.response_with_code( | ||
| request, | ||
| ResponseCode::InvalidParameter, | ||
| "group is blank.", | ||
| ))); | ||
| }; | ||
| let Some(client_id) = request_header.client_id.filter(|client_id| !client_id.is_empty()) else { | ||
| return Ok(Some(self.response_with_code( | ||
| request, | ||
| ResponseCode::InvalidParameter, | ||
| "clientId is blank.", | ||
| ))); | ||
| }; | ||
| if let Err((code, remark)) = self.validate_consumer_group(&group, &parent_topic) { | ||
| return Ok(Some(self.response_with_code(request, code, remark))); | ||
| } | ||
|
|
||
| let Some(subscription) = self | ||
| .broker_runtime_inner | ||
| .lite_subscription_registry() | ||
| .lite_subscription(&client_id, &group, &parent_topic) | ||
| else { | ||
| return Ok(Some(self.response_with_code( | ||
| request, | ||
| ResponseCode::QueryNotFound, | ||
| format!( | ||
| "Client [{}] has no lite subscription on [{}]-[{}].", | ||
| client_id, group, parent_topic | ||
| ), | ||
| ))); | ||
| }; |
Contributor
There was a problem hiding this comment.
Validate parent_topic exists in GetLiteClientInfo.
This handler validates the group but never the topic. If a parent topic is deleted while the group binding or registry entry remains, this path returns Success or QueryNotFound instead of TopicNotExist, unlike the other lite-manager handlers.
Possible fix
let Some(client_id) = request_header.client_id.filter(|client_id| !client_id.is_empty()) else {
return Ok(Some(self.response_with_code(
request,
ResponseCode::InvalidParameter,
"clientId is blank.",
)));
};
+ if !self
+ .broker_runtime_inner
+ .topic_config_manager()
+ .contains_topic(&parent_topic)
+ {
+ return Ok(Some(self.response_with_code(
+ request,
+ ResponseCode::TopicNotExist,
+ format!("Topic [{}] not exist.", parent_topic),
+ )));
+ }
if let Err((code, remark)) = self.validate_consumer_group(&group, &parent_topic) {
return Ok(Some(self.response_with_code(request, code, remark)));
}📝 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
| let request_header = request.decode_command_custom_header::<GetLiteClientInfoRequestHeader>()?; | |
| let Some(parent_topic) = request_header.parent_topic.filter(|topic| !topic.is_empty()) else { | |
| return Ok(Some(self.response_with_code( | |
| request, | |
| ResponseCode::InvalidParameter, | |
| "parentTopic is blank.", | |
| ))); | |
| }; | |
| let Some(group) = request_header.group.filter(|group| !group.is_empty()) else { | |
| return Ok(Some(self.response_with_code( | |
| request, | |
| ResponseCode::InvalidParameter, | |
| "group is blank.", | |
| ))); | |
| }; | |
| let Some(client_id) = request_header.client_id.filter(|client_id| !client_id.is_empty()) else { | |
| return Ok(Some(self.response_with_code( | |
| request, | |
| ResponseCode::InvalidParameter, | |
| "clientId is blank.", | |
| ))); | |
| }; | |
| if let Err((code, remark)) = self.validate_consumer_group(&group, &parent_topic) { | |
| return Ok(Some(self.response_with_code(request, code, remark))); | |
| } | |
| let Some(subscription) = self | |
| .broker_runtime_inner | |
| .lite_subscription_registry() | |
| .lite_subscription(&client_id, &group, &parent_topic) | |
| else { | |
| return Ok(Some(self.response_with_code( | |
| request, | |
| ResponseCode::QueryNotFound, | |
| format!( | |
| "Client [{}] has no lite subscription on [{}]-[{}].", | |
| client_id, group, parent_topic | |
| ), | |
| ))); | |
| }; | |
| let request_header = request.decode_command_custom_header::<GetLiteClientInfoRequestHeader>()?; | |
| let Some(parent_topic) = request_header.parent_topic.filter(|topic| !topic.is_empty()) else { | |
| return Ok(Some(self.response_with_code( | |
| request, | |
| ResponseCode::InvalidParameter, | |
| "parentTopic is blank.", | |
| ))); | |
| }; | |
| let Some(group) = request_header.group.filter(|group| !group.is_empty()) else { | |
| return Ok(Some(self.response_with_code( | |
| request, | |
| ResponseCode::InvalidParameter, | |
| "group is blank.", | |
| ))); | |
| }; | |
| let Some(client_id) = request_header.client_id.filter(|client_id| !client_id.is_empty()) else { | |
| return Ok(Some(self.response_with_code( | |
| request, | |
| ResponseCode::InvalidParameter, | |
| "clientId is blank.", | |
| ))); | |
| }; | |
| if !self | |
| .broker_runtime_inner | |
| .topic_config_manager() | |
| .contains_topic(&parent_topic) | |
| { | |
| return Ok(Some(self.response_with_code( | |
| request, | |
| ResponseCode::TopicNotExist, | |
| format!("Topic [{}] not exist.", parent_topic), | |
| ))); | |
| } | |
| if let Err((code, remark)) = self.validate_consumer_group(&group, &parent_topic) { | |
| return Ok(Some(self.response_with_code(request, code, remark))); | |
| } | |
| let Some(subscription) = self | |
| .broker_runtime_inner | |
| .lite_subscription_registry() | |
| .lite_subscription(&client_id, &group, &parent_topic) | |
| else { | |
| return Ok(Some(self.response_with_code( | |
| request, | |
| ResponseCode::QueryNotFound, | |
| format!( | |
| "Client [{}] has no lite subscription on [{}]-[{}].", | |
| client_id, group, parent_topic | |
| ), | |
| ))); | |
| }; |
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@rocketmq-broker/src/processor/lite_manager_processor.rs` around lines 244 -
283, After extracting parent_topic, also verify the topic actually exists before
proceeding: call the broker's topic registry/manager via
self.broker_runtime_inner (e.g., a topic_config_manager or topic_registry method
— whichever API provides a topic_exists/has_topic check) and if the topic is
missing return Ok(Some(self.response_with_code(request,
ResponseCode::TopicNotExist, format!("Parent topic [{}] does not exist.",
parent_topic)))) instead of continuing; keep this existence check before
validate_consumer_group and before looking up lite_subscription (so
GetLiteClientInfo mirrors other lite-manager handlers).
rocketmq-rust-bot
approved these changes
Mar 29, 2026
Codecov Report❌ Patch coverage is
Additional details and impacted files@@ Coverage Diff @@
## main #7009 +/- ##
==========================================
+ Coverage 54.24% 54.38% +0.13%
==========================================
Files 1053 1054 +1
Lines 170703 171264 +561
==========================================
+ Hits 92603 93141 +538
- Misses 78100 78123 +23 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
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
New Features
Tests