[Feature🚀] Implement UpdateControllerConfig Request Handler

rocketmq-common/src/common/controller/controller_config.rs

@@ -12,7 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::HashMap;
 use std::env;
+use std::fs::File;
+use std::io::Write;
 use std::net::SocketAddr;
 use std::path::PathBuf;
 
@@ -453,6 +456,147 @@ impl ControllerConfig {
 
         Ok(())
     }
+
+    /// Update configuration from a map of properties
+    ///
+    /// # Arguments
+    ///
+    /// * `properties` - A map of configuration keys to values
+    ///
+    /// # Returns
+    ///
+    /// A Result indicating success or failure
+    pub fn update(&mut self, properties: &HashMap<String, String>) -> Result<(), String> {
+        for (key, value) in properties {
+            match key.as_str() {
+                "rocketmqHome" => self.rocketmq_home = value.clone(),
+                "controllerType" => self.controller_type = value.clone(),
+                "scanNotActiveBrokerInterval" => {
+                    self.scan_not_active_broker_interval = value
+                        .parse()
+                        .map_err(|e| format!("Invalid scanNotActiveBrokerInterval: {}", e))?;
+                }
+                "controllerThreadPoolNums" => {
+                    self.controller_thread_pool_nums = value
+                        .parse()
+                        .map_err(|e| format!("Invalid controllerThreadPoolNums: {}", e))?;
+                }
+                "controllerRequestThreadPoolQueueCapacity" => {
+                    self.controller_request_thread_pool_queue_capacity = value
+                        .parse()
+                        .map_err(|e| format!("Invalid controllerRequestThreadPoolQueueCapacity: {}", e))?;
+                }
+                "mappedFileSize" => {
+                    self.mapped_file_size = value.parse().map_err(|e| format!("Invalid mappedFileSize: {}", e))?;
+                }
+                "controllerStorePath" => {
+                    self.controller_store_path = value.clone();
+                }
+                "electMasterMaxRetryCount" => {
+                    self.elect_master_max_retry_count = value
+                        .parse()
+                        .map_err(|e| format!("Invalid electMasterMaxRetryCount: {}", e))?;
+                }
+                "enableElectUncleanMaster" => {
+                    self.enable_elect_unclean_master = value
+                        .parse()
+                        .map_err(|e| format!("Invalid enableElectUncleanMaster: {}", e))?;
+                }
+                "isProcessReadEvent" => {
+                    self.is_process_read_event = value
+                        .parse()
+                        .map_err(|e| format!("Invalid isProcessReadEvent: {}", e))?;
+                }
+                "notifyBrokerRoleChanged" => {
+                    self.notify_broker_role_changed = value
+                        .parse()
+                        .map_err(|e| format!("Invalid notifyBrokerRoleChanged: {}", e))?;
+                }
+                "scanInactiveMasterInterval" => {
+                    self.scan_inactive_master_interval = value
+                        .parse()
+                        .map_err(|e| format!("Invalid scanInactiveMasterInterval: {}", e))?;
+                }
+                "metricsExporterType" => {
+                    let exporter_type: MetricsExporterType = value
+                        .parse()
+                        .map_err(|_| format!("Invalid metricsExporterType: {}", value))?;
+
+                    self.metrics_exporter_type = exporter_type;
+                }
+                "metricsGrpcExporterTarget" => self.metrics_grpc_exporter_target = value.clone(),
+                "metricsGrpcExporterHeader" => self.metrics_grpc_exporter_header = value.clone(),
+                "metricGrpcExporterTimeOutInMills" => {
+                    self.metric_grpc_exporter_time_out_in_mills = value
+                        .parse()
+                        .map_err(|e| format!("Invalid metricGrpcExporterTimeOutInMills: {}", e))?;
+                }
+                "metricGrpcExporterIntervalInMills" => {
+                    self.metric_grpc_exporter_interval_in_mills = value
+                        .parse()
+                        .map_err(|e| format!("Invalid metricGrpcExporterIntervalInMills: {}", e))?;
+                }
+                "metricLoggingExporterIntervalInMills" => {
+                    self.metric_logging_exporter_interval_in_mills = value
+                        .parse()
+                        .map_err(|e| format!("Invalid metricLoggingExporterIntervalInMills: {}", e))?;
+                }
+                "metricsPromExporterPort" => {
+                    self.metrics_prom_exporter_port = value
+                        .parse()
+                        .map_err(|e| format!("Invalid metricsPromExporterPort: {}", e))?;
+                }
+                "metricsPromExporterHost" => self.metrics_prom_exporter_host = value.clone(),
+                "metricsLabel" => self.metrics_label = value.clone(),
+                "metricsInDelta" => {
+                    self.metrics_in_delta = value.parse().map_err(|e| format!("Invalid metricsInDelta: {}", e))?;
+                }
+                "configBlackList" => self.config_black_list = value.clone(),
+                "electionTimeoutMs" => {
+                    self.election_timeout_ms =
+                        value.parse().map_err(|e| format!("Invalid electionTimeoutMs: {}", e))?;
+                }
+                "heartbeatIntervalMs" => {
+                    self.heartbeat_interval_ms = value
+                        .parse()
+                        .map_err(|e| format!("Invalid heartbeatIntervalMs: {}", e))?;
+                }
+                "storagePath" => self.storage_path = value.clone(),
+                "enableElectUncleanMasterLocal" => {
+                    self.enable_elect_unclean_master_local = value
+                        .parse()
+                        .map_err(|e| format!("Invalid enableElectUncleanMasterLocal: {}", e))?;
+                }
+                _ => return Err(format!("Unknown configuration key: {}", key)),
+            }
+        }
+
+        // Validate the updated configuration
+        self.validate()
+    }
+
+    /// Persist the configuration to disk
+    ///
+    /// # Returns
+    ///
+    /// A Result indicating success or failure
+    pub fn persist(&self) -> Result<(), String> {
+        // Ensure the directory exists
+        if let Some(parent) = self.config_store_path.parent() {
+            std::fs::create_dir_all(parent).map_err(|e| format!("Failed to create config directory: {}", e))?;
+        }
+
+        // Serialize the configuration to JSON
+        let json = serde_json::to_string_pretty(self).map_err(|e| format!("Failed to serialize config: {}", e))?;
+
+        // Write to file
+        let mut file =
+            File::create(&self.config_store_path).map_err(|e| format!("Failed to create config file: {}", e))?;
+        file.write_all(json.as_bytes())
+            .map_err(|e| format!("Failed to write config file: {}", e))?;
+
+        Ok(())
+    }
 }
 
 #[cfg(test)]
