Conversation
Collaborator
|
🔊@magogosora 🚀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-robot
added
Difficulty level/Moderate
rocketmq-rust-bot
added
AI review first
Contributor
WalkthroughAdds end-to-end support for UpdateColdDataFlowCtrGroupConfig: new admin CLI subcommand, client API call, and broker request handler that parses properties and returns structured errors (implementation hook for controller integration remains TODO). Changes
Sequence DiagramsequenceDiagram
participant AdminCLI as Admin CLI
participant SubCmd as UpdateColdDataFlowCtrGroupConfig<br/>SubCommand
participant MQAdminExt as DefaultMQAdminExt
participant ClientAPI as MQClientAPIImpl
participant Broker as AdminBrokerProcessor
AdminCLI->>SubCmd: execute(target, consumer_group, threshold)
SubCmd->>SubCmd: validate & build properties map
SubCmd->>MQAdminExt: start / call update_cold_data_flow_ctr_group_config
MQAdminExt->>ClientAPI: update_cold_data_flow_ctr_group_config(broker_addr, properties)
ClientAPI->>ClientAPI: serialize properties -> body, build RemotingCommand
ClientAPI->>Broker: send RemotingCommand (VIP channel if enabled)
Broker->>Broker: parse body -> properties, validate
Broker-->>ClientAPI: return response (OK / InvalidParameter / SystemError)
ClientAPI-->>MQAdminExt: propagate result/error
MQAdminExt-->>SubCmd: return outcome
SubCmd->>SubCmd: aggregate/log per-broker results
SubCmd-->>AdminCLI: report success/failures
Estimated Code Review Effort🎯 3 (Moderate) | ⏱️ ~25 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 2❌ Failed checks (2 warnings)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing touches
🧪 Generate unit tests (beta)
⚔️ Resolve merge conflicts (beta)
No actionable comments were generated in the recent review. 🎉 🧹 Recent nitpick comments
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
🤖 Fix all issues with AI agents
In
`@rocketmq-broker/src/processor/admin_broker_processor/update_cold_data_flow_ctr_group_config.rs`:
- Around line 55-62: The handler currently parses the request body but always
returns a SystemError stub (response.set_code(ResponseCode::SystemError))
instead of performing the update; replace the stub by retrieving the broker
controller/runtime, fetch the ColdDataCgCtrService from broker_runtime, call the
appropriate update method on ColdDataCgCtrService using the parsed properties
(from mix_all::string_to_properties) and set a successful response code/remark
on success, and on missing service return a clear unsupported response (not
SystemError) explaining ColdDataCgCtrService is not configured; ensure you
update the code paths in update_cold_data_flow_ctr_group_config.rs to use the
real service rather than the placeholder.
In
`@rocketmq-tools/rocketmq-admin/rocketmq-admin-core/src/commands/broker_commands/update_cold_data_flow_ctr_group_config_sub_command.rs`:
- Around line 95-157: The error messages are misleading: update the "Failed to
get user for brokers ..." message used in
UpdateColdDataFlowCtrGroupConfigSubCommand to accurately describe the failure
(e.g., "Failed to get broker addresses for brokers {addrs}") and in
update_config_for_cluster's Err branch include the cluster_name in the formatted
message (e.g., "Failed to fetch broker addresses for cluster {cluster_name}:
{e}"); locate the strings in the UpdateColdDataFlowCtrGroupConfigSubCommand flow
and the update_config_for_cluster function and replace the incorrect/missing
text while keeping existing context and error variable interpolation.
- Around line 32-67: The command currently only accepts a raw threshold string
and forwards it; update the UpdateColdDataFlowCtrGroupConfigSubCommand struct to
add new optional CLI args (e.g., --rate and --priority) and change threshold to
be parsed into a numeric type, then validate it in execute (parse threshold to
u64 or f64, reject parse errors and negative/zero where appropriate), and insert
validated values into the properties HashMap with explicit keys (e.g.,
consumerGroup, threshold, rate, priority) before sending to the broker; make
sure validation errors return RocketMQError::IllegalArgument from execute so
invalid inputs are rejected early.
Comment on lines
+55
to
+62
| let body = mix_all::string_to_properties(&String::from_utf8_lossy(body)); | ||
| match body { | ||
| Some(_body) => { | ||
| // TODO get broker controller and do operations | ||
| Ok(Some(response.set_code(ResponseCode::SystemError).set_remark( | ||
| "UpdateColdDataFlowCtrGroupConfig not implemented: ColdDataCgCtrService not configured in \ | ||
| broker_runtime", | ||
| ))) |
Contributor
There was a problem hiding this comment.
Handler is a stub: valid requests always return SystemError. Line 55–62 returns “not implemented” for any well-formed body, so the new admin command can never succeed. Please wire this to the ColdDataCgCtrService (or equivalent) and apply the updates, or gate the request with a clear unsupported response until implementation is ready.
🤖 Prompt for AI Agents
In
`@rocketmq-broker/src/processor/admin_broker_processor/update_cold_data_flow_ctr_group_config.rs`
around lines 55 - 62, The handler currently parses the request body but always
returns a SystemError stub (response.set_code(ResponseCode::SystemError))
instead of performing the update; replace the stub by retrieving the broker
controller/runtime, fetch the ColdDataCgCtrService from broker_runtime, call the
appropriate update method on ColdDataCgCtrService using the parsed properties
(from mix_all::string_to_properties) and set a successful response code/remark
on success, and on missing service return a clear unsupported response (not
SystemError) explaining ColdDataCgCtrService is not configured; ensure you
update the code paths in update_cold_data_flow_ctr_group_config.rs to use the
real service rather than the placeholder.
Comment on lines
32
to
67
| #[derive(Debug, Clone, Parser)] | ||
| #[command(group(ArgGroup::new("target") | ||
| .required(true) | ||
| .args(&["broker_addr", "cluster_name"])) | ||
| )] | ||
| pub struct UpdateColdDataFlowCtrGroupConfigSubCommand { | ||
| #[arg(short = 'b', long = "brokerAddr", required = false, help = "update which broker")] | ||
| broker_addr: Option<String>, | ||
|
|
||
| #[arg(short = 'c', long = "clusterName", required = false, help = "update which cluster")] | ||
| cluster_name: Option<String>, | ||
|
|
||
| #[arg( | ||
| short = 'g', | ||
| long = "consumerGroup", | ||
| required = true, | ||
| help = "specific consumerGroup" | ||
| )] | ||
| consumer_group: String, | ||
|
|
||
| #[arg(short = 'v', long = "threshold", required = true, help = "cold read threshold value")] | ||
| threshold: String, | ||
| } | ||
|
|
||
| impl CommandExecute for UpdateColdDataFlowCtrGroupConfigSubCommand { | ||
| async fn execute(&self, rpc_hook: Option<Arc<dyn RPCHook>>) -> RocketMQResult<()> { | ||
| let target = Target::new(&self.cluster_name, &self.broker_addr).map_err(|_| { | ||
| RocketMQError::IllegalArgument( | ||
| "UpdateColdDataFlowCtrGroupConfigSubCommand: Specify exactly one of --brokerAddr (-b) or \ | ||
| --clusterName (-c)" | ||
| .into(), | ||
| ) | ||
| })?; | ||
| let mut properties = HashMap::<CheetahString, CheetahString>::new(); | ||
| properties.insert(self.consumer_group.clone().into(), self.threshold.clone().into()); | ||
|
|
Contributor
There was a problem hiding this comment.
Missing validation and config fields vs. intended command capabilities. Line 32–67 only accepts consumerGroup + threshold and forwards the raw string; there’s no validation or support for cold-data rate limits / priority. This falls short of the stated objectives and risks sending invalid config to the broker. Consider parsing threshold to a numeric type (reject invalid/negative) and adding CLI args + properties for rate/priority.
🔧 Minimal validation example
- let mut properties = HashMap::<CheetahString, CheetahString>::new();
- properties.insert(self.consumer_group.clone().into(), self.threshold.clone().into());
+ let threshold: u64 = self.threshold.parse().map_err(|_| {
+ RocketMQError::IllegalArgument("threshold must be a non-negative integer".into())
+ })?;
+ let mut properties = HashMap::<CheetahString, CheetahString>::new();
+ properties.insert(self.consumer_group.clone().into(), threshold.to_string().into());🤖 Prompt for AI Agents
In
`@rocketmq-tools/rocketmq-admin/rocketmq-admin-core/src/commands/broker_commands/update_cold_data_flow_ctr_group_config_sub_command.rs`
around lines 32 - 67, The command currently only accepts a raw threshold string
and forwards it; update the UpdateColdDataFlowCtrGroupConfigSubCommand struct to
add new optional CLI args (e.g., --rate and --priority) and change threshold to
be parsed into a numeric type, then validate it in execute (parse threshold to
u64 or f64, reject parse errors and negative/zero where appropriate), and insert
validated values into the properties HashMap with explicit keys (e.g.,
consumerGroup, threshold, rate, priority) before sending to the broker; make
sure validation errors return RocketMQError::IllegalArgument from execute so
invalid inputs are rejected early.
...dmin-core/src/commands/broker_commands/update_cold_data_flow_ctr_group_config_sub_command.rs
Show resolved
Hide resolved
Codecov Report❌ Patch coverage is Additional details and impacted files@@ Coverage Diff @@
## main #6329 +/- ##
==========================================
- Coverage 42.62% 42.57% -0.05%
==========================================
Files 912 914 +2
Lines 128038 128224 +186
==========================================
+ Hits 54570 54595 +25
- Misses 73468 73629 +161 ☔ View full report in Codecov by Sentry. 🚀 New features to boost your workflow:
|
rocketmq-rust-bot
approved these changes
Feb 14, 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
4 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