mxsm · rocketmq-rust-bot · May 25, 2025 · May 25, 2025 · coderabbitai · May 25, 2025
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/rocketmq-client/src/admin/default_mq_admin_ext_impl.rs b/rocketmq-client/src/admin/default_mq_admin_ext_impl.rs
@@ -657,7 +657,16 @@
         name_servers: Vec<CheetahString>,
     ) -> rocketmq_error::RocketMQResult<HashMap<CheetahString, HashMap<CheetahString, CheetahString>>>
     {
-        todo!()
+        Ok(self
+            .client_instance
+            .as_ref()
+            .unwrap()
+            .mq_client_api_impl
+            .as_ref()
+            .unwrap()
+            .get_name_server_config(Some(name_servers), self.timeout_millis)
+            .await?
+            .unwrap_or_default())
     }
 
     async fn resume_check_half_message(

diff --git a/rocketmq-client/src/implementation/mq_client_api_impl.rs b/rocketmq-client/src/implementation/mq_client_api_impl.rs
@@ -18,6 +18,7 @@
 use std::collections::HashSet;
 use std::sync::atomic::AtomicU32;
 use std::sync::Arc;
+use std::time::Duration;
 use std::time::Instant;
 
 use cheetah_string::CheetahString;
@@ -36,11 +37,14 @@
 use rocketmq_common::common::namesrv::top_addressing::TopAddressing;
 use rocketmq_common::common::sys_flag::pull_sys_flag::PullSysFlag;
 use rocketmq_common::common::topic::TopicValidator;
+use rocketmq_common::utils::serde_json_utils::SerdeJsonUtils;
 use rocketmq_common::MessageDecoder;
 use rocketmq_error::client_broker_err;
 use rocketmq_error::mq_client_err;
+use rocketmq_error::ClientErr;
 use rocketmq_error::MQBrokerErr;
 use rocketmq_error::RocketMQResult;
+use rocketmq_error::RocketmqError;
 use rocketmq_error::RocketmqError::MQClientBrokerError;
 use rocketmq_remoting::base::connection_net_event::ConnectionNetEvent;
 use rocketmq_remoting::clients::rocketmq_default_impl::RocketmqDefaultClient;
@@ -1746,6 +1750,73 @@
         }
         Ok(())
     }