@@ -527,4 +671,55 @@ mod tests {
         assert_eq!(config.metrics_prom_exporter_port, 9090);
         assert_eq!(config.metrics_label, "instance_id:test,uid:123");
     }
+
+    #[test]
+    fn test_config_update() {
+        let mut config = ControllerConfig::default();
+
+        let mut properties = HashMap::new();
+        properties.insert("scanNotActiveBrokerInterval".to_string(), "10000".to_string());
+        properties.insert("controllerThreadPoolNums".to_string(), "32".to_string());
+        properties.insert("enableElectUncleanMaster".to_string(), "true".to_string());
+
+        let result = config.update(&properties);
+        assert!(result.is_ok());
+        assert_eq!(config.scan_not_active_broker_interval, 10000);
+        assert_eq!(config.controller_thread_pool_nums, 32);
+        assert!(config.enable_elect_unclean_master);
+    }
+
+    #[test]
+    fn test_config_update_invalid() {
+        let mut config = ControllerConfig::default();
+
+        let mut properties = HashMap::new();
+        properties.insert("scanNotActiveBrokerInterval".to_string(), "invalid".to_string());
+
+        let result = config.update(&properties);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_config_update_unknown_key() {
+        let mut config = ControllerConfig::default();
+
+        let mut properties = HashMap::new();
+        properties.insert("unknownKey".to_string(), "value".to_string());
+
+        let result = config.update(&properties);
+        assert!(result.is_err());
+    }
+
+    #[test]
+    fn test_config_persist() {
+        let config = ControllerConfig {
+            config_store_path: PathBuf::from("/tmp/test_controller_config.json"),
+            ..Default::default()
+        };
+
+        let result = config.persist();
+        assert!(result.is_ok());
+
+        std::fs::remove_file("/tmp/test_controller_config.json").ok();
+    }
 }

