rocketmq-broker/src/error.rs (1)

31-32: LGTM! Consider adding documentation for the new error variant.

The IllegalArgumentError variant is well-structured and appropriate for handling validation errors.

Consider adding documentation to clarify typical usage scenarios:

+    /// Represents errors caused by invalid arguments provided to broker operations.
+    /// For example, invalid topic names or configuration parameters.
     #[error("{0}")]
     IllegalArgumentError(String),

rocketmq-broker/src/processor/query_assignment_processor.rs (4)

322-323: Clarify function call to allocate

The function allocate is called without qualification, which may cause confusion. Since allocate is defined outside of the struct, it would be clearer to qualify the function call or refactor it as an associated method.

Consider qualifying the function or refactoring it:

-allocate(consumer_group, current_cid, mq_all, cid_all)
+self::allocate(consumer_group, current_cid, mq_all, cid_all)

Or move allocate inside the QueryAssignmentProcessor impl block.

---

376-383: Use tracing crate for consistent logging

In line 377, log::info! is used for logging, whereas the rest of the code uses the tracing crate. For consistency and to leverage structured logging, all logs should use tracing.

Update the logging statement:

-log::info!(
+tracing::info!(
     "[BUG] ConsumerGroup: {} The consumerId: {} not in cidAll: {:?}",
     consumer_group,
     current_cid,
     cid_all
 );

---

245-252: Provide a default strategy or clear error when strategy is unsupported

When an unsupported strategy is requested, the code logs a warning and returns None. This might leave the client without a clear understanding of the failure.

Consider responding with an error message to the client:

-if strategy.is_none() {
-    warn!(
-        "QueryLoad: unsupported strategy [{}],  {}",
-        strategy_name,
-        channel.remote_address()
-    );
-    return None;
-}
+let strategy = match self.load_strategy.get(strategy_name) {
+    Some(s) => s,
+    None => {
+        warn!(
+            "QueryLoad: unsupported strategy [{}],  {}",
+            strategy_name,
+            channel.remote_address()
+        );
+        return Some(RemotingCommand::create_response_command_with_code(ResponseCode::SystemError)
+            .set_remark(CheetahString::from_static_str("Unsupported load balancing strategy")));
+    }
+};

Alternatively, default to a standard strategy if appropriate.

---

182-194: Handle empty assignment for broadcasting message model

When no assignment is found in broadcasting mode, the code logs a warning but proceeds without an explicit response.

Consider returning a response indicating that no assignments are available:

-if assigned_queue_set.is_none() {
+if assigned_queue_set.as_ref().map_or(true, |set| set.is_empty()) {
     warn!(
         "QueryLoad: no assignment for group[{}], the topic[{}] does not exist.",
         consumer_group, topic
     );
+    return Some(RemotingCommand::create_response_command_with_code(ResponseCode::TopicNotExist)
+        .set_remark(CheetahString::from_static_str("No message queues available for topic")));
 }

Configuration used: .coderabbit.yaml
Review profile: CHILL

#!/bin/bash
# Description: Check for consistent error handling patterns across the codebase

# Look for places where MQClientError is handled
rg -t rust "MQClientError" -A 3

# Check for error conversion patterns
ast-grep --pattern 'match $_ {
  Err($_) => $_
}'

#!/bin/bash
# Description: Find usages of server_load_balancer_enable to ensure it's properly integrated

# Search for direct field access
rg "server_load_balancer_enable" -t rust

# Search for the property name in string form (camelCase as used in properties)
rg "serverLoadBalancerEnable" -t rust

#!/bin/bash
# Description: Verify QueryAssignmentProcessor constructor matches the parameters

# Check the QueryAssignmentProcessor implementation
ast-grep --pattern 'impl QueryAssignmentProcessor {
  $$$
  fn new($message_store_config: $_, $broker_config: $_, $topic_route_info_manager: $_, $consumer_manager: $_) {
    $$$
  }
  $$$
}'

#!/bin/bash
# Let's try a broader search to find the QueryAssignmentProcessor implementation
rg -A 10 "struct QueryAssignmentProcessor"

# Also search for new() implementation specifically
rg -A 10 "impl QueryAssignmentProcessor"

-let mut cid_all = if let Some(consumer_group_info) = consumer_group_info {
+let cid_all = if let Some(consumer_group_info) = consumer_group_info {
     consumer_group_info.get_all_client_ids()
 } else {
     vec![]
 };
+let mut cid_all = cid_all.clone();