Repository: robustmq/robustmq (language: rust)

Issue:
RBIP-32: MQTT supports slow subscription statistics
### What problem does the new feature solve?

Align EMQX

### What does the feature do?

https://docs.emqx.com/zh/emqx/latest/observability/slow-subscribers-statistics.html

### Implementation challenges

_No response_


Discussion / hints:
mark


Fix patch:
diff --git a/.gitignore b/.gitignore
index e2bad5521..75f3ffe78 100644
--- a/.gitignore
+++ b/.gitignore
@@ -10,4 +10,4 @@ logs
 nohup.out
 .idea
 /precommit_venv/
-robust-data*
\ No newline at end of file
+robust-data*
diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 8399763e2..defc5ac85 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -93,7 +93,7 @@ repos:
       - id: cargo-test
         name: cargo test
         description: unit test for the project
-        entry: bash -c 'cargo nextest run --workspace --exclude robustmq-test'
+        entry: bash -c 'make test'
         language: system
         files: \.rs$
         pass_filenames: false
diff --git a/Cargo.lock b/Cargo.lock
index f9e21a31e..d85be6964 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -104,9 +104,9 @@ dependencies = [
 
 [[package]]
 name = "anstyle"
-version = "1.0.5"
+version = "1.0.10"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "2faccea4cc4ab4a667ce676a30e8ec13922a692c99bb8f5b11f1502c72e04220"
+checksum = "55cc3b69f167a1ef2e161439aa98aed94e6028e5f9a59be9a6ffb47aef1651f9"
 
 [[package]]
 name = "anstyle-parse"
@@ -193,7 +193,7 @@ checksum = "16e62a023e7c117e27523144c5d2459f4397fcc3cab0085af8e2224f643a0193"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -204,7 +204,7 @@ checksum = "c980ee35e870bd1a4d2c8294d4c04d0499e67bca1e4b5cefcc693c2fa00caea9"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -457,7 +457,7 @@ dependencies = [
  "regex",
  "rustc-hash",
  "shlex",
- "syn 2.0.48",
+ "syn 2.0.90",
  "which",
 ]
 
@@ -514,10 +514,21 @@ dependencies = [
  "proc-macro-crate",
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
  "syn_derive",
 ]
 
+[[package]]
+name = "bstr"
+version = "1.11.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1a68f1f47cdf0ec8ee4b941b2eee2a80cb796db73118c0dd09ac63fbe405be22"
+dependencies = [
+ "memchr",
+ "regex-automata",
+ "serde",
+]
+
 [[package]]
 name = "btoi"
 version = "0.4.3"
@@ -695,7 +706,7 @@ dependencies = [
  "heck",
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -713,6 +724,7 @@ dependencies = [
  "common-base",
  "grpc-clients",
  "metadata-struct",
+ "prettytable-rs",
  "protocol",
  "serde_json",
  "thiserror",
@@ -888,6 +900,27 @@ dependencies = [
  "typenum",
 ]
 
+[[package]]
+name = "csv"
+version = "1.3.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "acdc4883a9c96732e4733212c01447ebd805833b7275a73ca3ee080fd77afdaf"
+dependencies = [
+ "csv-core",
+ "itoa",
+ "ryu",
+ "serde",
+]
+
+[[package]]
+name = "csv-core"
+version = "0.1.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5efa2b3d7902f4b634a20cae3c9c4e6209dc4779feb6863329607560143efa70"
+dependencies = [
+ "memchr",
+]
+
 [[package]]
 name = "darling"
 version = "0.20.8"
@@ -909,7 +942,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "strsim",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -920,7 +953,7 @@ checksum = "a668eda54683121533a393014d8692171709ff57a7d61f187b6e782719f8933f"
 dependencies = [
  "darling_core",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -981,7 +1014,7 @@ checksum = "cb7330aeadfbe296029522e6c40f315320aba36fc43a5b3632f3795348f3bd22"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
  "unicode-xid",
 ]
 
@@ -993,7 +1026,7 @@ checksum = "61bb5a1014ce6dfc2a378578509abe775a5aa06bff584a547555d9efdb81b926"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -1012,6 +1045,27 @@ dependencies = [
  "crypto-common",
 ]
 
+[[package]]
+name = "dirs-next"
+version = "2.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b98cf8ebf19c3d1b223e151f99a4f9f0690dca41414773390fc824184ac833e1"
+dependencies = [
+ "cfg-if",
+ "dirs-sys-next",
+]
+
+[[package]]
+name = "dirs-sys-next"
+version = "0.1.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4ebda144c4fe02d1f7ea1a7d9641b6fc6b580adcfa024ae48797ecdeb6825b4d"
+dependencies = [
+ "libc",
+ "redox_users",
+ "winapi",
+]
+
 [[package]]
 name = "downcast"
 version = "0.11.0"
@@ -1030,6 +1084,30 @@ version = "1.9.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a26ae43d7bcc3b814de94796a5e736d4029efb0ee900c12e2d54c993ad1a1e07"
 
+[[package]]
+name = "encode_unicode"
+version = "1.0.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "34aa73646ffb006b8f5147f3dc182bd4bcb190227ce861fc4a4844bf8e3cb2c0"
+
+[[package]]
+name = "encoding_rs"
+version = "0.8.35"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "75030f3c4f45dafd7586dd6780965a8c7e8e285a5ecb86713e63a79c5b2766f3"
+dependencies = [
+ "cfg-if",
+]
+
+[[package]]
+name = "encoding_rs_io"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "1cc3c5651fb62ab8aa3103998dade57efdd028544bd300516baa31840c252a83"
+dependencies = [
+ "encoding_rs",
+]
+
 [[package]]
 name = "env_logger"
 version = "0.10.2"
@@ -1149,7 +1227,7 @@ checksum = "b0fa992f1656e1707946bbba340ad244f0814009ef8c0118eb7b658395f19a2e"
 dependencies = [
  "frunk_proc_macro_helpers",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -1161,7 +1239,7 @@ dependencies = [
  "frunk_core",
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -1173,7 +1251,7 @@ dependencies = [
  "frunk_core",
  "frunk_proc_macro_helpers",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -1244,7 +1322,7 @@ checksum = "87750cf4b7a4c0625b1529e4c543c2182106e4dedc60a2a6455e00d212c489ac"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -1316,6 +1394,98 @@ version = "0.3.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d2fabcfbdc87f4758337ca535fb41a6d701b65693ce38287d856d1674551ec9b"
 
+[[package]]
+name = "globset"
+version = "0.4.15"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "15f1ce686646e7f1e19bf7d5533fe443a45dbfb990e00629110797578b42fb19"
+dependencies = [
+ "aho-corasick",
+ "bstr",
+ "log",
+ "regex-automata",
+ "regex-syntax",
+]
+
+[[package]]
+name = "grep"
+version = "0.3.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "308ae749734e28d749a86f33212c7b756748568ce332f970ac1d9cd8531f32e6"
+dependencies = [
+ "grep-cli",
+ "grep-matcher",
+ "grep-printer",
+ "grep-regex",
+ "grep-searcher",
+]
+
+[[package]]
+name = "grep-cli"
+version = "0.1.11"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "47f1288f0e06f279f84926fa4c17e3fcd2a22b357927a82f2777f7be26e4cec0"
+dependencies = [
+ "bstr",
+ "globset",
+ "libc",
+ "log",
+ "termcolor",
+ "winapi-util",
+]
+
+[[package]]
+name = "grep-matcher"
+version = "0.1.7"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "47a3141a10a43acfedc7c98a60a834d7ba00dfe7bec9071cbfc19b55b292ac02"
+dependencies = [
+ "memchr",
+]
+
+[[package]]
+name = "grep-printer"
+version = "0.2.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c112110ae4a891aa4d83ab82ecf734b307497d066f437686175e83fbd4e013fe"
+dependencies = [
+ "bstr",
+ "grep-matcher",
+ "grep-searcher",
+ "log",
+ "serde",
+ "serde_json",
+ "termcolor",
+]
+
+[[package]]
+name = "grep-regex"
+version = "0.1.13"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "9edd147c7e3296e7a26bd3a81345ce849557d5a8e48ed88f736074e760f91f7e"
+dependencies = [
+ "bstr",
+ "grep-matcher",
+ "log",
+ "regex-automata",
+ "regex-syntax",
+]
+
+[[package]]
+name = "grep-searcher"
+version = "0.1.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "b9b6c14b3fc2e0a107d6604d3231dec0509e691e62447104bc385a46a7892cda"
+dependencies = [
+ "bstr",
+ "encoding_rs",
+ "encoding_rs_io",
+ "grep-matcher",
+ "log",
+ "memchr",
+ "memmap2",
+]
+
 [[package]]
 name = "grpc-clients"
 version = "0.1.1"
@@ -1350,7 +1520,7 @@ dependencies = [
  "futures-sink",
  "futures-util",
  "http 0.2.12",
- "indexmap 2.2.1",
+ "indexmap 2.7.0",
  "slab",
  "tokio",
  "tokio-util",
@@ -1369,7 +1539,7 @@ dependencies = [
  "futures-sink",
  "futures-util",
  "http 1.0.0",
- "indexmap 2.2.1",
+ "indexmap 2.7.0",
  "slab",
  "tokio",
  "tokio-util",
@@ -1395,6 +1565,12 @@ dependencies = [
  "allocator-api2",
 ]
 
+[[package]]
+name = "hashbrown"
+version = "0.15.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "bf151400ff0baff5465007dd2f3e717f3fe502074ca563069ce3a6629d07b289"
+
 [[package]]
 name = "headers"
 version = "0.4.0"
@@ -1646,12 +1822,12 @@ dependencies = [
 
 [[package]]
 name = "indexmap"
-version = "2.2.1"
+version = "2.7.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "433de089bd45971eecf4668ee0ee8f4cec17db4f8bd8f7bc3197a6ce37aa7d9b"
+checksum = "62f822373a4fe84d4bb149bf54e584a7f4abec90e072ed49cda0edea5b95471f"
 dependencies = [
  "equivalent",
- "hashbrown 0.14.3",
+ "hashbrown 0.15.2",
 ]
 
 [[package]]
@@ -1792,6 +1968,16 @@ version = "0.2.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "4ec2a862134d2a7d32d7983ddcdd1c4923530833c9f2ea1a44fc5fa473989058"
 
+[[package]]
+name = "libredox"
+version = "0.1.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c0ff37bd590ca25063e35af745c343cb7a0271906fb7b37e4813e8f79f00268d"
+dependencies = [
+ "bitflags 2.4.2",
+ "libc",
+]
+
 [[package]]
 name = "librocksdb-sys"
 version = "0.16.0+8.10.0"
@@ -1849,9 +2035,9 @@ dependencies = [
 
 [[package]]
 name = "log"
-version = "0.4.20"
+version = "0.4.22"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "b5e6163cb8c49088c2c36f57875e58ccd8c87c7427f7fbd50ea6710b2f3f2e8f"
+checksum = "a7a70ba024b9dc04c27ea2f0c0548feb474ec5c54bba33a7f72f873a39d07b24"
 dependencies = [
  "serde",
 ]
@@ -1927,6 +2113,15 @@ version = "2.7.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "523dc4f511e55ab87b694dc30d0f820d60906ef06413f93d4d7a1385599cc149"
 
+[[package]]
+name = "memmap2"
+version = "0.9.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fd3f7eed9d3848f8b98834af67102b720745c4ec028fcd0aa0239277e7de374f"
+dependencies = [
+ "libc",
+]
+
 [[package]]
 name = "metadata-struct"
 version = "0.1.1"
@@ -2042,7 +2237,7 @@ dependencies = [
  "cfg-if",
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -2085,6 +2280,7 @@ dependencies = [
  "dashmap",
  "futures",
  "futures-util",
+ "grep",
  "grpc-clients",
  "ipnet",
  "lazy_static",
@@ -2163,7 +2359,7 @@ dependencies = [
  "proc-macro-error",
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
  "termcolor",
  "thiserror",
 ]
@@ -2370,7 +2566,7 @@ dependencies = [
  "proc-macro2",
  "quote",
  "semver",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -2396,7 +2592,7 @@ checksum = "a948666b637a0f465e8564c73e89d4dde00d72d4d473cc972f390fc3dcee7d9c"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -2531,7 +2727,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "e1d3afd2628e69da2be385eb6f2fd57c8ac7977ceeff6dc166ff1657b0e386a9"
 dependencies = [
  "fixedbitset",
- "indexmap 2.2.1",
+ "indexmap 2.7.0",
 ]
 
 [[package]]
@@ -2551,7 +2747,7 @@ checksum = "266c042b60c9c76b8d53061e52b2e0d1116abc57cefc8c5cd671619a56ac3690"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -2644,16 +2840,30 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "a41cf62165e97c7f814d2221421dbb9afcbcdb0a88068e5ea206e19951c2cbb5"
 dependencies = [
  "proc-macro2",
- "syn 2.0.48",
+ "syn 2.0.90",
+]
+
+[[package]]
+name = "prettytable-rs"
+version = "0.10.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "eea25e07510aa6ab6547308ebe3c036016d162b8da920dbb079e3ba8acf3d95a"
+dependencies = [
+ "csv",
+ "encode_unicode",
+ "is-terminal",
+ "lazy_static",
+ "term",
+ "unicode-width",
 ]
 
 [[package]]
 name = "proc-macro-crate"
-version = "3.1.0"
+version = "3.2.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "6d37c51ca738a55da99dc0c4a34860fd675453b8b36209178c2249bb13651284"
+checksum = "8ecf48c7ca261d60b74ab1a7b20da18bede46776b2e55535cb958eb595c5fa7b"
 dependencies = [
- "toml_edit",
+ "toml_edit 0.22.22",
 ]
 
 [[package]]
@@ -2682,9 +2892,9 @@ dependencies = [
 
 [[package]]
 name = "proc-macro2"
-version = "1.0.78"
+version = "1.0.92"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "e2422ad645d89c99f8f3e6b88a9fdeca7fabeac836b1002371c4367c8f984aae"
+checksum = "37d3544b3f2748c54e147655edb5025752e2303145b5aefb3c3ea2c78b973bb0"
 dependencies = [
  "unicode-ident",
 ]
@@ -2731,7 +2941,7 @@ dependencies = [
  "prost",
  "prost-types",
  "regex",
- "syn 2.0.48",
+ "syn 2.0.90",
  "tempfile",
  "which",
 ]
@@ -2746,7 +2956,7 @@ dependencies = [
  "itertools",
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -2805,9 +3015,9 @@ dependencies = [
 
 [[package]]
 name = "quote"
-version = "1.0.35"
+version = "1.0.37"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "291ec9ab5efd934aaf503a6466c5d5251535d108ee747472c3977cc5acc868ef"
+checksum = "b5b9d34b8991d19d98081b46eacdd8eb58c6f2b201139f7c5f643cc155a633af"
 dependencies = [
  "proc-macro2",
 ]
@@ -2857,11 +3067,22 @@ dependencies = [
  "bitflags 1.3.2",
 ]
 
+[[package]]
+name = "redox_users"
+version = "0.4.6"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "ba009ff324d1fc1b900bd1fdb31564febe58a8ccc8a6fdbb93b543d33b13ca43"
+dependencies = [
+ "getrandom",
+ "libredox",
+ "thiserror",
+]
+
 [[package]]
 name = "regex"
-version = "1.10.4"
+version = "1.11.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c117dbdfde9c8308975b6a18d71f3f385c89461f7b3fb054288ecf2a2058ba4c"
+checksum = "b544ef1b4eac5dc2db33ea63606ae9ffcfac26c1416a2806ae0bf5f56b201191"
 dependencies = [
  "aho-corasick",
  "memchr",
@@ -2871,9 +3092,9 @@ dependencies = [
 
 [[package]]
 name = "regex-automata"
-version = "0.4.5"
+version = "0.4.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "5bb987efffd3c6d0d8f5f89510bb458559eab11e4f869acb20bf845e016259cd"
+checksum = "809e8dc61f6de73b46c85f4c96486310fe304c434cfa43669d7b40f711150908"
 dependencies = [
  "aho-corasick",
  "memchr",
@@ -2882,9 +3103,9 @@ dependencies = [
 
 [[package]]
 name = "regex-syntax"
-version = "0.8.2"
+version = "0.8.5"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "c08c74e62047bb2de4ff487b251e4a92e24f48745648451635cec7d591162d9f"
+checksum = "2b15c43186be67a4fd63bee50d0303afffcef381492ebe2c5d87f324e1b8815c"
 
 [[package]]
 name = "rend"
@@ -3198,7 +3419,7 @@ checksum = "33c85360c95e7d137454dc81d9a4ed2b8efd8fbe19cee57357b32b9771fccb67"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -3249,7 +3470,7 @@ version = "0.9.34+deprecated"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "6a8b1a1a2ebf674015cc02edccce75287f1a0130d394307b36743c2f5d504b47"
 dependencies = [
- "indexmap 2.2.1",
+ "indexmap 2.7.0",
  "itoa",
  "ryu",
  "serde",
@@ -3408,9 +3629,9 @@ dependencies = [
 
 [[package]]
 name = "syn"
-version = "2.0.48"
+version = "2.0.90"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "0f3531638e407dfc0814761abb7c00a5b54992b849452a0646b7f65c9f770f3f"
+checksum = "919d3b74a5dd0ccd15aeb8f93e7006bd9e14c295087c9896a110f490752bcf31"
 dependencies = [
  "proc-macro2",
  "quote",
@@ -3426,7 +3647,7 @@ dependencies = [
  "proc-macro-error",
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -3454,6 +3675,17 @@ dependencies = [
  "windows-sys 0.52.0",
 ]
 
+[[package]]
+name = "term"
+version = "0.7.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "c59df8ac95d96ff9bede18eb7300b0fda5e5d8d90960e76f8e14ae765eedbf1f"
+dependencies = [
+ "dirs-next",
+ "rustversion",
+ "winapi",
+]
+
 [[package]]
 name = "termcolor"
 version = "1.4.1"
@@ -3495,7 +3727,7 @@ checksum = "fa0faa943b50f3db30a20aa7e265dbc66076993efed8463e8de414e5d06d3471"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -3599,7 +3831,7 @@ checksum = "5b8a1e28f2deaa14e508979454cb3a223b10b938b45af148bc0986de36f1923b"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -3669,14 +3901,14 @@ dependencies = [
  "serde",
  "serde_spanned",
  "toml_datetime",
- "toml_edit",
+ "toml_edit 0.21.0",
 ]
 
 [[package]]
 name = "toml_datetime"
-version = "0.6.5"
+version = "0.6.8"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3550f4e9685620ac18a50ed434eb3aec30db8ba93b0287467bca5826ea25baf1"
+checksum = "0dd7358ecb8fc2f8d014bf86f6f638ce72ba252a2c3a2572f2a795f1d23efb41"
 dependencies = [
  "serde",
 ]
@@ -3687,11 +3919,22 @@ version = "0.21.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d34d383cd00a163b4a5b85053df514d45bc330f6de7737edfe0a93311d1eaa03"
 dependencies = [
- "indexmap 2.2.1",
+ "indexmap 2.7.0",
  "serde",
  "serde_spanned",
  "toml_datetime",
- "winnow",
+ "winnow 0.5.35",
+]
+
+[[package]]
+name = "toml_edit"
+version = "0.22.22"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "4ae48d6208a266e853d946088ed816055e556cc6028c5e8e2b84d9fa5dd7c7f5"
+dependencies = [
+ "indexmap 2.7.0",
+ "toml_datetime",
+ "winnow 0.6.20",
 ]
 
 [[package]]
@@ -3731,7 +3974,7 @@ dependencies = [
  "proc-macro2",
  "prost-build",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -3786,7 +4029,7 @@ checksum = "34704c8d6ebcbc939824180af020566b01a7c01f80641264eba0999f6c2b6be7"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -3893,9 +4136,9 @@ checksum = "08f95100a766bf4f8f28f90d77e0a5461bbdb219042e7679bebe79004fed8d75"
 
 [[package]]
 name = "unicode-ident"
-version = "1.0.12"
+version = "1.0.14"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "3354b9ac3fae1ff6755cb6db53683adb661634f67557942dea4facebec0fee4b"
+checksum = "adb9e6ca4f869e1180728b7950e35922a7fc6397f7b641499e8f3ef06e50dc83"
 
 [[package]]
 name = "unicode-normalization"
@@ -3906,6 +4149,12 @@ dependencies = [
  "tinyvec",
 ]
 
+[[package]]
+name = "unicode-width"
+version = "0.1.14"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7dd6e30e90baa6f72411720665d41d89b9a3d039dc45b8faea1ddd07f617f6af"
+
 [[package]]
 name = "unicode-xid"
 version = "0.2.6"
@@ -3998,7 +4247,7 @@ dependencies = [
  "proc-macro-error",
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -4065,7 +4314,7 @@ dependencies = [
  "once_cell",
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
  "wasm-bindgen-shared",
 ]
 
@@ -4087,7 +4336,7 @@ checksum = "afc340c74d9005395cf9dd098506f7f44e38f2b4a21c6aaacf9a105ea5e1e836"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
  "wasm-bindgen-backend",
  "wasm-bindgen-shared",
 ]
@@ -4291,6 +4540,15 @@ dependencies = [
  "memchr",
 ]
 
+[[package]]
+name = "winnow"
+version = "0.6.20"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "36c1fec1a2bb5866f07c25f68c26e565c4c200aebb96d7e55710c19d3e8ac49b"
+dependencies = [
+ "memchr",
+]
+
 [[package]]
 name = "wyz"
 version = "0.5.1"
@@ -4317,7 +4575,7 @@ checksum = "9ce1b18ccd8e73a9321186f97e46f9f04b778851177567b1975109d26a08d2a6"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
@@ -4337,7 +4595,7 @@ checksum = "ce36e65b0d2999d2aafac989fb249189a141aee1f53c612c1f37d72631959f69"
 dependencies = [
  "proc-macro2",
  "quote",
- "syn 2.0.48",
+ "syn 2.0.90",
 ]
 
 [[package]]
diff --git a/Cargo.toml b/Cargo.toml
index 3a4c6cb19..5a05ace4c 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -13,6 +13,7 @@
 # limitations under the License.
 
 [workspace]
+name = "robustmq"
 members = [
     "src/mqtt-bridge/kafka",
     "src/mqtt-bridge/elasticsearch",
@@ -44,21 +45,44 @@ edition = "2021"
 license = "Apache-2.0"
 
 [workspace.dependencies]
-serde = { version = "1.0", features = ["derive"] }
-serde_json = "1.0"
-signal-hook = "0.3.17"
+## async lib
+futures = "0.3"
 tokio = { version = "1", features = ["full"] }
 tokio-util = { version = "0.7.9", features = ["codec"] }
+tokio-rustls = "0.26"
+## web lib
 axum = { version = "0.7.2", features = ["ws"] }
 axum-server = { version = "0.6.0", features = ["tls-rustls"] }
+axum-extra = { version = "0.9.3", features = ["typed-header"] }
+## storage lib
+rocksdb = "0.22.0"
+mysql = "*"
+## serde lib
+serde = { version = "1.0", features = ["derive"] }
+serde_json = "1.0"
+serde_yaml = "0.9"
+humantime-serde = "1.1.1"
+## log lib
+log4rs = "1.2.0"
+log = "0.4.0"
+tracing = "0.1.40"
+tracing-subscriber = { version = "0.3.0", features = ["env-filter"] }
+## cmd lib
+clap-cargo = "0.14.1"
+clap = { version = "4.4.7", features = ["derive"] }
+## unit test lib
+mockall = "0.13.1"
+## text handle lib
+regex = "1.10.4"
+grep = "0.3.2"
+## other
+signal-hook = "0.3.17"
 prometheus = "0.13.3"
 prometheus_exporter = "0.8"
 lazy_static = "^1.4"
-futures = "0.3"
 thiserror = "1"
 bytes = { version = "1", features = ["serde"] }
 futures-util = "0.3.30"
-rocksdb = "0.22.0"
 tonic = "0.11.0"
 tonic-build = "0.11.0"
 bincode = "1.3.3"
@@ -70,20 +94,12 @@ uuid = { version = "1.7.0", features = ["v4"] }
 mobc = "0.8.3"
 dashmap = { version = "6.0.1", features = ["serde"] }
 snowflake = "1.3.0"
-regex = "1.10.4"
 rumqttc = "0.24.0"
-axum-extra = { version = "0.9.3", features = ["typed-header"] }
 rustls-pemfile = "2"
-tokio-rustls = "0.26"
-humantime-serde = "1.1.1"
-mysql = "*"
 paho-mqtt = { version = "0.12.5", default-features = false, features = [
     "bundled",
     "vendored-ssl",
 ] }
-serde_yaml = "0.9"
-log4rs = "1.2.0"
-log = "0.4.0"
 ipnet = "2.3.0"
 os_info = "3.8.2"
 openraft = { git = "https://github.com/databendlabs/openraft.git", features = [
@@ -91,15 +107,11 @@ openraft = { git = "https://github.com/databendlabs/openraft.git", features = [
     "type-alias",
     "loosen-follower-log-revert",
 ] }
-tracing = "0.1.40"
-tracing-subscriber = { version = "0.3.0", features = ["env-filter"] }
 validator = { version = "0.18", features = ["derive"] }
 rand = "0.8.5"
-clap-cargo = "0.14.1"
-clap = { version = "4.4.7", features = ["derive"] }
+#format
+prettytable-rs = "^0.10"
 
-## unit test framework
-mockall = "0.13.1"
 
 
 ## workspaces members
diff --git a/_typos.toml b/_typos.toml
index bcc7e4ff8..3dd270d8a 100644
--- a/_typos.toml
+++ b/_typos.toml
@@ -21,6 +21,7 @@
 Identifer = "Identifer"
 
 # Ignore the unsubscribe term as it's used in the MQTT context.
+# todo fix spelling error
 UNSUBSCRIBE = "UNSUBSCRIBE"
 unsubscrible = "unsubscrible"
 SUBSCRIBLE = "SUBSCRIBLE"
diff --git a/config/log-config/mqtt-log4rs.yaml b/config/log-config/mqtt-log4rs.yaml
new file mode 100644
index 000000000..f3be48be0
--- /dev/null
+++ b/config/log-config/mqtt-log4rs.yaml
@@ -0,0 +1,63 @@
+# Copyright 2023 RobustMQ Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+appenders:
+  stdout:
+    kind: console
+    encoder:
+      pattern: "{d(%Y-%m-%d %H:%M:%S)} {f}-{L} {h({l})} {m}{n}"
+
+  server:
+    kind: rolling_file
+    path: "{$path}/server.log"
+    encoder:
+      pattern: "{d(%Y-%m-%d %H:%M:%S)} {h({l})} {m}{n}"
+    policy:
+      trigger:
+        kind: size
+        limit: 1 gb
+      roller:
+        kind: fixed_window
+        pattern: "{$path}/server-{}.log"
+        base: 0
+        count: 50
+
+  slow_sub:
+    kind: rolling_file
+    path: "{$path}/slow_sub.log"
+    encoder:
+      pattern: "{m}{n}"
+    policy:
+      trigger:
+        kind: size
+        limit: 1 gb
+      roller:
+        kind: fixed_window
+        pattern: "{$path}/slow_sub-{}.log"
+        base: 0
+        count: 50
+
+root:
+  level: info
+  appenders:
+    - stdout
+    - server
+
+
+loggers:
+  mqtt-broker::observability::slow:
+    level: info
+    appenders:
+      - slow_sub
+    additive: false
diff --git a/config/log4rs.yaml b/config/log4rs.yaml
index 176aa62a1..ba50fe0bd 100644
--- a/config/log4rs.yaml
+++ b/config/log4rs.yaml
@@ -32,7 +32,6 @@ appenders:
             pattern: "{$path}/server-{}.log"
             base: 0
             count: 50
-
 root:
   level: info
   appenders:
diff --git a/config/mqtt-server.toml b/config/mqtt-server.toml
index 5e892167c..a1eb5da38 100644
--- a/config/mqtt-server.toml
+++ b/config/mqtt-server.toml
@@ -55,5 +55,5 @@ default_password = "pwd123"
 storage_type = "memory"
 
 [log]
-log_config = "./config/log4rs.yaml"
+log_config = "./config/log-config/mqtt-log4rs.yaml"
 log_path = "./robust-data/mqtt-broker/logs"
diff --git a/deny.toml b/deny.toml
index 6231da742..cc3289683 100644
--- a/deny.toml
+++ b/deny.toml
@@ -114,6 +114,7 @@ allow = [
     "EPL-2.0",
     "OpenSSL",
     "EPL-1.0",
+    "Unicode-3.0",
 ]
 # The confidence threshold for detecting a license from license text.
 # The higher the value, the more closely the license text must be to the
diff --git a/example/mqtt-cluster/mqtt-server/node-1.toml b/example/mqtt-cluster/mqtt-server/node-1.toml
index 9de7d102e..5f169d625 100644
--- a/example/mqtt-cluster/mqtt-server/node-1.toml
+++ b/example/mqtt-cluster/mqtt-server/node-1.toml
@@ -55,5 +55,5 @@ default_password = "pwd123"
 storage_type = "memory"
 
 [log]
-log_config = "./config/log4rs.yaml"
+log_config = "./config/log-config/mqtt-log4rs.yaml"
 log_path = "./robust-data-test/mqtt-broker-1/logs"
diff --git a/example/mqtt-cluster/mqtt-server/node-2.toml b/example/mqtt-cluster/mqtt-server/node-2.toml
index b734b5cf4..801211022 100644
--- a/example/mqtt-cluster/mqtt-server/node-2.toml
+++ b/example/mqtt-cluster/mqtt-server/node-2.toml
@@ -55,6 +55,5 @@ default_password = "pwd123"
 storage_type = "memory"
 
 [log]
-log_config = "./config/log4rs.yaml"
+log_config = "./config/log-config/mqtt-log4rs.yaml"
 log_path = "./robust-data/mqtt-broker-2/logs"
-
diff --git a/example/mqtt-cluster/mqtt-server/node-3.toml b/example/mqtt-cluster/mqtt-server/node-3.toml
index 0d6e8b3da..ac8b90530 100644
--- a/example/mqtt-cluster/mqtt-server/node-3.toml
+++ b/example/mqtt-cluster/mqtt-server/node-3.toml
@@ -55,6 +55,5 @@ default_password = "pwd123"
 storage_type = "memory"
 
 [log]
-log_config = "./config/log4rs.yaml"
+log_config = "./config/log-config/mqtt-log4rs.yaml"
 log_path = "./robust-data/mqtt-broker-3/logs"
-
diff --git a/rust-toolchain.toml b/rust-toolchain.toml
new file mode 100644
index 000000000..74fc8b383
--- /dev/null
+++ b/rust-toolchain.toml
@@ -0,0 +1,17 @@
+# Copyright 2023 RobustMQ Team
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+#     http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+[toolchain]
+channel = "nightly-2024-11-08"
+components = [ "rustfmt", "rust-src", "miri", "rust-analyzer"]
diff --git a/src/cli-command/Cargo.toml b/src/cli-command/Cargo.toml
index 6b4a747ab..c28aef2ce 100644
--- a/src/cli-command/Cargo.toml
+++ b/src/cli-command/Cargo.toml
@@ -27,4 +27,5 @@ grpc-clients.workspace = true
 common-base.workspace = true
 metadata-struct.workspace = true
 protocol.workspace = true
-serde_json.workspace = true
\ No newline at end of file
+serde_json.workspace = true
+prettytable-rs.workspace = true
diff --git a/src/cli-command/src/mqtt.rs b/src/cli-command/src/mqtt.rs
index f5b835b7f..f1e35cc48 100644
--- a/src/cli-command/src/mqtt.rs
+++ b/src/cli-command/src/mqtt.rs
@@ -12,8 +12,10 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::str::FromStr;
 use std::sync::Arc;
 
+use common_base::enum_type::common_enum::SortType;
 use grpc_clients::mqtt::admin::call::{
     cluster_status, mqtt_broker_create_user, mqtt_broker_delete_user,
     mqtt_broker_enable_slow_subscribe, mqtt_broker_list_connection,
@@ -21,6 +23,7 @@ use grpc_clients::mqtt::admin::call::{
 };
 use grpc_clients::pool::ClientPool;
 use metadata_struct::mqtt::user::MqttUser;
+use prettytable::{row, Table};
 use protocol::broker_mqtt::broker_mqtt_admin::{
     ClusterStatusRequest, CreateUserRequest, DeleteUserRequest, EnableSlowSubscribeRequest,
     ListConnectionRequest, ListSlowSubscribeRequest, ListTopicRequest, ListUserRequest,
@@ -34,7 +37,7 @@ pub struct MqttCliCommandParam {
     pub action: MqttActionType,
 }
 
-#[derive(Clone)]
+#[derive(Debug, Clone, PartialEq)]
 pub enum MqttActionType {
     Status,
 
@@ -48,7 +51,7 @@ pub enum MqttActionType {
 
     // observability: slow-ub
     EnableSlowSubscribe(EnableSlowSubscribeRequest),
-    ListSlowSubscribe,
+    ListSlowSubscribe(ListSlowSubscribeRequest),
 
     ListTopic(ListTopicRequest),
 }
@@ -94,8 +97,9 @@ impl MqttBrokerCommand {
                 self.list_topic(&client_pool, params.clone(), request.clone())
                     .await;
             }
-            MqttActionType::ListSlowSubscribe => {
-                self.list_slow_subscribe(&client_pool, params.clone()).await;
+            MqttActionType::ListSlowSubscribe(ref request) => {
+                self.list_slow_subscribe(&client_pool, params.clone(), request.clone())
+                    .await;
             }
         }
     }
@@ -126,7 +130,7 @@ impl MqttBrokerCommand {
     ) {
         match mqtt_broker_create_user(client_pool, &grpc_addr(params.server), cli_request).await {
             Ok(_) => {
-                println!("Created successfully!",)
+                println!("Created successfully!")
             }
             Err(e) => {
                 println!("MQTT broker create user normal exception");
@@ -211,12 +215,55 @@ impl MqttBrokerCommand {
         }
     }
 
-    async fn list_slow_subscribe(&self, client_pool: &ClientPool, params: MqttCliCommandParam) {
-        let request = ListSlowSubscribeRequest {};
-        match mqtt_broker_list_slow_subscribe(client_pool, &grpc_addr(params.server), request).await
+    async fn list_slow_subscribe(
+        &self,
+        client_pool: &ClientPool,
+        params: MqttCliCommandParam,
+        cli_request: ListSlowSubscribeRequest,
+    ) {
+        let slow_subscribe_request = ListSlowSubscribeRequest {
+            sub_name: cli_request.sub_name,
+            list: cli_request.list,
+            client_id: cli_request.client_id,
+            topic: cli_request.topic,
+            sort: cli_request.sort,
+        };
+        let sort = slow_subscribe_request.sort.clone();
+        match mqtt_broker_list_slow_subscribe(
+            client_pool,
+            &grpc_addr(params.server),
+            slow_subscribe_request,
+        )
+        .await
         {
             Ok(data) => {
-                println!("{:?}", data)
+                // sort
+                let sort_type = SortType::from_str(sort.as_str()).unwrap_or(SortType::DESC);
+                let mut list_slow_sub_raw = data.list_slow_subscribe_raw;
+                match sort_type {
+                    SortType::ASC => list_slow_sub_raw.sort_by(|a, b| a.time_ms.cmp(&b.time_ms)),
+                    SortType::DESC => list_slow_sub_raw.sort_by(|a, b| b.time_ms.cmp(&a.time_ms)),
+                }
+                // format table
+                let mut table = Table::new();
+                table.add_row(row![
+                    "client_id",
+                    "topic",
+                    "sub_name",
+                    "time_ms",
+                    "create_time"
+                ]);
+                for raw in list_slow_sub_raw {
+                    table.add_row(row![
+                        raw.client_id,
+                        raw.topic,
+                        raw.sub_name,
+                        raw.time_ms,
+                        raw.create_time
+                    ]);
+                }
+                // output cmd
+                table.printstd()
             }
             Err(e) => {
                 println!("MQTT broker list slow subscribe info exception");
@@ -256,3 +303,125 @@ impl MqttBrokerCommand {
         }
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use common_base::error::common::CommonError;
+    use protocol::broker_mqtt::broker_mqtt_admin::{ListSlowSubScribeRaw, ListSlowSubscribeReply};
+
+    fn set_up_slow_sub_config() -> Result<ListSlowSubscribeReply, CommonError> {
+        let mut list_slow_sub_raw: Vec<ListSlowSubScribeRaw> = Vec::new();
+        let raw1 = ListSlowSubScribeRaw {
+            client_id: "ed280344fec44aad8a78b00ff1dec99a".to_string(),
+            topic: "/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e".to_string(),
+            time_ms: 543,
+            node_info: "RobustMQ-MQTT@172.22.194.185".to_string(),
+            create_time: 1733898597,
+            sub_name: "/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e".to_string(),
+        };
+        list_slow_sub_raw.push(raw1);
+        let raw3 = ListSlowSubScribeRaw {
+            client_id: "49e10a8d8a494cefa904a00dcf0b30af".to_string(),
+            topic: "/request/131edb8526804e80b32b387fa2340d35".to_string(),
+            time_ms: 13,
+            node_info: "RobustMQ-MQTT@172.22.194.185".to_string(),
+            create_time: 1733898601,
+            sub_name: "/request/131edb8526804e80b32b387fa2340d35".to_string(),
+        };
+        list_slow_sub_raw.push(raw3);
+        let raw2 = ListSlowSubScribeRaw {
+            client_id: "49e10a8d8a494cefa904a00dcf0b30af".to_string(),
+            topic: "/request/131edb8526804e80b32b387fa2340d35".to_string(),
+            time_ms: 273,
+            node_info: "RobustMQ-MQTT@172.22.194.185".to_string(),
+            create_time: 1733898601,
+            sub_name: "/request/131edb8526804e80b32b387fa2340d35".to_string(),
+        };
+        list_slow_sub_raw.push(raw2);
+
+        Ok(ListSlowSubscribeReply {
+            list_slow_subscribe_raw: list_slow_sub_raw,
+        })
+    }
+    #[test]
+    fn test_get_sort_data_asc() {
+        let mut reply = set_up_slow_sub_config().unwrap();
+        reply
+            .list_slow_subscribe_raw
+            .sort_by(|a, b| a.time_ms.cmp(&b.time_ms));
+        assert_eq!(
+            ListSlowSubScribeRaw {
+                client_id: "49e10a8d8a494cefa904a00dcf0b30af".to_string(),
+                topic: "/request/131edb8526804e80b32b387fa2340d35".to_string(),
+                time_ms: 13,
+                node_info: "RobustMQ-MQTT@172.22.194.185".to_string(),
+                create_time: 1733898601,
+                sub_name: "/request/131edb8526804e80b32b387fa2340d35".to_string(),
+            },
+            reply.list_slow_subscribe_raw[0]
+        );
+        assert_eq!(
+            ListSlowSubScribeRaw {
+                client_id: "49e10a8d8a494cefa904a00dcf0b30af".to_string(),
+                topic: "/request/131edb8526804e80b32b387fa2340d35".to_string(),
+                time_ms: 273,
+                node_info: "RobustMQ-MQTT@172.22.194.185".to_string(),
+                create_time: 1733898601,
+                sub_name: "/request/131edb8526804e80b32b387fa2340d35".to_string(),
+            },
+            reply.list_slow_subscribe_raw[1]
+        );
+        assert_eq!(
+            ListSlowSubScribeRaw {
+                client_id: "ed280344fec44aad8a78b00ff1dec99a".to_string(),
+                topic: "/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e".to_string(),
+                time_ms: 543,
+                node_info: "RobustMQ-MQTT@172.22.194.185".to_string(),
+                create_time: 1733898597,
+                sub_name: "/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e".to_string(),
+            },
+            reply.list_slow_subscribe_raw[2]
+        );
+    }
+
+    #[test]
+    fn test_get_sort_data_desc() {
+        let mut reply = set_up_slow_sub_config().unwrap();
+        reply
+            .list_slow_subscribe_raw
+            .sort_by(|a, b| b.time_ms.cmp(&a.time_ms));
+        assert_eq!(
+            ListSlowSubScribeRaw {
+                client_id: "49e10a8d8a494cefa904a00dcf0b30af".to_string(),
+                topic: "/request/131edb8526804e80b32b387fa2340d35".to_string(),
+                time_ms: 13,
+                node_info: "RobustMQ-MQTT@172.22.194.185".to_string(),
+                create_time: 1733898601,
+                sub_name: "/request/131edb8526804e80b32b387fa2340d35".to_string(),
+            },
+            reply.list_slow_subscribe_raw[2]
+        );
+        assert_eq!(
+            ListSlowSubScribeRaw {
+                client_id: "49e10a8d8a494cefa904a00dcf0b30af".to_string(),
+                topic: "/request/131edb8526804e80b32b387fa2340d35".to_string(),
+                time_ms: 273,
+                node_info: "RobustMQ-MQTT@172.22.194.185".to_string(),
+                create_time: 1733898601,
+                sub_name: "/request/131edb8526804e80b32b387fa2340d35".to_string(),
+            },
+            reply.list_slow_subscribe_raw[1]
+        );
+        assert_eq!(
+            ListSlowSubScribeRaw {
+                client_id: "ed280344fec44aad8a78b00ff1dec99a".to_string(),
+                topic: "/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e".to_string(),
+                time_ms: 543,
+                node_info: "RobustMQ-MQTT@172.22.194.185".to_string(),
+                create_time: 1733898597,
+                sub_name: "/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e".to_string(),
+            },
+            reply.list_slow_subscribe_raw[0]
+        );
+    }
+}
diff --git a/src/cmd/src/cli-command/command.rs b/src/cmd/src/cli-command/command.rs
index e411c8390..05b8944c5 100644
--- a/src/cmd/src/cli-command/command.rs
+++ b/src/cmd/src/cli-command/command.rs
@@ -13,7 +13,6 @@
 // limitations under the License.
 
 pub(crate) mod mqtt;
-mod utils;
 
 use clap::{arg, Parser, Subcommand, ValueEnum};
 use cli_command::mqtt::{MqttActionType, MqttBrokerCommand, MqttCliCommandParam};
@@ -27,7 +26,7 @@ use protocol::placement_center::placement_center_openraft::{
     AddLearnerRequest, ChangeMembershipRequest, Node,
 };
 
-use crate::mqtt::admin::{CreateUserArgs, DeleteUserArgs, SlowSubArgs};
+use crate::mqtt::admin::{process_slow_sub_args, CreateUserArgs, DeleteUserArgs, SlowSubArgs};
 
 #[derive(Parser)] // requires `derive` feature
 #[command(name = "robust-ctl")]
@@ -184,7 +183,7 @@ async fn handle_mqtt(args: MqttArgs, cmd: MqttBrokerCommand) {
                     MatchOption::S => 2,
                 },
             }),
-            _ => unreachable!("UnSupport command"),
+            MQTTAction::SlowSub(args) => process_slow_sub_args(args), // _ => unreachable!("UnSupport command"),
         },
     };
     cmd.start(params).await;
diff --git a/src/cmd/src/cli-command/mqtt/admin.rs b/src/cmd/src/cli-command/mqtt/admin.rs
index 488da8c7f..164fbce0b 100644
--- a/src/cmd/src/cli-command/mqtt/admin.rs
+++ b/src/cmd/src/cli-command/mqtt/admin.rs
@@ -16,7 +16,11 @@ use clap::builder::{
     ArgAction, BoolishValueParser, EnumValueParser, NonEmptyStringValueParser, RangedU64ValueParser,
 };
 use clap::Parser;
+use cli_command::mqtt::MqttActionType;
 use common_base::enum_type::common_enum::SortType;
+use protocol::broker_mqtt::broker_mqtt_admin::{
+    EnableSlowSubscribeRequest, ListSlowSubscribeRequest,
+};
 
 // security: user feat
 #[derive(clap::Args, Debug)]
@@ -83,29 +87,103 @@ pub(crate) struct SlowSubArgs {
     client_id: Option<String>,
 }
 
-#[allow(dead_code)]
-impl SlowSubArgs {
-    pub fn get_is_enable(&self) -> Option<bool> {
-        self.is_enable
+pub fn process_slow_sub_args(args: SlowSubArgs) -> MqttActionType {
+    if args.is_enable.is_none() {
+        if args.list.is_none() {
+            panic!("Please provide at least one argument(--is-enable | --list). Use --help for more information.")
+        } else {
+            MqttActionType::ListSlowSubscribe(ListSlowSubscribeRequest {
+                list: args.list.unwrap_or(100),
+                sub_name: args.sub_name.unwrap_or("".to_string()),
+                topic: args.topic.unwrap_or("".to_string()),
+                client_id: args.client_id.unwrap_or("".to_string()),
+                sort: args.sort.unwrap_or(SortType::DESC).to_string(),
+            })
+        }
+    } else {
+        MqttActionType::EnableSlowSubscribe(EnableSlowSubscribeRequest {
+            is_enable: args.is_enable.unwrap(),
+        })
     }
+}
 
-    pub fn get_list(&self) -> Option<u64> {
-        self.list
-    }
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_process_slow_sub_args_function_field_is_enable_not_none() {
+        let args = SlowSubArgs {
+            is_enable: Some(true),
+            list: None,
+            sort: None,
+            topic: None,
+            sub_name: None,
+            client_id: None,
+        };
 
-    pub fn get_sort(&self) -> Option<SortType> {
-        self.sort
+        let action_type = process_slow_sub_args(args);
+        assert_eq!(
+            MqttActionType::EnableSlowSubscribe(EnableSlowSubscribeRequest { is_enable: true }),
+            action_type
+        )
     }
 
-    pub fn get_topic(&self) -> Option<String> {
-        self.topic.clone()
+    #[tokio::test]
+    #[should_panic(
+        expected = "Please provide at least one argument(--is-enable | --list). Use --help for more information."
+    )]
+    async fn test_process_slow_sub_args_function_filed_is_enable_none() {
+        let args = SlowSubArgs {
+            is_enable: None,
+            list: None,
+            sort: None,
+            topic: None,
+            sub_name: None,
+            client_id: None,
+        };
+
+        process_slow_sub_args(args);
     }
 
-    pub fn get_sub_name(&self) -> Option<String> {
-        self.sub_name.clone()
+    #[tokio::test]
+    #[should_panic(
+        expected = "Please provide at least one argument(--is-enable | --list). Use --help for more information."
+    )]
+    async fn test_process_slow_sub_args_function_filed_is_enable_none_1() {
+        let args = SlowSubArgs {
+            is_enable: None,
+            list: None,
+            sort: Some(SortType::ASC),
+            topic: Some("topic_name".to_string()),
+            sub_name: Some("sub_name".to_string()),
+            client_id: Some("client_id".to_string()),
+        };
+
+        process_slow_sub_args(args);
     }
 
-    pub fn get_client_id(&self) -> Option<String> {
-        self.client_id.clone()
+    #[tokio::test]
+    async fn test_process_slow_sub_args_function_filed_is_enable_none_2() {
+        let args = SlowSubArgs {
+            is_enable: None,
+            list: Some(100),
+            sort: Some(SortType::ASC),
+            topic: Some("topic_name".to_string()),
+            sub_name: Some("sub_name".to_string()),
+            client_id: Some("client_id".to_string()),
+        };
+
+        let action_type = process_slow_sub_args(args);
+        assert_eq!(
+            MqttActionType::ListSlowSubscribe(ListSlowSubscribeRequest {
+                list: 100,
+                sub_name: "sub_name".to_string(),
+                topic: "topic_name".to_string(),
+                client_id: "client_id".to_string(),
+                sort: "asc".to_string(),
+            }),
+            action_type
+        )
     }
 }
diff --git a/src/common/base/src/config/broker_mqtt.rs b/src/common/base/src/config/broker_mqtt.rs
index b435c4e41..3ca6f5550 100644
--- a/src/common/base/src/config/broker_mqtt.rs
+++ b/src/common/base/src/config/broker_mqtt.rs
@@ -207,11 +207,15 @@ mod tests {
         assert_eq!(config.storage.journal_addr, "".to_string());
         assert_eq!(config.storage.mysql_addr, "".to_string());
 
+        assert_eq!(
+            config.log.log_config,
+            "./config/log-config/mqtt-log4rs.yaml"
+        );
+
         assert_eq!(
             config.log.log_path,
             "./robust-data/mqtt-broker/logs".to_string()
         );
-        assert_eq!(config.log.log_config, "./config/log4rs.yaml");
 
         assert_eq!(config.auth.storage_type, "placement".to_string());
         assert_eq!(config.auth.journal_addr, "".to_string());
@@ -262,7 +266,10 @@ mod tests {
             config.log.log_path,
             "./robust-data/mqtt-broker/logs".to_string()
         );
-        assert_eq!(config.log.log_config, "./config/log4rs.yaml");
+        assert_eq!(
+            config.log.log_config,
+            "./config/log-config/mqtt-log4rs.yaml"
+        );
 
         assert_eq!(config.auth.storage_type, "placement".to_string());
         assert_eq!(config.auth.journal_addr, "".to_string());
diff --git a/src/common/base/src/utils/file_utils.rs b/src/common/base/src/utils/file_utils.rs
new file mode 100644
index 000000000..269b7536b
--- /dev/null
+++ b/src/common/base/src/utils/file_utils.rs
@@ -0,0 +1,55 @@
+// Copyright 2023 RobustMQ Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use std::fs::read_dir;
+use std::io::ErrorKind;
+use std::path::PathBuf;
+use std::{env, io};
+
+/// Get the project root (relative to closest Cargo.lock file)
+pub fn get_project_root() -> io::Result<PathBuf> {
+    let path = env::current_dir()?;
+    let path_ancestors = path.as_path().ancestors();
+
+    for p in path_ancestors {
+        let has_cargo = read_dir(p)?.any(|p| p.unwrap().file_name() == "Cargo.lock");
+        if has_cargo {
+            return Ok(PathBuf::from(p));
+        }
+    }
+
+    Err(io::Error::new(
+        ErrorKind::NotFound,
+        "Ran out of places to find Cargo.toml",
+    ))
+}
+
+#[cfg(test)]
+mod tests {
+    use std::fs::read_to_string;
+
+    use crate::utils::file_utils::get_project_root;
+
+    #[test]
+    fn it_should_find_our_project_root() {
+        let crate_name = "name = \"robustmq\"";
+
+        let project_root = get_project_root().expect("There is no project root");
+
+        let toml_path = project_root.to_str().unwrap().to_owned() + "/Cargo.toml";
+        let toml_string = read_to_string(toml_path).unwrap();
+
+        assert!(toml_string.contains(crate_name));
+    }
+}
diff --git a/src/common/base/src/utils/mod.rs b/src/common/base/src/utils/mod.rs
index d549d9cdd..c7028cc61 100644
--- a/src/common/base/src/utils/mod.rs
+++ b/src/common/base/src/utils/mod.rs
@@ -12,5 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+pub mod file_utils;
 pub mod topic_util;
 pub mod vec_util;
diff --git a/src/common/metadata-struct/src/mqtt/cluster.rs b/src/common/metadata-struct/src/mqtt/cluster.rs
index f0226583d..39774420a 100644
--- a/src/common/metadata-struct/src/mqtt/cluster.rs
+++ b/src/common/metadata-struct/src/mqtt/cluster.rs
@@ -70,7 +70,7 @@ pub struct MqttClusterDynamicConfigFeature {
 #[derive(Serialize, Deserialize, Default, Clone)]
 pub struct MqttClusterDynamicSlowSub {
     pub enable: bool,
-    pub whole_ms: u128,
+    pub whole_ms: u64,
     pub internal_ms: u32,
     pub response_ms: u32,
 }
diff --git a/src/mqtt-broker/Cargo.toml b/src/mqtt-broker/Cargo.toml
index 182457332..589dae631 100644
--- a/src/mqtt-broker/Cargo.toml
+++ b/src/mqtt-broker/Cargo.toml
@@ -50,3 +50,4 @@ log.workspace = true
 ipnet.workspace = true
 os_info.workspace = true
 bincode.workspace = true
+grep.workspace = true
diff --git a/src/mqtt-broker/src/handler/cache.rs b/src/mqtt-broker/src/handler/cache.rs
index 5192ff84e..3e34fd4cf 100644
--- a/src/mqtt-broker/src/handler/cache.rs
+++ b/src/mqtt-broker/src/handler/cache.rs
@@ -257,13 +257,6 @@ impl CacheManager {
         }
     }
 
-    pub fn get_cluster_info(&self) -> MqttClusterDynamicConfig {
-        if let Some(cluster) = self.cluster_info.get(&self.cluster_name) {
-            return cluster.clone();
-        }
-        MqttClusterDynamicConfig::new()
-    }
-
     pub fn add_user(&self, user: MqttUser) {
         self.user_info.insert(user.username.clone(), user);
     }
diff --git a/src/mqtt-broker/src/handler/cluster_config.rs b/src/mqtt-broker/src/handler/cluster_config.rs
index e76e4709a..76043e23c 100644
--- a/src/mqtt-broker/src/handler/cluster_config.rs
+++ b/src/mqtt-broker/src/handler/cluster_config.rs
@@ -12,18 +12,28 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use common_base::error::common::CommonError;
-use metadata_struct::mqtt::cluster::MqttClusterDynamicConfig;
+use metadata_struct::mqtt::cluster::{MqttClusterDynamicConfig, MqttClusterDynamicSlowSub};
 
 use crate::handler::cache::CacheManager;
+use crate::handler::error::MqttBrokerError;
 use crate::storage::cluster::ClusterStorage;
 
+/// This section primarily implements cache management for cluster-related configuration operations.
+/// Through this implementation, we can retrieve configuration information within the cluster
+/// and set corresponding cluster configuration attributes.
 impl CacheManager {
     pub(super) fn set_cluster_info(&self, cluster: MqttClusterDynamicConfig) {
         self.cluster_info.insert(self.cluster_name.clone(), cluster);
     }
 
-    pub async fn enable_slow_sub(&self, is_enable: bool) -> Result<(), CommonError> {
+    pub fn get_cluster_info(&self) -> MqttClusterDynamicConfig {
+        if let Some(cluster) = self.cluster_info.get(&self.cluster_name) {
+            return cluster.clone();
+        }
+        MqttClusterDynamicConfig::new()
+    }
+
+    pub async fn enable_slow_sub(&self, is_enable: bool) -> Result<(), MqttBrokerError> {
         // save in cache
         let mut dynamic_config = self.get_cluster_info();
         dynamic_config.slow.enable = is_enable;
@@ -37,4 +47,8 @@ impl CacheManager {
             .await?;
         Ok(())
     }
+
+    pub fn get_slow_sub_config(&self) -> MqttClusterDynamicSlowSub {
+        self.get_cluster_info().slow
+    }
 }
diff --git a/src/mqtt-broker/src/handler/error.rs b/src/mqtt-broker/src/handler/error.rs
index 4c0a30942..3eb1006bf 100644
--- a/src/mqtt-broker/src/handler/error.rs
+++ b/src/mqtt-broker/src/handler/error.rs
@@ -32,6 +32,9 @@ pub enum MqttBrokerError {
     #[error("{0}")]
     SerdeJsonError(#[from] serde_json::Error),
 
+    #[error("{0}")]
+    GrepError(#[from] grep::regex::Error),
+
     #[error("{0}")]
     FromMysqlError(#[from] mysql::Error),
 
diff --git a/src/mqtt-broker/src/lib.rs b/src/mqtt-broker/src/lib.rs
index 03191465a..6fe1d234a 100644
--- a/src/mqtt-broker/src/lib.rs
+++ b/src/mqtt-broker/src/lib.rs
@@ -54,7 +54,7 @@ lazy_static! {
 }
 
 pub mod handler;
-mod observability;
+pub mod observability;
 pub mod security;
 mod server;
 pub mod storage;
diff --git a/src/mqtt-broker/src/observability/slow/request.rs b/src/mqtt-broker/src/observability/slow/request.rs
index ad9478749..8a582a11c 100644
--- a/src/mqtt-broker/src/observability/slow/request.rs
+++ b/src/mqtt-broker/src/observability/slow/request.rs
@@ -37,7 +37,7 @@ pub fn try_record_total_request_ms(cache_manager: Arc<CacheManager>, package: Re
 
     let whole = cluster_config.slow.whole_ms;
     let time_ms = now_mills() - package.receive_ms;
-    if time_ms < whole {
+    if time_ms < whole.into() {
         return;
     }
 
diff --git a/src/mqtt-broker/src/observability/slow/sub.rs b/src/mqtt-broker/src/observability/slow/sub.rs
index 47f38b226..9acb0a252 100644
--- a/src/mqtt-broker/src/observability/slow/sub.rs
+++ b/src/mqtt-broker/src/observability/slow/sub.rs
@@ -12,23 +12,33 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use std::collections::VecDeque;
+use std::fs::File;
+use std::path::PathBuf;
+
 use common_base::tools::{get_local_ip, now_second};
+use grep::matcher::Matcher;
+use grep::regex::RegexMatcher;
+use grep::searcher::sinks::UTF8;
+use grep::searcher::Searcher;
+use log::info;
+use protocol::broker_mqtt::broker_mqtt_admin::ListSlowSubscribeRequest;
 use serde::{Deserialize, Serialize};
 
 use crate::handler::error::MqttBrokerError;
 
-#[derive(Serialize, Deserialize, Default, Clone)]
+#[derive(Debug, PartialEq, Serialize, Deserialize, Default, Clone)]
 pub struct SlowSubData {
-    sub_name: String,
-    client_id: String,
-    topic: String,
-    time_ms: u128,
-    node_info: String,
-    create_time: u64,
+    pub(crate) sub_name: String,
+    pub(crate) client_id: String,
+    pub(crate) topic: String,
+    pub(crate) time_ms: u64,
+    pub(crate) node_info: String,
+    pub(crate) create_time: u64,
 }
 
 impl SlowSubData {
-    pub fn build(sub_name: String, client_id: String, topic_name: String, time_ms: u128) -> Self {
+    pub fn build(sub_name: String, client_id: String, topic_name: String, time_ms: u64) -> Self {
         let ip = get_local_ip();
         let node_info = format!("RobustMQ-MQTT@{}", ip);
         SlowSubData {
@@ -42,7 +52,116 @@ impl SlowSubData {
     }
 }
 
-pub fn record_slow_sub_data(slow_data: SlowSubData) -> Result<(), MqttBrokerError> {
-    let _ = serde_json::to_string(&slow_data)?;
+pub fn record_slow_sub_data(slow_data: SlowSubData, whole_ms: u64) -> Result<(), MqttBrokerError> {
+    let data = serde_json::to_string(&slow_data)?;
+
+    if slow_data.time_ms > whole_ms {
+        info!("{}", data);
+    }
+
     Ok(())
 }
+
+pub fn connect_regex_pattern(sub_name: String, client_id: String, topic: String) -> String {
+    let mut pattern: String = String::new();
+    pattern += "\\{";
+    if !sub_name.is_empty() {
+        pattern += ".*";
+        pattern += sub_name.as_str();
+    }
+    if !client_id.is_empty() {
+        pattern += ".*";
+        pattern += client_id.as_str();
+    }
+    if !topic.is_empty() {
+        pattern += ".*";
+        pattern += topic.as_str();
+    }
+    pattern += ".*";
+    pattern += "\\}";
+    pattern
+}
+
+pub fn read_slow_sub_record(
+    search_options: ListSlowSubscribeRequest,
+    path: PathBuf,
+) -> Result<VecDeque<String>, MqttBrokerError> {
+    let regex_pattern = connect_regex_pattern(
+        search_options.sub_name,
+        search_options.client_id,
+        search_options.topic,
+    );
+    let file = File::open(path);
+    let mut matches_queue: VecDeque<String> = VecDeque::new();
+    if file.is_ok() {
+        let matcher = RegexMatcher::new(regex_pattern.as_str())?;
+        Searcher::new().search_file(
+            &matcher,
+            &file.unwrap(),
+            UTF8(|_lnum, line| {
+                let match_byte = matcher.find(line.as_bytes())?.unwrap();
+                if matches_queue.len() == search_options.list as usize {
+                    matches_queue.pop_front();
+                    matches_queue.push_back(line[match_byte].to_string())
+                } else {
+                    matches_queue.push_back(line[match_byte].to_string());
+                }
+                Ok(true)
+            }),
+        )?;
+    }
+
+    Ok(matches_queue)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_regex_pattern_param_is_empty() {
+        let sub_name = "".to_string();
+        let client_id = "".to_string();
+        let topic_id = "".to_string();
+
+        let regex_pattern = connect_regex_pattern(sub_name, client_id, topic_id);
+
+        assert_eq!(regex_pattern, "\\{.*\\}");
+    }
+
+    #[test]
+    fn test_regex_pattern_param_is_one() {
+        let sub_name = "sub_name_test".to_string();
+        let client_id = "".to_string();
+        let topic_id = "".to_string();
+
+        let regex_pattern = connect_regex_pattern(sub_name, client_id, topic_id);
+
+        assert_eq!(regex_pattern, "\\{.*sub_name_test.*\\}");
+    }
+
+    #[test]
+    fn test_regex_pattern_param_is_two() {
+        let sub_name = "sub_name_test".to_string();
+        let client_id = "client_id_test".to_string();
+        let topic_id = "".to_string();
+
+        let regex_pattern = connect_regex_pattern(sub_name, client_id, topic_id);
+
+        assert_eq!(regex_pattern, "\\{.*sub_name_test.*client_id_test.*\\}");
+    }
+
+    #[test]
+    fn test_regex_pattern_param_is_three() {
+        let sub_name = "sub_name_test".to_string();
+        let client_id = "client_id_test".to_string();
+        let topic_id = "topic_id_test".to_string();
+
+        let regex_pattern = connect_regex_pattern(sub_name, client_id, topic_id);
+
+        assert_eq!(
+            regex_pattern,
+            "\\{.*sub_name_test.*client_id_test.*topic_id_test.*\\}"
+        );
+    }
+}
diff --git a/src/mqtt-broker/src/server/grpc/admin/services.rs b/src/mqtt-broker/src/server/grpc/admin/services.rs
index 66758598a..d184801c2 100644
--- a/src/mqtt-broker/src/server/grpc/admin/services.rs
+++ b/src/mqtt-broker/src/server/grpc/admin/services.rs
@@ -16,6 +16,7 @@ use std::sync::Arc;
 
 use common_base::config::broker_mqtt::broker_mqtt_conf;
 use common_base::tools::serialize_value;
+use common_base::utils::file_utils::get_project_root;
 use grpc_clients::pool::ClientPool;
 use metadata_struct::acl::mqtt_acl::MqttAcl;
 use metadata_struct::acl::mqtt_blacklist::{MqttAclBlackList, MqttAclBlackListType};
@@ -34,6 +35,7 @@ use protocol::broker_mqtt::broker_mqtt_admin::{
 use tonic::{Request, Response, Status};
 
 use crate::handler::cache::CacheManager;
+use crate::observability::slow::sub::{read_slow_sub_record, SlowSubData};
 use crate::security::AuthDriver;
 use crate::server::connection_manager::ConnectionManager;
 use crate::storage::cluster::ClusterStorage;
@@ -81,7 +83,7 @@ impl MqttBrokerAdminService for GrpcAdminServices {
             }
         }
         reply.nodes = broker_node_list;
-        return Ok(Response::new(reply));
+        Ok(Response::new(reply))
     }
 
     // --- user ---
@@ -98,12 +100,8 @@ impl MqttBrokerAdminService for GrpcAdminServices {
 
         let auth_driver = AuthDriver::new(self.cache_manager.clone(), self.client_pool.clone());
         match auth_driver.save_user(mqtt_user).await {
-            Ok(_) => {
-                return Ok(Response::new(CreateUserReply::default()));
-            }
-            Err(e) => {
-                return Err(Status::cancelled(e.to_string()));
-            }
+            Ok(_) => Ok(Response::new(CreateUserReply::default())),
+            Err(e) => Err(Status::cancelled(e.to_string())),
         }
     }
 
@@ -115,10 +113,8 @@ impl MqttBrokerAdminService for GrpcAdminServices {
 
         let auth_driver = AuthDriver::new(self.cache_manager.clone(), self.client_pool.clone());
         match auth_driver.delete_user(req.username).await {
-            Ok(_) => return Ok(Response::new(DeleteUserReply::default())),
-            Err(e) => {
-                return Err(Status::cancelled(e.to_string()));
-            }
+            Ok(_) => Ok(Response::new(DeleteUserReply::default())),
+            Err(e) => Err(Status::cancelled(e.to_string())),
         }
     }
 
@@ -135,11 +131,9 @@ impl MqttBrokerAdminService for GrpcAdminServices {
                     users.push(ele.1.encode());
                 }
                 reply.users = users;
-                return Ok(Response::new(reply));
-            }
-            Err(e) => {
-                return Err(Status::cancelled(e.to_string()));
+                Ok(Response::new(reply))
             }
+            Err(e) => Err(Status::cancelled(e.to_string())),
         }
     }
 
@@ -153,6 +147,7 @@ impl MqttBrokerAdminService for GrpcAdminServices {
         match auth_driver.read_all_acl().await {
             Ok(data) => {
                 let mut acls_list = Vec::new();
+                // todo finish get_items
                 for ele in data {
                     match ele.encode() {
                         Ok(acl) => acls_list.push(acl),
@@ -160,11 +155,9 @@ impl MqttBrokerAdminService for GrpcAdminServices {
                     }
                 }
                 reply.acls = acls_list;
-                return Ok(Response::new(reply));
-            }
-            Err(e) => {
-                return Err(Status::cancelled(e.to_string()));
+                Ok(Response::new(reply))
             }
+            Err(e) => Err(Status::cancelled(e.to_string())),
         }
     }
 
@@ -178,10 +171,8 @@ impl MqttBrokerAdminService for GrpcAdminServices {
 
         let auth_driver = AuthDriver::new(self.cache_manager.clone(), self.client_pool.clone());
         match auth_driver.save_acl(mqtt_acl).await {
-            Ok(_) => return Ok(Response::new(CreateAclReply::default())),
-            Err(e) => {
-                return Err(Status::cancelled(e.to_string()));
-            }
+            Ok(_) => Ok(Response::new(CreateAclReply::default())),
+            Err(e) => Err(Status::cancelled(e.to_string())),
         }
     }
 
@@ -194,10 +185,8 @@ impl MqttBrokerAdminService for GrpcAdminServices {
 
         let auth_driver = AuthDriver::new(self.cache_manager.clone(), self.client_pool.clone());
         match auth_driver.delete_acl(mqtt_acl).await {
-            Ok(_) => return Ok(Response::new(DeleteAclReply::default())),
-            Err(e) => {
-                return Err(Status::cancelled(e.to_string()));
-            }
+            Ok(_) => Ok(Response::new(DeleteAclReply::default())),
+            Err(e) => Err(Status::cancelled(e.to_string())),
         }
     }
 
@@ -217,11 +206,9 @@ impl MqttBrokerAdminService for GrpcAdminServices {
                     }
                 }
                 reply.blacklists = blacklists;
-                return Ok(Response::new(reply));
-            }
-            Err(e) => {
-                return Err(Status::cancelled(e.to_string()));
+                Ok(Response::new(reply))
             }
+            Err(e) => Err(Status::cancelled(e.to_string())),
         }
     }
 
@@ -247,10 +234,8 @@ impl MqttBrokerAdminService for GrpcAdminServices {
 
         let auth_driver = AuthDriver::new(self.cache_manager.clone(), self.client_pool.clone());
         match auth_driver.delete_blacklist(mqtt_blacklist).await {
-            Ok(_) => return Ok(Response::new(DeleteBlacklistReply::default())),
-            Err(e) => {
-                return Err(Status::cancelled(e.to_string()));
-            }
+            Ok(_) => Ok(Response::new(DeleteBlacklistReply::default())),
+            Err(e) => Err(Status::cancelled(e.to_string())),
         }
     }
 
@@ -263,10 +248,8 @@ impl MqttBrokerAdminService for GrpcAdminServices {
 
         let auth_driver = AuthDriver::new(self.cache_manager.clone(), self.client_pool.clone());
         match auth_driver.save_blacklist(mqtt_blacklist).await {
-            Ok(_) => return Ok(Response::new(CreateBlacklistReply::default())),
-            Err(e) => {
-                return Err(Status::cancelled(e.to_string()));
-            }
+            Ok(_) => Ok(Response::new(CreateBlacklistReply::default())),
+            Err(e) => Err(Status::cancelled(e.to_string())),
         }
     }
 
@@ -319,8 +302,32 @@ impl MqttBrokerAdminService for GrpcAdminServices {
         &self,
         request: Request<ListSlowSubscribeRequest>,
     ) -> Result<Response<ListSlowSubscribeReply>, Status> {
-        let _list_slow_subscribe_request = request.into_inner();
-        let list_slow_subscribe_raw: Vec<ListSlowSubScribeRaw> = Vec::new();
+        let list_slow_subscribe_request = request.into_inner();
+        let mut list_slow_subscribe_raw: Vec<ListSlowSubScribeRaw> = Vec::new();
+        let mqtt_config = broker_mqtt_conf();
+        if self.cache_manager.get_slow_sub_config().enable {
+            let path = mqtt_config.log.log_path.clone();
+            let path_buf = get_project_root()?.join(path.replace("./", "") + "/slow_sub.log");
+            let deque = read_slow_sub_record(list_slow_subscribe_request, path_buf)?;
+            for slow_sub_data in deque {
+                match serde_json::from_str::<SlowSubData>(slow_sub_data.as_str()) {
+                    Ok(data) => {
+                        let raw = ListSlowSubScribeRaw {
+                            client_id: data.client_id,
+                            topic: data.topic,
+                            time_ms: data.time_ms,
+                            node_info: data.node_info,
+                            create_time: data.create_time,
+                            sub_name: data.sub_name,
+                        };
+                        list_slow_subscribe_raw.push(raw);
+                    }
+                    Err(e) => {
+                        return Err(Status::cancelled(e.to_string()));
+                    }
+                }
+            }
+        }
         Ok(Response::new(ListSlowSubscribeReply {
             list_slow_subscribe_raw,
         }))
@@ -367,6 +374,6 @@ impl MqttBrokerAdminService for GrpcAdminServices {
             topics: topic_query_result,
         };
 
-        return Ok(Response::new(reply));
+        Ok(Response::new(reply))
     }
 }
diff --git a/src/mqtt-broker/src/server/tcp/server.rs b/src/mqtt-broker/src/server/tcp/server.rs
index d936a7221..ebe204ec4 100644
--- a/src/mqtt-broker/src/server/tcp/server.rs
+++ b/src/mqtt-broker/src/server/tcp/server.rs
@@ -34,7 +34,7 @@ use crate::server::tcp::tls_server::acceptor_tls_process;
 use crate::subscribe::subscribe_manager::SubscribeManager;
 
 pub async fn start_tcp_server<S>(
-    sucscribe_manager: Arc<SubscribeManager>,
+    subscribe_manager: Arc<SubscribeManager>,
     cache_manager: Arc<CacheManager>,
     connection_manager: Arc<ConnectionManager>,
     message_storage_adapter: Arc<S>,
@@ -48,7 +48,7 @@ pub async fn start_tcp_server<S>(
     let command = Command::new(
         cache_manager.clone(),
         message_storage_adapter.clone(),
-        sucscribe_manager.clone(),
+        subscribe_manager.clone(),
         client_pool.clone(),
         connection_manager.clone(),
         auth_driver.clone(),
@@ -65,7 +65,7 @@ pub async fn start_tcp_server<S>(
         proc_config,
         stop_sx.clone(),
         connection_manager.clone(),
-        sucscribe_manager.clone(),
+        subscribe_manager.clone(),
         cache_manager.clone(),
         client_pool.clone(),
     );
@@ -76,7 +76,7 @@ pub async fn start_tcp_server<S>(
         proc_config,
         stop_sx.clone(),
         connection_manager,
-        sucscribe_manager.clone(),
+        subscribe_manager.clone(),
         cache_manager,
         client_pool,
     );
diff --git a/src/mqtt-broker/src/subscribe/sub_common.rs b/src/mqtt-broker/src/subscribe/sub_common.rs
index eeff489b9..43c89685a 100644
--- a/src/mqtt-broker/src/subscribe/sub_common.rs
+++ b/src/mqtt-broker/src/subscribe/sub_common.rs
@@ -205,6 +205,7 @@ pub async fn publish_message_to_client(
     resp: ResponsePackage,
     sub_pub_param: &SubPublishParam,
     connection_manager: &Arc<ConnectionManager>,
+    metadata_cache: &Arc<CacheManager>,
 ) -> Result<(), MqttBrokerError> {
     if let Some(protocol) = connection_manager.get_connect_protocol(resp.connection_id) {
         let response: MqttPacketWrapper = MqttPacketWrapper {
@@ -229,16 +230,15 @@ pub async fn publish_message_to_client(
                 .write_tcp_frame(resp.connection_id, response)
                 .await?
         }
-
         // record slow sub data
-        if sub_pub_param.create_time > 0 {
+        if metadata_cache.get_slow_sub_config().enable && sub_pub_param.create_time > 0 {
             let slow_data = SlowSubData::build(
                 sub_pub_param.subscribe.sub_path.clone(),
                 sub_pub_param.subscribe.client_id.clone(),
                 sub_pub_param.subscribe.topic_name.clone(),
-                now_mills() - sub_pub_param.create_time,
+                (now_mills() - sub_pub_param.create_time) as u64,
             );
-            record_slow_sub_data(slow_data)?;
+            record_slow_sub_data(slow_data, metadata_cache.get_slow_sub_config().whole_ms)?;
         }
     }
 
@@ -304,6 +304,7 @@ pub async fn qos2_send_publish(
                 resp.clone(),
                 sub_pub_param,
                 connection_manager,
+                metadata_cache
             ) =>{
                 match val{
                     Ok(_) => {
@@ -363,6 +364,7 @@ pub async fn qos2_send_pubrel(
                 pubrel_resp.clone(),
                 sub_pub_param,
                 connection_manager,
+                metadata_cache
             ) =>{
                 match val{
                     Ok(_) => {
@@ -458,7 +460,14 @@ pub async fn publish_message_qos0(
     };
 
     // 2. publish to mqtt client
-    match publish_message_to_client(resp.clone(), sub_pub_param, connection_manager).await {
+    match publish_message_to_client(
+        resp.clone(),
+        sub_pub_param,
+        connection_manager,
+        metadata_cache,
+    )
+    .await
+    {
         Ok(_) => {}
         Err(e) => {
             error!(
diff --git a/src/mqtt-broker/src/subscribe/sub_exclusive.rs b/src/mqtt-broker/src/subscribe/sub_exclusive.rs
index f3fd949e9..06002229d 100644
--- a/src/mqtt-broker/src/subscribe/sub_exclusive.rs
+++ b/src/mqtt-broker/src/subscribe/sub_exclusive.rs
@@ -424,7 +424,9 @@ pub async fn exclusive_publish_message_qos1(
             }
         };
 
-        match publish_message_to_client(resp, sub_pub_param, connection_manager).await {
+        match publish_message_to_client(resp, sub_pub_param, connection_manager, metadata_cache)
+            .await
+        {
             Ok(_) => {
                 if let Some(data) = wait_packet_ack(wait_puback_sx).await {
                     if data.ack_type == QosAckPackageType::PubAck && data.pkid == sub_pub_param.pkid
diff --git a/src/mqtt-broker/src/subscribe/sub_share_follower.rs b/src/mqtt-broker/src/subscribe/sub_share_follower.rs
index 094e3b2fc..1e4569920 100644
--- a/src/mqtt-broker/src/subscribe/sub_share_follower.rs
+++ b/src/mqtt-broker/src/subscribe/sub_share_follower.rs
@@ -567,7 +567,9 @@ async fn resub_publish_message_qos1(
         };
 
         // 2. publish to mqtt client
-        match publish_message_to_client(resp, sub_pub_param, connection_manager).await {
+        match publish_message_to_client(resp, sub_pub_param, connection_manager, metadata_cache)
+            .await
+        {
             Ok(_) => {
                 // 3. wait mqtt client puback
                 if let Some(data) = wait_packet_ack(wait_puback_sx).await {
diff --git a/src/mqtt-broker/src/subscribe/sub_share_leader.rs b/src/mqtt-broker/src/subscribe/sub_share_leader.rs
index ce87df48b..8e78b040e 100644
--- a/src/mqtt-broker/src/subscribe/sub_share_leader.rs
+++ b/src/mqtt-broker/src/subscribe/sub_share_leader.rs
@@ -514,7 +514,14 @@ async fn share_leader_publish_message_qos1(
         }
     };
 
-    match publish_message_to_client(resp.clone(), sub_pub_param, connection_manager).await {
+    match publish_message_to_client(
+        resp.clone(),
+        sub_pub_param,
+        connection_manager,
+        metadata_cache,
+    )
+    .await
+    {
         Ok(_) => {
             if let Some(data) = wait_packet_ack(wait_puback_sx).await {
                 if data.ack_type == QosAckPackageType::PubAck && data.pkid == sub_pub_param.pkid {
diff --git a/src/protocol/src/broker_mqtt/proto/admin.proto b/src/protocol/src/broker_mqtt/proto/admin.proto
index 507c3eb66..de9a8515a 100644
--- a/src/protocol/src/broker_mqtt/proto/admin.proto
+++ b/src/protocol/src/broker_mqtt/proto/admin.proto
@@ -157,6 +157,28 @@ message EnableSlowSubScribeReply {
     bool is_enable = 1;
 }
 
+message ListSlowSubscribeRequest {
+    uint64 list = 1;
+    string sub_name = 3;
+    string topic = 2;
+    string client_id = 4;
+    string sort = 5;
+}
+
+message ListSlowSubscribeReply {
+    repeated ListSlowSubScribeRaw list_slow_subscribe_raw = 1;
+}
+
+message ListSlowSubScribeRaw {
+    string client_id = 1;
+    string topic = 2;
+    uint64 time_ms = 3;
+    string node_info = 4;
+    uint64 create_time = 5;
+    string sub_name = 6;
+}
+
+
 // --------- mqtt topic --------
 enum MatchOption {
     E = 0;
@@ -177,19 +199,3 @@ message MqttTopic {
     string topic_name = 3;
     bool is_contain_retain_message = 4;
 }
-
-message ListSlowSubscribeRequest {
-
-}
-
-message ListSlowSubscribeReply {
-    repeated ListSlowSubScribeRaw list_slow_subscribe_raw = 1;
-}
-
-message ListSlowSubScribeRaw {
-    string client_id = 1;
-    string topic = 2;
-    uint32 time_ms = 3;
-    string node_info = 4;
-    uint64 create_time = 5;
-}


Test patch:
diff --git a/tests/tests/mqtt_client/slow_sub_test.rs b/tests/tests/mqtt_client/slow_sub_test.rs
index cb05070ad..2fb59ef2c 100644
--- a/tests/tests/mqtt_client/slow_sub_test.rs
+++ b/tests/tests/mqtt_client/slow_sub_test.rs
@@ -49,7 +49,13 @@ mod tests {
     async fn test_list_slow_subscribe() {
         let client_pool = Arc::new(ClientPool::new(3));
         let grpc_addr = vec![broker_grpc_addr()];
-        let request = ListSlowSubscribeRequest {};
+        let request = ListSlowSubscribeRequest {
+            list: 100,
+            sub_name: "".to_string(),
+            topic: "".to_string(),
+            client_id: "".to_string(),
+            sort: "asc".to_string(),
+        };
 
         match mqtt_broker_list_slow_subscribe(&client_pool, &grpc_addr, request).await {
             Ok(data) => {
diff --git a/tests/tests/mqtt_server/example/slow_sub.log b/tests/tests/mqtt_server/example/slow_sub.log
new file mode 100644
index 000000000..102fe808f
--- /dev/null
+++ b/tests/tests/mqtt_server/example/slow_sub.log
@@ -0,0 +1,10 @@
+{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","time_ms":543,"node_info":"RobustMQ-MQTT@172.22.194.185","create_time":1733898597}
+{"sub_name":"/request/131edb8526804e80b32b387fa2340d35","client_id":"49e10a8d8a494cefa904a00dcf0b30af","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":273,"node_info":"RobustMQ-MQTT@172.22.194.185","create_time":1733898601}
+{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"","topic":"","time_ms":543,"node_info":"RobustMQ-MQTT@172.22.194.185","create_time":1733898597}
+{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"","topic":"","time_ms":543,"node_info":"RobustMQ-MQTT@172.22.194.185","create_time":1733898597}
+{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":543,"node_info":"RobustMQ-MQTT@172.22.194.185","create_time":1733898597}
+{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","time_ms":13,"node_info":"RobustMQ-MQTT@182.22.194.185","create_time":1733488597}
+{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","time_ms":3,"node_info":"RobustMQ-MQTT@172.22.111.185","create_time":1733398597}
+{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":143,"node_info":"RobustMQ-MQTT@172.22.111.85","create_time":1733864597}
+{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"","topic":"","time_ms":543,"node_info":"RobustMQ-MQTT@172.22.194.185","create_time":1733898597}
+{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":543,"node_info":"RobustMQ-MQTT@192.32.194.185","create_time":1733858597}
diff --git a/tests/tests/mqtt_server/mod.rs b/tests/tests/mqtt_server/mod.rs
index c361b2219..d0ef1956e 100644
--- a/tests/tests/mqtt_server/mod.rs
+++ b/tests/tests/mqtt_server/mod.rs
@@ -13,6 +13,7 @@
 // limitations under the License.
 
 pub mod cluster_storage_test;
+mod observability;
 pub mod session_storage_test;
 pub mod topic_storage_test;
 pub mod user_storage_test;
diff --git a/src/cmd/src/cli-command/utils.rs b/tests/tests/mqtt_server/observability/mod.rs
similarity index 96%
rename from src/cmd/src/cli-command/utils.rs
rename to tests/tests/mqtt_server/observability/mod.rs
index 88041165f..02c2020f0 100644
--- a/src/cmd/src/cli-command/utils.rs
+++ b/tests/tests/mqtt_server/observability/mod.rs
@@ -11,3 +11,5 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
+
+mod slow_sub_test;
diff --git a/tests/tests/mqtt_server/observability/slow_sub_test.rs b/tests/tests/mqtt_server/observability/slow_sub_test.rs
new file mode 100644
index 000000000..66f53f108
--- /dev/null
+++ b/tests/tests/mqtt_server/observability/slow_sub_test.rs
@@ -0,0 +1,209 @@
+// Copyright 2023 RobustMQ Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+#[cfg(test)]
+mod tests {
+    use std::path::PathBuf;
+
+    use mqtt_broker::observability::slow::sub::read_slow_sub_record;
+    use protocol::broker_mqtt::broker_mqtt_admin::ListSlowSubscribeRequest;
+
+    #[test]
+    fn test_read_slow_record_path_is_empty() {
+        let path = "";
+        let path_buf = PathBuf::new().join(path);
+        let record = ListSlowSubscribeRequest {
+            sub_name: "/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e".to_string(),
+            client_id: "".to_string(),
+            topic: "".to_string(),
+            list: 5,
+            sort: "asc".to_string(),
+        };
+        let result = read_slow_sub_record(record, path_buf).unwrap();
+        assert_eq!(0, result.clone().len())
+    }
+
+    #[test]
+    fn test_read_slow_record_param_is_one_1() {
+        let path = "./tests/mqtt_server/example/slow_sub.log";
+        let path_buf = PathBuf::new().join(path);
+        let record = ListSlowSubscribeRequest {
+            sub_name: "/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e".to_string(),
+            client_id: "".to_string(),
+            topic: "".to_string(),
+            list: 5,
+            sort: "asc".to_string(),
+        };
+
+        let result = read_slow_sub_record(record, path_buf).unwrap();
+
+        assert_eq!(5, result.clone().len());
+        assert_eq!(
+            result.clone()[0],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","time_ms":13,"node_info":"RobustMQ-MQTT@182.22.194.185","create_time":1733488597}"#
+        );
+        assert_eq!(
+            result.clone()[1],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","time_ms":3,"node_info":"RobustMQ-MQTT@172.22.111.185","create_time":1733398597}"#
+        );
+        assert_eq!(
+            result.clone()[2],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":143,"node_info":"RobustMQ-MQTT@172.22.111.85","create_time":1733864597}"#
+        );
+        assert_eq!(
+            result.clone()[3],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"","topic":"","time_ms":543,"node_info":"RobustMQ-MQTT@172.22.194.185","create_time":1733898597}"#
+        );
+        assert_eq!(
+            result.clone()[4],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":543,"node_info":"RobustMQ-MQTT@192.32.194.185","create_time":1733858597}"#
+        );
+    }
+
+    #[test]
+    fn test_read_slow_record_param_is_one_2() {
+        let path = "./tests/mqtt_server/example/slow_sub.log";
+        let path_buf = PathBuf::new().join(path);
+        let record = ListSlowSubscribeRequest {
+            sub_name: "".to_string(),
+            client_id: "ed280344fec44aad8a78b00ff1dec99a".to_string(),
+            topic: "".to_string(),
+            list: 5,
+            sort: "asc".to_string(),
+        };
+
+        let result = read_slow_sub_record(record, path_buf).unwrap();
+
+        assert_eq!(5, result.clone().len());
+        assert_eq!(
+            result.clone()[0],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":543,"node_info":"RobustMQ-MQTT@172.22.194.185","create_time":1733898597}"#
+        );
+        assert_eq!(
+            result.clone()[1],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","time_ms":13,"node_info":"RobustMQ-MQTT@182.22.194.185","create_time":1733488597}"#
+        );
+        assert_eq!(
+            result.clone()[2],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","time_ms":3,"node_info":"RobustMQ-MQTT@172.22.111.185","create_time":1733398597}"#
+        );
+        assert_eq!(
+            result.clone()[3],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":143,"node_info":"RobustMQ-MQTT@172.22.111.85","create_time":1733864597}"#
+        );
+        assert_eq!(
+            result.clone()[4],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":543,"node_info":"RobustMQ-MQTT@192.32.194.185","create_time":1733858597}"#
+        );
+    }
+
+    #[test]
+    fn test_read_slow_record_param_is_one_3() {
+        let path = "./tests/mqtt_server/example/slow_sub.log";
+        let path_buf = PathBuf::new().join(path);
+        let record = ListSlowSubscribeRequest {
+            sub_name: "".to_string(),
+            client_id: "".to_string(),
+            topic: "/request/131edb8526804e80b32b387fa2340d35".to_string(),
+            list: 5,
+            sort: "asc".to_string(),
+        };
+
+        let result = read_slow_sub_record(record, path_buf).unwrap();
+
+        assert_eq!(4, result.clone().len());
+        assert_eq!(
+            result.clone()[0],
+            r#"{"sub_name":"/request/131edb8526804e80b32b387fa2340d35","client_id":"49e10a8d8a494cefa904a00dcf0b30af","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":273,"node_info":"RobustMQ-MQTT@172.22.194.185","create_time":1733898601}"#
+        );
+        assert_eq!(
+            result.clone()[1],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":543,"node_info":"RobustMQ-MQTT@172.22.194.185","create_time":1733898597}"#
+        );
+        assert_eq!(
+            result.clone()[2],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":143,"node_info":"RobustMQ-MQTT@172.22.111.85","create_time":1733864597}"#
+        );
+        assert_eq!(
+            result.clone()[3],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":543,"node_info":"RobustMQ-MQTT@192.32.194.185","create_time":1733858597}"#
+        );
+    }
+
+    #[test]
+    fn test_read_slow_record_param_is_two() {
+        let path = "./tests/mqtt_server/example/slow_sub.log";
+        let path_buf = PathBuf::new().join(path);
+        let record = ListSlowSubscribeRequest {
+            sub_name: "/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e".to_string(),
+            client_id: "ed280344fec44aad8a78b00ff1dec99a".to_string(),
+            topic: "".to_string(),
+            list: 5,
+            sort: "asc".to_string(),
+        };
+
+        let result = read_slow_sub_record(record, path_buf).unwrap();
+
+        assert_eq!(5, result.clone().len());
+        assert_eq!(
+            result.clone()[0],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":543,"node_info":"RobustMQ-MQTT@172.22.194.185","create_time":1733898597}"#
+        );
+        assert_eq!(
+            result.clone()[1],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","time_ms":13,"node_info":"RobustMQ-MQTT@182.22.194.185","create_time":1733488597}"#
+        );
+        assert_eq!(
+            result.clone()[2],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","time_ms":3,"node_info":"RobustMQ-MQTT@172.22.111.185","create_time":1733398597}"#
+        );
+        assert_eq!(
+            result.clone()[3],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":143,"node_info":"RobustMQ-MQTT@172.22.111.85","create_time":1733864597}"#
+        );
+        assert_eq!(
+            result.clone()[4],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/request/131edb8526804e80b32b387fa2340d35","time_ms":543,"node_info":"RobustMQ-MQTT@192.32.194.185","create_time":1733858597}"#
+        );
+    }
+
+    #[test]
+    fn test_read_slow_record_param_is_three() {
+        let path = "./tests/mqtt_server/example/slow_sub.log";
+        let path_buf = PathBuf::new().join(path);
+        let record = ListSlowSubscribeRequest {
+            sub_name: "/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e".to_string(),
+            client_id: "ed280344fec44aad8a78b00ff1dec99a".to_string(),
+            topic: "/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e".to_string(),
+            list: 5,
+            sort: "asc".to_string(),
+        };
+
+        let result = read_slow_sub_record(record, path_buf).unwrap();
+
+        assert_eq!(3, result.clone().len());
+        assert_eq!(
+            result.clone()[0],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","time_ms":543,"node_info":"RobustMQ-MQTT@172.22.194.185","create_time":1733898597}"#
+        );
+        assert_eq!(
+            result.clone()[1],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","time_ms":13,"node_info":"RobustMQ-MQTT@182.22.194.185","create_time":1733488597}"#
+        );
+        assert_eq!(
+            result.clone()[2],
+            r#"{"sub_name":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","client_id":"ed280344fec44aad8a78b00ff1dec99a","topic":"/packet_tcp_ssl/7fce56aa49ef4cea90dc4be77d6a775e","time_ms":3,"node_info":"RobustMQ-MQTT@172.22.111.185","create_time":1733398597}"#
+        );
+    }
+}