rocketmq-controller/src/processor/controller_request_processor.rs

@@ -421,9 +421,94 @@ impl ControllerRequestProcessor {
         &mut self,
         _channel: Channel,
         _ctx: ConnectionHandlerContext,
-        _request: &mut RemotingCommand,
+        request: &mut RemotingCommand,
     ) -> RocketMQResult<Option<RemotingCommand>> {
-        unimplemented!("unimplemented handle_update_controller_config")
+        use rocketmq_error::RocketMQError;
+        use rocketmq_remoting::code::response_code::ResponseCode;
+
+        // Parse request body as properties map
+        let properties = if let Some(body) = request.body() {
+            self.parse_properties_from_string(body).map_err(|e| {
+                RocketMQError::broker_operation_failed(
+                    "UPDATE_CONTROLLER_CONFIG",
+                    ResponseCode::ControllerInvalidRequest.to_i32(),
+                    format!("Failed to parse config properties: {:?}", e),
+                )
+            })?
+        } else {
+            return Err(RocketMQError::broker_operation_failed(
+                "UPDATE_CONTROLLER_CONFIG",
+                ResponseCode::ControllerInvalidRequest.to_i32(),
+                "Request body is empty".to_string(),
+            ));
+        };
+
+        // Validate blacklist configs
+        if self.validate_blacklist_config_exist(&properties) {
+            let error_msg = "Cannot update blacklisted configurations".to_string();
+            return Ok(Some(RemotingCommand::create_response_command_with_code_remark(
+                ResponseCode::NoPermission,
+                error_msg,
+            )));
+        }
+
+        // Check if this controller is the leader
+        if !self.controller_manager.controller().is_leader() {
+            let error_msg = "Only leader controller can handle configuration updates".to_string();
+            return Ok(Some(RemotingCommand::create_response_command_with_code_remark(
+                ResponseCode::ControllerNotLeader,
+                error_msg,
+            )));
+        }
+
+        let mut controller_config = self.controller_manager.controller_config().clone();
+
+        // Update configuration
+        match controller_config.update(&properties) {
+            Ok(_) => {
+                // Persist the updated configuration
+                if let Err(e) = controller_config.persist() {
+                    let error_msg = format!("Failed to persist configuration: {}", e);
+                    return Ok(Some(RemotingCommand::create_response_command_with_code_remark(
+                        ResponseCode::SystemError,
+                        error_msg,
+                    )));
+                }
+                Ok(Some(RemotingCommand::create_response_command()))
+            }
+            Err(e) => {
+                let error_msg = format!("Failed to update configuration: {}", e);
+                Ok(Some(RemotingCommand::create_response_command_with_code_remark(
+                    ResponseCode::ControllerInvalidRequest,
+                    error_msg,
+                )))
+            }
+        }
+    }
+
+    /// Helper function to parse properties from string format
+    ///
+    /// # Arguments
+    ///
+    /// * `body` - Request body containing properties in "key=value" format
+    ///
+    /// # Returns
+    ///
+    /// Result containing parsed properties map
+    fn parse_properties_from_string(&self, body: &[u8]) -> RocketMQResult<std::collections::HashMap<String, String>> {
+        let content = String::from_utf8_lossy(body);
+        let mut properties = std::collections::HashMap::new();
+
+        for line in content.lines() {
+            if let Some((key, value)) = line.split_once('=') {
+                let trimmed_key = key.trim();
+                if !trimmed_key.is_empty() {
+                    properties.insert(trimmed_key.to_string(), value.trim().to_string());
+                }
+            }
+        }
+
+        Ok(properties)
     }
 
     /// Handle GET_CONTROLLER_CONFIG request