+
+    pub async fn get_name_server_config(
+        &self,
+        name_servers: Option<Vec<CheetahString>>,
+        timeout_millis: Duration,
+    ) -> rocketmq_error::RocketMQResult<
+        Option<HashMap<CheetahString, HashMap<CheetahString, CheetahString>>>,
+    > {
+        // Determine which name servers to invoke
+        let invoke_name_servers = match name_servers {
+            Some(servers) if !servers.is_empty() => servers,
+            _ => self.remoting_client.get_name_server_address_list().to_vec(),
+        };
+
+        if invoke_name_servers.is_empty() {
+            return Ok(None);
+        }
+
+        // Create request command
+        let request = RemotingCommand::create_remoting_command(RequestCode::GetNamesrvConfig);
+        let mut config_map = HashMap::with_capacity(4);
+        // Iterate through each name server
+        for name_server in invoke_name_servers {
+            // Make synchronous call with timeout
+            let response = self
+                .remoting_client
+                .invoke_async(
+                    Some(&name_server),
+                    request.clone(),
+                    timeout_millis.as_millis() as u64,
+                )
+                .await?;
+            // Check response code
+            match ResponseCode::from(response.code()) {
+                ResponseCode::Success => {
+                    // Parse response body as properties
+                    match response.get_body() {
+                        Some(body) => {
+                            let body_str = String::from_utf8_lossy(body.as_ref()).to_string();
+                            // if body_str contains =, return from Java version
+                            let properties = if body_str.contains('=') {
+                                mix_all::string_to_properties(&body_str).unwrap_or_default()
+                            } else {
+                                SerdeJsonUtils::from_json_str::<
+                                    HashMap<CheetahString, CheetahString>,
+                                >(&body_str).expect("failed to parse JSON")
+                            };
-                                SerdeJsonUtils::from_json_str::<
-                                    HashMap<CheetahString, CheetahString>,
-                                >(&body_str).expect("failed to parse JSON")
-                            };
+                            } else {
+                                match SerdeJsonUtils::from_json_str::<
+                                    HashMap<CheetahString, CheetahString>,
+                                >(&body_str) {
+                                    Ok(json_props) => json_props,
+                                    Err(_) => {
+                                        return Err(RocketmqError::MQClientErr(ClientErr::new(
+                                            format!("Failed to parse JSON response from server {}", name_server)
+                                        )));
+                                    }
+                                }
+                            };
-                                SerdeJsonUtils::from_json_str::<
-                                    HashMap<CheetahString, CheetahString>,
-                                >(&body_str).expect("failed to parse JSON")
-                            };
+                            } else {
+                                match SerdeJsonUtils::from_json_str::<
+                                    HashMap<CheetahString, CheetahString>,
+                                >(&body_str) {
+                                    Ok(json_props) => json_props,
+                                    Err(_) => {
+                                        return Err(RocketmqError::MQClientErr(ClientErr::new(
+                                            format!("Failed to parse JSON response from server {}", name_server)
+                                        )));
+                                    }
+                                }
+                            };
+
+                            config_map.insert(name_server.clone(), properties);
+                        }
+                        None => {
+                            return Err(RocketmqError::MQClientErr(ClientErr::new(
+                                "Body is empty".to_string(),
+                            )))
+                        }
+                    }
+                }
+                code => {
+                    return Err(RocketmqError::MQClientErr(ClientErr::new_with_code(
+                        response.code(),
+                        response.remark().map_or("".to_string(), |s| s.to_string()),
+                    )));
+                }
+            }
+        }
+        Ok(Some(config_map))
+    }
 }
 
 fn build_queue_offset_sorted_map(

diff --git a/rocketmq-tools/Cargo.toml b/rocketmq-tools/Cargo.toml
@@ -20,11 +20,12 @@ rocketmq-error = { workspace = true }
 
 serde.workspace = true
 serde_json.workspace = true
-tokio = {workspace = true}
+tokio = { workspace = true }
 
 cheetah-string = { workspace = true }
 clap = { version = "4.5.38", features = ["derive"] }
 tabled = { version = &
9E7A
quot;0.19.0", features = ["derive"] }
+futures = "0.3.31"
-futures = "0.3.31"
-futures = "0.3.31"
 
 [[bin]]
 name = "rocketmq-admin-cli-rust"

diff --git a/rocketmq-tools/src/commands/namesrv_commands/get_namesrv_config_command.rs b/rocketmq-tools/src/commands/namesrv_commands/get_namesrv_config_command.rs
@@ -14,6 +14,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+use std::collections::HashMap;
 use std::sync::Arc;
 
 use cheetah_string::CheetahString;
@@ -22,6 +23,12 @@
 use rocketmq_common::TimeUtils::get_current_millis;
 use rocketmq_error::RocketMQResult;
 use rocketmq_remoting::runtime::RPCHook;
+use tabled::settings::object::Rows;
+use tabled::settings::Alignment;
+use tabled::settings::Modify;
+use tabled::settings::Style;
+use tabled::Table;
+use tabled::Tabled;
 
 use crate::admin::default_mq_admin_ext::DefaultMQAdminExt;
 use crate::commands::CommandExecute;
@@ -68,12 +75,57 @@
         let server_list = self.parse_server_list();
         if let Some(server_list) = server_list {
             admin.start().await?;
-            let _ = admin.get_name_server_config(server_list).await;
+            let configs = admin.get_name_server_config(server_list).await?;
+            display_configs_with_table(&configs);
+            admin.shutdown().await;
+            return Ok(());
         } else {
             eprintln!("Please set the namesrvAddr parameter");
-            return Ok(());
         }
 
-        unimplemented!("GetNamesrvConfigCommand is not implemented yet");
+        Ok(())
+    }
+}
+
+/// Configuration entry for table display
+#[derive(Debug, Clone, Tabled)]
+struct ConfigEntry {
+    #[tabled(rename = "Configuration Key")]
+    key: CheetahString,
+    #[tabled(rename = "Value")]
+    value: CheetahString,
+}
+
+/// Display configurations using tabled for formatted output
+fn display_configs_with_table(
+    configs: &HashMap<CheetahString, HashMap<CheetahString, CheetahString>>,
+) {
+    for (server_addr, properties) in configs {
+        println!("============Name server: {server_addr}============",);
+
+        // Convert properties to ConfigEntry vector
+        let mut config_entries: Vec<ConfigEntry> = properties
+            .iter()
+            .map(|(key, value)| ConfigEntry {
+                key: key.clone(),
+                value: value.clone(),
+            })
+            .collect();
+
+        // Sort by key for consistent output
+        config_entries.sort_by(|a, b| a.key.cmp(&b.key));
+
+        // Create and display table
+        if !config_entries.is_empty() {
+            let table = Table::new(&config_entries)
+                .with(Style::modern())
+                .with(Modify::new(Rows::new(1..)).with(Alignment::left()))
+                .to_string();
+
+            println!("{table}");
+        } else {
+            println!("No configuration found for this server.");
+        }
+        println!(); // Add blank line between servers
     }
 }