[ISSUE #6278]🚀Implement BrokerConsumeStats command in rocketmq-admin-core

Cargo.toml

@@ -15,25 +15,25 @@
 
 [workspace]
 members = [
-    "rocketmq",
-    "rocketmq-auth",
-    "rocketmq-broker",
-    "rocketmq-client",
-    "rocketmq-common",
-    "rocketmq-controller",
-    "rocketmq-error",
-    "rocketmq-filter",
-    "rocketmq-macros",
-    "rocketmq-namesrv",
-    "rocketmq-proxy",
-    "rocketmq-remoting",
-    "rocketmq-runtime",
-    "rocketmq-store",
-    "rocketmq-tools/rocketmq-admin/rocketmq-admin-cli",
-    "rocketmq-tools/rocketmq-admin/rocketmq-admin-core",
-    "rocketmq-tools/rocketmq-admin/rocketmq-admin-tui",
-    "rocketmq-tools/rocketmq-store-inspect",
-    "rocketmq-dashboard/rocketmq-dashboard-common"
+  "rocketmq",
+  "rocketmq-auth",
+  "rocketmq-broker",
+  "rocketmq-client",
+  "rocketmq-common",
+  "rocketmq-controller",
+  "rocketmq-error",
+  "rocketmq-filter",
+  "rocketmq-macros",
+  "rocketmq-namesrv",
+  "rocketmq-proxy",
+  "rocketmq-remoting",
+  "rocketmq-runtime",
+  "rocketmq-store",
+  "rocketmq-tools/rocketmq-admin/rocketmq-admin-cli",
+  "rocketmq-tools/rocketmq-admin/rocketmq-admin-core",
+  "rocketmq-tools/rocketmq-admin/rocketmq-admin-tui",
+  "rocketmq-tools/rocketmq-store-inspect",
+  "rocketmq-dashboard/rocketmq-dashboard-common",
 ]
 resolver = "2"
 
@@ -45,11 +45,11 @@ homepage = "https://github.com/mxsm/rocketmq-rust"
 repository = "https://github.com/mxsm/rocketmq-rust"
 license = "MIT OR Apache-2.0"
 keywords = [
-    "apache-rocketmq",
-    "rocketmq-rust",
-    "rocketmq-rs",
-    "rust",
-    "rocketmq-client",
+  "apache-rocketmq",
+  "rocketmq-rust",
+  "rocketmq-rs",
+  "rust",
+  "rocketmq-client",
 ]
 categories = ["asynchronous", "network-programming", "development-tools"]
 readme = "README.md"
@@ -58,64 +58,63 @@ Unofficial Rust implementation of Apache RocketMQ
 """
 rust-version = "1.85.0"
 [workspace.dependencies]
-rocketmq-common = { path = "./rocketmq-common" }
-rocketmq-runtime = { path = "./rocketmq-runtime" }
-rocketmq-macros = { path = "./rocketmq-macros" }
-rocketmq-rust = { path = "./rocketmq" }
-rocketmq-filter = { path = "./rocketmq-filter" }
-rocketmq-store = { path = "./rocketmq-store", default-features = true }
-rocketmq-remoting = { path = "./rocketmq-remoting" }
+rocketmq-common      = { path = "./rocketmq-common" }
+rocketmq-runtime     = { path = "./rocketmq-runtime" }
+rocketmq-macros      = { path = "./rocketmq-macros" }
+rocketmq-rust        = { path = "./rocketmq" }
+rocketmq-filter      = { path = "./rocketmq-filter" }
+rocketmq-store       = { path = "./rocketmq-store", default-features = true }
+rocketmq-remoting    = { path = "./rocketmq-remoting" }
 rocketmq-client-rust = { path = "./rocketmq-client" }
-rocketmq-tools = { path = "./rocketmq-tools/rocketmq-admin/rocketmq-admin-core", package = "rocketmq-admin-core" }
-rocketmq-error = { path = "./rocketmq-error" }
-rocketmq-auth = { path = "./rocketmq-auth" }
+rocketmq-error       = { path = "./rocketmq-error" }
+rocketmq-auth        = { path = "./rocketmq-auth" }
 
-tokio = { version = "1.49", features = ["full"] }
-tokio-util = { version = "0.7.18", features = ["full"] }
+tokio        = { version = "1.49", features = ["full"] }
+tokio-util   = { version = "0.7.18", features = ["full"] }
 tokio-stream = { version = "0.1.18", features = ["full"] }
 
-tracing = "0.1.44"
+tracing            = "0.1.44"
 tracing-subscriber = { version = "0.3.22", features = ["time"] }
-tracing-appender = "0.2.4"
+tracing-appender   = "0.2.4"
 
 thiserror = "2.0.18"
 
 
 #serde
-serde = { version = "1.0", features = ["derive"] }
-serde_json = "1.0"
-serde_yaml = "0.9"
+serde              = { version = "1.0", features = ["derive"] }
+serde_json         = "1.0"
+serde_yaml         = "0.9"
 serde_json_any_key = "2.0.0"
-anyhow = "1.0.101"
-bytes = "1.11.1"
-rand = "0.10.0"
-num_cpus = "1.17"
+anyhow             = "1.0.101"
+bytes              = "1.11.1"
+rand               = "0.10.0"
+num_cpus           = "1.17"
 
 config = "0.15.19"
 
-parking_lot = "0.12"
-dirs = "6.0"
+parking_lot   = "0.12"
+dirs          = "6.0"
 trait-variant = "0.1.2"
 
 mockall = "0.14.0"
-cfg-if = "1.0.4"
+cfg-if  = "1.0.4"
 
 sysinfo = "0.38.2"
 uuid = { version = "1.21.0", features = [
-    "v4", # Lets you generate random UUIDs
-    "fast-rng", # Use a faster (but still sufficiently random) RNG
-    "macro-diagnostics",
+  "v4",                # Lets you generate random UUIDs
+  "fast-rng",          # Use a faster (but still sufficiently random) RNG
+  "macro-diagnostics",
 ] }
 
 
 futures = "0.3"
 
 cheetah-string = { version = "1.0.1", features = ["serde", "bytes", "simd"] }
 
-flate2 = "1.1.9"
-dashmap = "6.1.0"
-strum = { version = "0.27.2", features = ["derive"] }
+flate2   = "1.1.9"
+dashmap  = "6.1.0"
+strum    = { version = "0.27.2", features = ["derive"] }
 smallvec = "1.13"
 
-opentelemetry = { version = "0.31", features = ["metrics"] }
+opentelemetry     = { version = "0.31", features = ["metrics"] }
 opentelemetry_sdk = { version = "0.31", features = ["metrics", "rt-tokio"] }

rocketmq-broker/Cargo.toml

@@ -41,7 +41,7 @@ rocketmq-filter      = { workspace = true }
 rocketmq-runtime     = { workspace = true }
 rocketmq-client-rust = { workspace = true }
 rocketmq-error       = { workspace = true }
-rocketmq-auth       = { workspace = true }
+rocketmq-auth        = { workspace = true }
 
 anyhow.workspace = true
 

rocketmq-dashboard/rocketmq-dashboard-common/Cargo.toml

@@ -28,22 +28,12 @@ description            = "Common shared code for RocketMQ Dashboard implementati
 rust-version.workspace = true
 
 [dependencies]
-rocketmq-common = { workspace = true }
-rocketmq-client-rust = { workspace = true }
-rocketmq-remoting = { workspace = true }
-
-# Async runtime
-tokio = { workspace = true }
 
 # Async trait support
 async-trait = "0.1"
 
 # Serialization
 serde = { workspace = true }
-serde_json = { workspace = true }
-
-# Logging
-tracing = { workspace = true }
 
 # Error handling
 anyhow = { workspace = true }

rocketmq-example/Cargo.toml

@@ -17,17 +17,23 @@
 # This is a standalone project, not part of the parent workspace
 
 [package]
-name           = "rocketmq-example"
-version        = "0.8.0"
-authors        = ["mxsm <mxsm@apache.org>"]
-edition        = "2024"
-homepage       = "https://github.com/mxsm/rocketmq-rust"
-repository     = "https://github.com/mxsm/rocketmq-rust"
-license        = "MIT OR Apache-2.0"
-keywords       = ["apache-rocketmq", "rocketmq-rust", "rocketmq-rs", "rust", "rocketmq-client"]
-readme         = "README.md"
-description    = "Examples for RocketMQ Rust implementation"
-rust-version   = "1.85.0"
+name = "rocketmq-example"
+version = "0.8.0"
+authors = ["mxsm <mxsm@apache.org>"]
+edition = "2024"
+homepage = "https://github.com/mxsm/rocketmq-rust"
+repository = "https://github.com/mxsm/rocketmq-rust"
+license = "MIT OR Apache-2.0"
+keywords = [
+  "apache-rocketmq",
+  "rocketmq-rust",
+  "rocketmq-rs",
+  "rust",
+  "rocketmq-client",
+]
+readme = "README.md"
+description = "Examples for RocketMQ Rust implementation"
+rust-version = "1.85.0"
 resolver = "3"
 
 [dependencies]
@@ -68,4 +74,4 @@ path = "examples/producer/send_to_queue.rs"
 
 [[example]]
 name = "producer-send-with-selector"
-path = "examples/producer/send_with_selector.rs"
+path = "examples/producer/send_with_selector.rs"

rocketmq-remoting/Cargo.toml

@@ -29,10 +29,10 @@ rust-version.workspace = true
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-rocketmq-common  = { workspace = true }
-rocketmq-macros  = { workspace = true }
-rocketmq-rust    = { workspace = true }
-rocketmq-error   = { workspace = true }
+rocketmq-common = { workspace = true }
+rocketmq-macros = { workspace = true }
+rocketmq-rust   = { workspace = true }
+rocketmq-error  = { workspace = true }
 
 anyhow.workspace = true
 bytes.workspace  = true

rocketmq-tools/rocketmq-admin/rocketmq-admin-cli/Cargo.toml

@@ -14,23 +14,20 @@
 
 
 [package]
-name = "rocketmq-admin-cli"
-version.workspace = true
-authors.workspace = true
-edition = "2024"
-homepage.workspace = true
-repository.workspace = true
-license.workspace = true
-keywords = ["rocketmq", "cli", "admin", "messaging", "rocketmq-rust"]
-categories = ["command-line-utilities", "development-tools"]
-readme = "README.md"
-description = "Command-line interface for managing RocketMQ-Rust and Apache RocketMQ clusters"
+name                   = "rocketmq-admin-cli"
+version.workspace      = true
+authors.workspace      = true
+edition                = "2024"
+homepage.workspace     = true
+repository.workspace   = true
+license.workspace      = true
+keywords               = ["rocketmq", "cli", "admin", "messaging", "rocketmq-rust"]
+categories             = ["command-line-utilities", "development-tools"]
+readme                 = "README.md"
+description            = "Command-line interface for managing RocketMQ-Rust and Apache RocketMQ clusters"
 rust-version.workspace = true
 
 [dependencies]
-rocketmq-tools = { workspace = true }
-
-tokio = { workspace = true, features = ["full"] }
 
 [[bin]]
 name = "rocketmq-admin-cli"

rocketmq-tools/rocketmq-admin/rocketmq-admin-core/Cargo.toml

@@ -47,6 +47,7 @@ tokio                = { workspace = true }
 tracing              = { workspace = true }
 
 cheetah-string = { workspace = true }
+chrono         = "0.4"
 clap           = { version = "4.5.59", features = ["derive"] }
 clap_complete  = "4.5"
 tabled         = { version = "0.20.0", features = ["derive"] }

rocketmq-tools/rocketmq-admin/rocketmq-admin-core/src/commands.rs

@@ -177,6 +177,11 @@ impl CommandExecute for ClassificationTablePrint {
                 command: "updateAcl",
                 remark: "Update ACL.",
             },
+            Command {
+                category: "Broker",
+                command: "brokerConsumeStats",
+                remark: "Fetch broker consume stats data.",
+            },
             Command {
                 category: "Broker",
                 command: "brokerStatus",

rocketmq-tools/rocketmq-admin/rocketmq-admin-core/src/commands/broker_commands.rs

@@ -1,4 +1,4 @@
-// Copyright 2026 The RocketMQ Rust Authors
+// Copyright 2023 The RocketMQ Rust Authors
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+mod broker_consume_stats_sub_command;
 mod broker_status_sub_command;
 mod clean_expired_cq_sub_command;
 mod clean_unused_topic_sub_command;
@@ -32,6 +33,7 @@ use clap::Subcommand;
 use rocketmq_error::RocketMQResult;
 use rocketmq_remoting::runtime::RPCHook;
 
+use crate::commands::broker_commands::broker_consume_stats_sub_command::BrokerConsumeStatsSubCommand;
 use crate::commands::broker_commands::broker_status_sub_command::BrokerStatusSubCommand;
 use crate::commands::broker_commands::clean_expired_cq_sub_command::CleanExpiredCQSubCommand;
 use crate::commands::broker_commands::clean_unused_topic_sub_command::CleanUnusedTopicSubCommand;
@@ -49,6 +51,13 @@ use crate::commands::CommandExecute;
 
 #[derive(Subcommand)]
 pub enum BrokerCommands {
+    #[command(
+        name = "brokerConsumeStats",
+        about = "Fetch broker consume stats data.",
+        long_about = None,
+    )]
+    BrokerConsumeStats(BrokerConsumeStatsSubCommand),
+
     #[command(
         name = "brokerStatus",
         about = "Fetch broker runtime status data.",
@@ -144,6 +153,7 @@ pub enum BrokerCommands {
 impl CommandExecute for BrokerCommands {
     async fn execute(&self, rpc_hook: Option<Arc<dyn RPCHook>>) -> RocketMQResult<()> {
         match self {
+            BrokerCommands::BrokerConsumeStats(value) => value.execute(rpc_hook).await,
             BrokerCommands::BrokerStatus(cmd) => cmd.execute(rpc_hook).await,
             BrokerCommands::CleanExpiredCQ(value) => value.execute(rpc_hook).await,
             BrokerCommands::CleanUnusedTopic(value) => value.execute(rpc_hook).await,