@@ -573,6 +658,8 @@ impl RequestProcessor for ControllerRequestProcessor {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use rocketmq_common::common::controller::ControllerConfig;
+    use std::collections::HashMap;
 
     #[test]
     fn test_config_blacklist() {
@@ -587,4 +674,73 @@ mod tests {
         assert!(blacklist.contains("configStorePath"));
         assert!(blacklist.contains("rocketmqHome"));
     }
+
+    #[test]
+    fn test_parse_properties_from_string() {
+        // Test parse_properties_from_string method
+        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
+        let controller_manager = rt.block_on(async {
+            crate::controller::controller_manager::ControllerManager::new(ControllerConfig::default())
+                .await
+                .unwrap()
+        });
+        let processor = ControllerRequestProcessor::new(rocketmq_rust::ArcMut::new(controller_manager));
+
+        // Test valid properties
+        let body =
+            "scanNotActiveBrokerInterval=10000\ncontrollerThreadPoolNums=32\nenableElectUncleanMaster=true".as_bytes();
+        let properties = processor.parse_properties_from_string(body).unwrap();
+
+        assert_eq!(
+            properties.get("scanNotActiveBrokerInterval"),
+            Some(&"10000".to_string())
+        );
+        assert_eq!(properties.get("controllerThreadPoolNums"), Some(&"32".to_string()));
+        assert_eq!(properties.get("enableElectUncleanMaster"), Some(&"true".to_string()));
+
+        // Test empty body
+        let empty_body = "".as_bytes();
+        let empty_properties = processor.parse_properties_from_string(empty_body).unwrap();
+        assert!(empty_properties.is_empty());
+
+        // Test body with whitespace
+        let body_with_whitespace =
+            "  scanNotActiveBrokerInterval  =  10000  \n  controllerThreadPoolNums  =  32  ".as_bytes();
+        let properties_with_whitespace = processor.parse_properties_from_string(body_with_whitespace).unwrap();
+
+        assert_eq!(
+            properties_with_whitespace.get("scanNotActiveBrokerInterval"),
+            Some(&"10000".to_string())
+        );
+        assert_eq!(
+            properties_with_whitespace.get("controllerThreadPoolNums"),
+            Some(&"32".to_string())
+        );
+    }
+
+    #[test]
+    fn test_validate_blacklist_config_exist() {
+        // Test validate_blacklist_config_exist method
+        let rt = tokio::runtime::Builder::new_current_thread().build().unwrap();
+        let controller_manager = rt.block_on(async {
+            crate::controller::controller_manager::ControllerManager::new(ControllerConfig::default())
+                .await
+                .unwrap()
+        });
+        let processor = ControllerRequestProcessor::new(rocketmq_rust::ArcMut::new(controller_manager));
+
+        // Test with blacklisted config
+        let mut properties_with_blacklist = HashMap::new();
+        properties_with_blacklist.insert("configBlackList".to_string(), "newBlacklist".to_string());
+        assert!(processor.validate_blacklist_config_exist(&properties_with_blacklist));
+
+        // Test with non-blacklisted config
+        let mut properties_without_blacklist = HashMap::new();
+        properties_without_blacklist.insert("scanNotActiveBrokerInterval".to_string(), "10000".to_string());
+        assert!(!processor.validate_blacklist_config_exist(&properties_without_blacklist));
+
+        // Test with empty properties
+        let empty_properties = HashMap::new();
+        assert!(!processor.validate_blacklist_config_exist(&empty_properties));
+    }
 }