Repository: robustmq/robustmq (language: rust)

Issue:
[Feat] MQTT offline messages
### Have you checked the documentation for submitting an Issue?

- [x] Yes.

### What problem does the new feature solve?

MQTT offline messages

### What does the feature do?

MQTT offline messages

### Implementation challenges

_No response_

### Are you willing to submit PR?

- [ ] Yes. I would be willing to submit a PR with guidance from the RobustMQ community to improve.
- [ ] No. I cannot submit a PR at this time.


Discussion / hints:

Fix patch:
diff --git a/docs/.design/plan/mqtt-stable-to-do.md b/docs/.design/plan/mqtt-stable-to-do.md
index a22f7e400..e60802df8 100644
--- a/docs/.design/plan/mqtt-stable-to-do.md
+++ b/docs/.design/plan/mqtt-stable-to-do.md
@@ -1,24 +1,25 @@
 ## 功能覆盖度
-1. 离线消息
-2. 订阅逻辑优化
-3. 数据集成
-   1. 文件传输
-   2. Kafka 集成
-   3. 企业及数据集成
-4. 限流
+1. 离线消息【done】
+2. 订阅逻辑优化【done】
+3. MQTT 测试用例失败 case 修复
+4. 自动订阅
+5. 数据集成
+   1. sink local file
+   3. Kafka 集成
+6. 限流
    1. 速率限制
    2. QOS 限制
-5. 可观测体系搭建
-6. ACL/User
-7. Schema: Schema Registry、灵活地消息格式转换：JSON、Avro、Protobuf、Custom codec (HTTP/gRPC)
-8. 消息验证：确保消息的完整性和合法性
-9.  规则引擎
-10. Flow 设计器
-11. 故障排查
-12. Cloud-Native & K8s
-13. 边缘计算
-14. MQTT Over QUIC
-15. 多协议网关
+7. 可观测体系搭建
+8. ACL/User
+9. Schema: Schema Registry、灵活地消息格式转换：JSON、Avro、Protobuf、Custom codec (HTTP/gRPC)
+10. 消息验证：确保消息的完整性和合法性
+11. 规则引擎
+12. Flow 设计器
+13. 故障排查
+14. Cloud-Native & K8s
+15. 边缘计算
+16. MQTT Over QUIC
+17. 多协议网关
 
 ## 测试用例覆盖度
 1. 代码测试覆盖度
diff --git a/src/mqtt-broker/src/handler/cache_update.rs b/src/mqtt-broker/src/handler/cache_update.rs
index 3ecce61ef..12a51a3aa 100644
--- a/src/mqtt-broker/src/handler/cache_update.rs
+++ b/src/mqtt-broker/src/handler/cache_update.rs
@@ -29,7 +29,7 @@ use protocol::broker_mqtt::broker_mqtt_inner::{
 };
 use std::sync::Arc;
 
-use super::{cache::CacheManager, sub_exclusive::try_remove_exclusive_subscribe_by_path};
+use super::{cache::CacheManager, sub_exclusive::remove_exclusive_subscribe_by_path};
 
 pub async fn load_metadata_cache(
     cache_manager: &Arc<CacheManager>,
@@ -181,10 +181,7 @@ pub async fn update_cache_metadata(
                         if let Some(_sub) =
                             subscribe_manager.get_subscribe(&subscribe.client_id, &subscribe.path)
                         {
-                            try_remove_exclusive_subscribe_by_path(
-                                subscribe_manager,
-                                &subscribe.path,
-                            );
+                            remove_exclusive_subscribe_by_path(subscribe_manager, &subscribe.path);
                         }
                         subscribe_manager.remove_subscribe(&subscribe.client_id, &subscribe.path)
                     }
diff --git a/src/mqtt-broker/src/handler/connection.rs b/src/mqtt-broker/src/handler/connection.rs
index 4e2c1b535..64a7399b3 100644
--- a/src/mqtt-broker/src/handler/connection.rs
+++ b/src/mqtt-broker/src/handler/connection.rs
@@ -24,11 +24,8 @@ use protocol::mqtt::common::{Connect, ConnectProperties};
 use super::cache::CacheManager;
 use super::error::MqttBrokerError;
 use super::keep_alive::client_keep_live_time;
-use super::sub_exclusive::try_remove_exclusive_subscribe_by_client_id;
-use super::unsubscribe::stop_push_by_client_id;
 use crate::server::connection_manager::ConnectionManager;
 use crate::storage::session::SessionStorage;
-use crate::subscribe::subscribe_manager::SubscribeManager;
 
 pub const REQUEST_RESPONSE_PREFIX_NAME: &str = "/sys/request_response/";
 
@@ -117,17 +114,11 @@ pub async fn disconnect_connection(
     cache_manager: &Arc<CacheManager>,
     client_pool: &Arc<ClientPool>,
     connection_manager: &Arc<ConnectionManager>,
-    subscribe_manager: &Arc<SubscribeManager>,
 ) -> Result<(), MqttBrokerError> {
-    // remove the exclusive subscribe
-    try_remove_exclusive_subscribe_by_client_id(subscribe_manager, client_id);
-
     // Remove the connection cache
     cache_manager.remove_connection(connect_id);
     // Remove the client id bound connection information
     cache_manager.update_session_connect_id(client_id, None);
-    // Once the connection is dropped, the push thread for the Client ID dimension is paused
-    stop_push_by_client_id(subscribe_manager, client_id);
 
     // Remove the Connect id of the Session in the Placement Center
     let session_storage = SessionStorage::new(client_pool.clone());
diff --git a/src/mqtt-broker/src/handler/error.rs b/src/mqtt-broker/src/handler/error.rs
index 0efd4424f..17d8f3f8a 100644
--- a/src/mqtt-broker/src/handler/error.rs
+++ b/src/mqtt-broker/src/handler/error.rs
@@ -29,6 +29,9 @@ pub enum MqttBrokerError {
     #[error("{0}")]
     FromCommonError(#[from] CommonError),
 
+    #[error("{0}")]
+    TokioBroadcastSendError(#[from] tokio::sync::broadcast::error::SendError<bool>),
+
     #[error("{0}")]
     SerdeJsonError(#[from] serde_json::Error),
 
diff --git a/src/mqtt-broker/src/handler/keep_alive.rs b/src/mqtt-broker/src/handler/keep_alive.rs
index fa6bb3412..5c21bb885 100644
--- a/src/mqtt-broker/src/handler/keep_alive.rs
+++ b/src/mqtt-broker/src/handler/keep_alive.rs
@@ -32,20 +32,17 @@ use super::cache::{CacheManager, ConnectionLiveTime};
 use super::connection::disconnect_connection;
 use super::response::response_packet_mqtt_distinct_by_reason;
 use crate::server::connection_manager::ConnectionManager;
-use crate::subscribe::subscribe_manager::SubscribeManager;
 
 pub struct ClientKeepAlive {
     cache_manager: Arc<CacheManager>,
     stop_send: broadcast::Sender<bool>,
     client_pool: Arc<ClientPool>,
     connection_manager: Arc<ConnectionManager>,
-    subscribe_manager: Arc<SubscribeManager>,
 }
 
 impl ClientKeepAlive {
     pub fn new(
         client_pool: Arc<ClientPool>,
-        subscribe_manager: Arc<SubscribeManager>,
         connection_manager: Arc<ConnectionManager>,
         cache_manager: Arc<CacheManager>,
         stop_send: broadcast::Sender<bool>,
@@ -55,7 +52,6 @@ impl ClientKeepAlive {
             connection_manager,
             cache_manager,
             stop_send,
-            subscribe_manager,
         }
     }
 
@@ -107,7 +103,6 @@ impl ClientKeepAlive {
                                     &self.cache_manager,
                                     &self.client_pool,
                                     &self.connection_manager,
-                                    &self.subscribe_manager,
                                 )
                                 .await
                                 {
@@ -154,7 +149,6 @@ impl ClientKeepAlive {
                                     &self.cache_manager,
                                     &self.client_pool,
                                     &self.connection_manager,
-                                    &self.subscribe_manager,
                                 )
                                 .await
                                 {
@@ -251,7 +245,6 @@ mod test {
     use crate::handler::cache::CacheManager;
     use crate::handler::keep_alive::ClientKeepAlive;
     use crate::server::connection_manager::ConnectionManager;
-    use crate::subscribe::subscribe_manager::SubscribeManager;
 
     #[tokio::test]
     pub async fn keep_live_time_test() {
@@ -277,12 +270,9 @@ mod test {
             conf.cluster_name.clone(),
         ));
 
-        let subscribe_manager = Arc::new(SubscribeManager::new());
-
         let connection_manager = Arc::new(ConnectionManager::new(cache_manager.clone()));
         let alive = ClientKeepAlive::new(
             client_pool,
-            subscribe_manager,
             connection_manager,
             cache_manager.clone(),
             stop_send,
diff --git a/src/mqtt-broker/src/handler/mod.rs b/src/mqtt-broker/src/handler/mod.rs
index e952767d8..a7233c80b 100644
--- a/src/mqtt-broker/src/handler/mod.rs
+++ b/src/mqtt-broker/src/handler/mod.rs
@@ -32,6 +32,7 @@ pub mod pkid;
 pub mod response;
 pub mod retain;
 pub mod session;
+pub mod sub_auto;
 pub mod sub_exclusive;
 pub mod sub_parse_topic;
 pub mod subscribe;
diff --git a/src/mqtt-broker/src/handler/mqtt.rs b/src/mqtt-broker/src/handler/mqtt.rs
index c390babdf..b5d541f01 100644
--- a/src/mqtt-broker/src/handler/mqtt.rs
+++ b/src/mqtt-broker/src/handler/mqtt.rs
@@ -31,7 +31,9 @@ use storage_adapter::storage::StorageAdapter;
 use super::connection::disconnect_connection;
 use super::offline_message::save_message;
 use super::retain::try_send_retain_message;
-use super::subscribe::{remove_subscribe, save_subscribe};
+use super::sub_auto::start_auto_subscribe;
+use super::subscribe::save_subscribe;
+use super::unsubscribe::remove_subscribe;
 use crate::handler::cache::{
     CacheManager, ConnectionLiveTime, QosAckPackageData, QosAckPackageType,
 };
@@ -199,7 +201,7 @@ where
             }
         };
 
-        match save_session(
+        if let Err(e) = save_session(
             connect_id,
             session.clone(),
             new_session,
@@ -208,18 +210,15 @@ where
         )
         .await
         {
-            Ok(()) => {}
-            Err(e) => {
-                return response_packet_mqtt_connect_fail(
-                    &self.protocol,
-                    ConnectReturnCode::MalformedPacket,
-                    &connect_properties,
-                    Some(e.to_string()),
-                );
-            }
+            return response_packet_mqtt_connect_fail(
+                &self.protocol,
+                ConnectReturnCode::MalformedPacket,
+                &connect_properties,
+                Some(e.to_string()),
+            );
         }
 
-        match save_last_will_message(
+        if let Err(e) = save_last_will_message(
             client_id.clone(),
             &last_will,
             &last_will_properties,
@@ -227,15 +226,21 @@ where
         )
         .await
         {
-            Ok(()) => {}
-            Err(e) => {
-                return response_packet_mqtt_connect_fail(
-                    &self.protocol,
-                    ConnectReturnCode::UnspecifiedError,
-                    &connect_properties,
-                    Some(e.to_string()),
-                );
-            }
+            return response_packet_mqtt_connect_fail(
+                &self.protocol,
+                ConnectReturnCode::UnspecifiedError,
+                &connect_properties,
+                Some(e.to_string()),
+            );
+        }
+
+        if let Err(e) = start_auto_subscribe().await {
+            return response_packet_mqtt_connect_fail(
+                &self.protocol,
+                ConnectReturnCode::UnspecifiedError,
+                &connect_properties,
+                Some(e.to_string()),
+            );
         }
 
         let live_time = ConnectionLiveTime {
@@ -900,7 +905,6 @@ where
             &self.cache_manager,
             &self.client_pool,
             &self.connection_manager,
-            &self.subscribe_manager,
         )
         .await
         {
diff --git a/src/mqtt-broker/src/handler/offline_message.rs b/src/mqtt-broker/src/handler/offline_message.rs
index fa97dd663..6ddfd0502 100644
--- a/src/mqtt-broker/src/handler/offline_message.rs
+++ b/src/mqtt-broker/src/handler/offline_message.rs
@@ -15,7 +15,10 @@
 use std::sync::Arc;
 
 use super::{cache::CacheManager, error::MqttBrokerError, message::build_message_expire};
-use crate::{storage::message::MessageStorage, subscribe::subscribe_manager::SubscribeManager};
+use crate::{
+    observability::metrics::packets::record_messages_dropped_no_subscribers_metrics,
+    storage::message::MessageStorage, subscribe::subscribe_manager::SubscribeManager,
+};
 use metadata_struct::mqtt::{message::MqttMessage, topic::MqttTopic};
 use protocol::mqtt::common::{Publish, PublishProperties};
 use storage_adapter::storage::StorageAdapter;
@@ -40,7 +43,7 @@ where
     if !is_exist_subscribe(subscribe_manager, &topic.topic_name)
         && !cache_manager.get_cluster_info().offline_message.enable
     {
-        //todo
+        record_messages_dropped_no_subscribers_metrics(publish.qos);
         return Ok(None);
     }
 
diff --git a/src/mqtt-broker/src/handler/sub_auto.rs b/src/mqtt-broker/src/handler/sub_auto.rs
new file mode 100644
index 000000000..b7ff924ef
--- /dev/null
+++ b/src/mqtt-broker/src/handler/sub_auto.rs
@@ -0,0 +1,19 @@
+// 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 super::error::MqttBrokerError;
+
+pub async fn start_auto_subscribe() -> Result<(), MqttBrokerError> {
+    Ok(())
+}
diff --git a/src/mqtt-broker/src/handler/sub_exclusive.rs b/src/mqtt-broker/src/handler/sub_exclusive.rs
index 0046a5533..7fd8537c1 100644
--- a/src/mqtt-broker/src/handler/sub_exclusive.rs
+++ b/src/mqtt-broker/src/handler/sub_exclusive.rs
@@ -48,7 +48,7 @@ pub fn check_exclusive_subscribe(
     true
 }
 
-pub fn try_add_exclusive_subscribe(
+pub fn add_exclusive_subscribe(
     subscribe_manager: &Arc<SubscribeManager>,
     path: &str,
     client_id: &str,
@@ -61,7 +61,7 @@ pub fn try_add_exclusive_subscribe(
     subscribe_manager.add_exclusive_subscribe(topic_name, client_id);
 }
 
-pub fn try_remove_exclusive_subscribe(
+pub fn remove_exclusive_subscribe(
     subscribe_manager: &Arc<SubscribeManager>,
     un_subscribe: Unsubscribe,
 ) {
@@ -75,10 +75,7 @@ pub fn try_remove_exclusive_subscribe(
     }
 }
 
-pub fn try_remove_exclusive_subscribe_by_path(
-    subscribe_manager: &Arc<SubscribeManager>,
-    path: &str,
-) {
+pub fn remove_exclusive_subscribe_by_path(subscribe_manager: &Arc<SubscribeManager>, path: &str) {
     if is_exclusive_sub(path) {
         return;
     }
@@ -86,10 +83,3 @@ pub fn try_remove_exclusive_subscribe_by_path(
     let topic_name = decode_exclusive_sub_path_to_topic_name(path);
     subscribe_manager.remove_exclusive_subscribe_by_topic(topic_name);
 }
-
-pub fn try_remove_exclusive_subscribe_by_client_id(
-    subscribe_manager: &Arc<SubscribeManager>,
-    client_id: &str,
-) {
-    subscribe_manager.remove_exclusive_subscribe_by_client_id(client_id);
-}
diff --git a/src/mqtt-broker/src/handler/subscribe.rs b/src/mqtt-broker/src/handler/subscribe.rs
index 6ff6c010b..dff2672f8 100644
--- a/src/mqtt-broker/src/handler/subscribe.rs
+++ b/src/mqtt-broker/src/handler/subscribe.rs
@@ -15,17 +15,14 @@
 use std::sync::Arc;
 
 use common_base::config::broker_mqtt::broker_mqtt_conf;
-use grpc_clients::{
-    placement::mqtt::call::{placement_delete_subscribe, placement_set_subscribe},
-    pool::ClientPool,
-};
+use grpc_clients::{placement::mqtt::call::placement_set_subscribe, pool::ClientPool};
 use log::error;
 use metadata_struct::mqtt::{
     cluster::AvailableFlag, subscribe_data::MqttSubscribe, topic::MqttTopic,
 };
 use protocol::{
-    mqtt::common::{Filter, MqttProtocol, Subscribe, SubscribeProperties, Unsubscribe},
-    placement_center::placement_center_mqtt::{DeleteSubscribeRequest, SetSubscribeRequest},
+    mqtt::common::{Filter, MqttProtocol, Subscribe, SubscribeProperties},
+    placement_center::placement_center_mqtt::SetSubscribeRequest,
 };
 use serde::{Deserialize, Serialize};
 
@@ -38,12 +35,7 @@ use crate::subscribe::{
     subscriber::Subscriber,
 };
 
-use super::{
-    cache::CacheManager,
-    error::MqttBrokerError,
-    sub_exclusive::{try_add_exclusive_subscribe, try_remove_exclusive_subscribe},
-    unsubscribe::unsubscribe_by_path,
-};
+use super::{cache::CacheManager, error::MqttBrokerError, sub_exclusive::add_exclusive_subscribe};
 
 #[derive(Clone, Deserialize, Serialize)]
 struct ParseShareQueueSubscribeRequest {
@@ -115,38 +107,6 @@ pub async fn save_subscribe(
     Ok(())
 }
 
-pub async fn remove_subscribe(
-    client_id: &str,
-    un_subscribe: &Unsubscribe,
-    client_pool: &Arc<ClientPool>,
-    subscribe_manager: &Arc<SubscribeManager>,
-    cache_manager: &Arc<CacheManager>,
-) -> Result<(), MqttBrokerError> {
-    let conf = broker_mqtt_conf();
-
-    for path in un_subscribe.filters.clone() {
-        let request = DeleteSubscribeRequest {
-            cluster_name: conf.cluster_name.to_owned(),
-            client_id: client_id.to_owned(),
-            path: path.clone(),
-        };
-        placement_delete_subscribe(client_pool, &conf.placement_center, request).await?;
-
-        subscribe_manager.remove_subscribe(client_id, &path);
-    }
-
-    try_remove_exclusive_subscribe(subscribe_manager, un_subscribe.clone());
-
-    unsubscribe_by_path(
-        cache_manager,
-        subscribe_manager,
-        client_id,
-        &un_subscribe.filters,
-    );
-
-    Ok(())
-}
-
 #[allow(clippy::too_many_arguments)]
 pub async fn parse_subscribe(
     client_pool: &Arc<ClientPool>,
@@ -172,10 +132,10 @@ pub async fn parse_subscribe(
         == AvailableFlag::Disable;
 
     if enable_exclusive_sub {
-        try_add_exclusive_subscribe(subscribe_manager, &filter.path, client_id);
+        add_exclusive_subscribe(subscribe_manager, &filter.path, client_id);
     }
 
-    if is_share_sub(filter.path.clone()) {
+    if is_share_sub(&filter.path) {
         parse_share_subscribe(
             client_pool,
             subscribe_manager,
@@ -192,7 +152,7 @@ pub async fn parse_subscribe(
             },
         )
         .await;
-    } else if is_queue_sub(filter.path.clone()) {
+    } else if is_queue_sub(&filter.path) {
         parse_queue_subscribe(
             client_pool,
             subscribe_manager,
@@ -210,10 +170,9 @@ pub async fn parse_subscribe(
         )
         .await;
     } else {
-        add_exclusive_subscribe(
+        add_exclusive_push(
             subscribe_manager,
-            &topic.topic_name,
-            &topic.topic_id,
+            topic,
             client_id,
             protocol,
             &sub_identifier,
@@ -227,8 +186,8 @@ async fn parse_share_subscribe(
     subscribe_manager: &Arc<SubscribeManager>,
     req: &mut ParseShareQueueSubscribeRequest,
 ) {
-    let (group_name, sub_name) = decode_share_info(req.filter.path.clone());
-    req.group_name = format!("{}{}", group_name, sub_name);
+    let (group_name, sub_name) = decode_share_info(&req.filter.path);
+    req.group_name = format!("{}_{}", group_name, sub_name);
     req.sub_name = sub_name;
     parse_share_queue_subscribe_common(client_pool, subscribe_manager, req).await;
 }
@@ -238,9 +197,9 @@ async fn parse_queue_subscribe(
     subscribe_manager: &Arc<SubscribeManager>,
     req: &mut ParseShareQueueSubscribeRequest,
 ) {
-    let sub_name = decode_queue_info(req.filter.path.clone());
+    let sub_name = decode_queue_info(&req.filter.path);
     // queueSub is a special shareSub
-    let group_name = format!("$queue{}", sub_name);
+    let group_name = format!("$queue_{}", sub_name);
     req.group_name = group_name;
     req.sub_name = sub_name;
     parse_share_queue_subscribe_common(client_pool, subscribe_manager, req).await;
@@ -252,13 +211,13 @@ async fn parse_share_queue_subscribe_common(
     req: &ParseShareQueueSubscribeRequest,
 ) {
     let conf = broker_mqtt_conf();
-    if path_regex_match(req.topic_name.clone(), req.sub_name.clone()) {
-        match get_share_sub_leader(client_pool.clone(), req.group_name.clone()).await {
+    if path_regex_match(&req.topic_name, &req.sub_name) {
+        match get_share_sub_leader(client_pool, &req.group_name).await {
             Ok(reply) => {
                 if reply.broker_id == conf.broker_id {
-                    add_share_subscribe_leader(subscribe_manager, req).await;
+                    add_share_push_leader(subscribe_manager, req).await;
                 } else {
-                    add_share_subscribe_follower(subscribe_manager, req).await;
+                    add_share_push_follower(subscribe_manager, req).await;
                 }
             }
             Err(e) => {
@@ -271,7 +230,7 @@ async fn parse_share_queue_subscribe_common(
     }
 }
 
-async fn add_share_subscribe_leader(
+async fn add_share_push_leader(
     subscribe_manager: &Arc<SubscribeManager>,
     req: &ParseShareQueueSubscribeRequest,
 ) {
@@ -293,7 +252,7 @@ async fn add_share_subscribe_leader(
     subscribe_manager.add_share_subscribe_leader(&req.sub_name, sub);
 }
 
-async fn add_share_subscribe_follower(
+async fn add_share_push_follower(
     subscribe_manager: &Arc<SubscribeManager>,
     req: &ParseShareQueueSubscribeRequest,
 ) {
@@ -315,30 +274,29 @@ async fn add_share_subscribe_follower(
     );
 }
 
-fn add_exclusive_subscribe(
+fn add_exclusive_push(
     subscribe_manager: &Arc<SubscribeManager>,
-    topic_name: &str,
-    topic_id: &str,
+    topic: &MqttTopic,
     client_id: &str,
     protocol: &MqttProtocol,
     sub_identifier: &Option<usize>,
     filter: &Filter,
 ) {
-    if path_regex_match(topic_name.to_owned(), filter.path.clone()) {
+    if path_regex_match(&topic.topic_name, &filter.path) {
         let sub = Subscriber {
-            protocol: protocol.clone(),
+            protocol: protocol.to_owned(),
             client_id: client_id.to_owned(),
-            topic_name: topic_name.to_owned(),
+            topic_name: topic.topic_name.to_owned(),
             group_name: None,
-            topic_id: topic_id.to_owned(),
+            topic_id: topic.topic_id.to_owned(),
             qos: filter.qos,
             nolocal: filter.nolocal,
             preserve_retain: filter.preserve_retain,
-            retain_forward_rule: filter.retain_forward_rule.clone(),
+            retain_forward_rule: filter.retain_forward_rule.to_owned(),
             subscription_identifier: sub_identifier.to_owned(),
-            sub_path: filter.path.clone(),
+            sub_path: filter.path.to_owned(),
         };
-        subscribe_manager.add_topic_subscribe(topic_name, client_id, &filter.path);
-        subscribe_manager.add_exclusive_push(client_id, &filter.path, topic_id, sub);
+        subscribe_manager.add_topic_subscribe(&topic.topic_name, client_id, &filter.path);
+        subscribe_manager.add_exclusive_push(client_id, &filter.path, &topic.topic_id, sub);
     }
 }
diff --git a/src/mqtt-broker/src/handler/topic.rs b/src/mqtt-broker/src/handler/topic.rs
index c7a7023db..499630a59 100644
--- a/src/mqtt-broker/src/handler/topic.rs
+++ b/src/mqtt-broker/src/handler/topic.rs
@@ -32,10 +32,6 @@ use crate::subscribe::sub_common::{
     decode_queue_info, decode_share_info, is_queue_sub, is_share_sub,
 };
 
-pub fn is_system_topic(_: String) -> bool {
-    true
-}
-
 pub fn payload_format_validator(
     payload: &Bytes,
     payload_format_indicator: u8,
@@ -151,14 +147,14 @@ where
 
 pub fn gen_rewrite_topic(input: &str, pattern: &str, template: &str) -> Option<String> {
     let mut prefix = String::new();
-    let topic = if is_share_sub(input.to_string()) {
-        let (group, group_path) = decode_share_info(input.to_string());
+    let topic = if is_share_sub(input) {
+        let (group, group_path) = decode_share_info(input);
         let share_prefix = format!("$share/{}", group);
         prefix = share_prefix.clone();
         group_path
-    } else if is_queue_sub(input.to_string()) {
+    } else if is_queue_sub(input) {
         prefix = "$queue".to_string();
-        decode_queue_info(input.to_string())
+        decode_queue_info(input)
     } else {
         input.to_string()
     };
diff --git a/src/mqtt-broker/src/handler/topic_rewrite.rs b/src/mqtt-broker/src/handler/topic_rewrite.rs
index 1719d5543..aade8b6ce 100644
--- a/src/mqtt-broker/src/handler/topic_rewrite.rs
+++ b/src/mqtt-broker/src/handler/topic_rewrite.rs
@@ -37,7 +37,7 @@ pub fn process_sub_topic_rewrite(
             {
                 continue;
             }
-            if path_regex_match(filter.path.clone(), topic_rewrite_rule.source_topic.clone()) {
+            if path_regex_match(&filter.path, &topic_rewrite_rule.source_topic) {
                 if let Some(val) = gen_rewrite_topic(
                     &filter.path,
                     &topic_rewrite_rule.regex,
@@ -66,7 +66,7 @@ pub fn process_unsub_topic_rewrite(
             {
                 continue;
             }
-            if path_regex_match(filter.to_string(), topic_rewrite_rule.source_topic.clone()) {
+            if path_regex_match(filter, &topic_rewrite_rule.source_topic) {
                 if let Some(val) = gen_rewrite_topic(
                     filter,
                     &topic_rewrite_rule.regex,
@@ -94,7 +94,7 @@ pub fn process_publish_topic_rewrite(
         {
             continue;
         }
-        if path_regex_match(topic_name.clone(), topic_rewrite_rule.source_topic.clone()) {
+        if path_regex_match(&topic_name, &topic_rewrite_rule.source_topic) {
             let rewrite_topic = gen_rewrite_topic(
                 &topic_name,
                 &topic_rewrite_rule.regex,
diff --git a/src/mqtt-broker/src/handler/unsubscribe.rs b/src/mqtt-broker/src/handler/unsubscribe.rs
index 8a212a23c..5fdca458d 100644
--- a/src/mqtt-broker/src/handler/unsubscribe.rs
+++ b/src/mqtt-broker/src/handler/unsubscribe.rs
@@ -12,35 +12,66 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use super::{
+    cache::CacheManager, error::MqttBrokerError, sub_exclusive::remove_exclusive_subscribe,
+};
 use crate::subscribe::{
     sub_common::{decode_share_info, is_share_sub, path_regex_match},
     subscribe_manager::SubscribeManager,
 };
-use log::error;
+use common_base::config::broker_mqtt::broker_mqtt_conf;
+use grpc_clients::{placement::mqtt::call::placement_delete_subscribe, pool::ClientPool};
+use protocol::{
+    mqtt::common::Unsubscribe, placement_center::placement_center_mqtt::DeleteSubscribeRequest,
+};
 use std::sync::Arc;
 
-use super::cache::CacheManager;
+pub async fn remove_subscribe(
+    client_id: &str,
+    un_subscribe: &Unsubscribe,
+    client_pool: &Arc<ClientPool>,
+    subscribe_manager: &Arc<SubscribeManager>,
+    cache_manager: &Arc<CacheManager>,
+) -> Result<(), MqttBrokerError> {
+    let conf = broker_mqtt_conf();
+
+    for path in un_subscribe.filters.clone() {
+        let request = DeleteSubscribeRequest {
+            cluster_name: conf.cluster_name.to_owned(),
+            client_id: client_id.to_owned(),
+            path: path.clone(),
+        };
+        placement_delete_subscribe(client_pool, &conf.placement_center, request).await?;
+
+        subscribe_manager.remove_subscribe(client_id, &path);
+    }
+
+    remove_exclusive_subscribe(subscribe_manager, un_subscribe.clone());
+
+    unsubscribe_by_path(
+        cache_manager,
+        subscribe_manager,
+        client_id,
+        &un_subscribe.filters,
+    )?;
 
-pub fn stop_push_by_client_id(subscribe_manager: &Arc<SubscribeManager>, client_id: &str) {
-    subscribe_manager.remove_exclusive_push_by_client_id(client_id);
-    subscribe_manager.remove_share_subscribe_leader_by_client_id(client_id);
-    subscribe_manager.remove_share_subscribe_follower_by_client_id(client_id);
+    Ok(())
 }
 
-pub fn unsubscribe_by_path(
+fn unsubscribe_by_path(
     cache_manager: &Arc<CacheManager>,
     subscribe_manager: &Arc<SubscribeManager>,
     client_id: &str,
     filter_path: &[String],
-) {
+) -> Result<(), MqttBrokerError> {
     for (topic_name, _) in cache_manager.topic_info.clone() {
         for path in filter_path {
-            if !path_regex_match(topic_name.clone(), path.clone()) {
+            if !path_regex_match(&topic_name, path) {
                 continue;
             }
 
-            if is_share_sub(path.clone()) {
-                let (group_name, sub_name) = decode_share_info(path.clone());
+            if is_share_sub(path) {
+                let (group_name, sub_name) = decode_share_info(path);
                 // share leader
                 for (key, data) in subscribe_manager.share_leader_push.clone() {
                     let mut flag = false;
@@ -63,10 +94,7 @@ pub fn unsubscribe_by_path(
 
                     if flag {
                         if let Some(sx) = subscribe_manager.share_leader_push_thread.get(&key) {
-                            match sx.send(true) {
-                                Ok(_) => {}
-                                Err(e) => error!("{}", e),
-                            }
+                            sx.send(true)?;
                         }
                     }
                 }
@@ -76,10 +104,7 @@ pub fn unsubscribe_by_path(
                     if data.client_id == *client_id && data.filter.path == *path {
                         subscribe_manager.share_follower_resub.remove(&key);
                         if let Some(sx) = subscribe_manager.share_follower_resub_thread.get(&key) {
-                            match sx.send(true) {
-                                Ok(_) => {}
-                                Err(e) => error!("{}", e),
-                            }
+                            sx.send(true)?;
                         }
                     }
                 }
@@ -87,10 +112,7 @@ pub fn unsubscribe_by_path(
                 for (key, subscriber) in subscribe_manager.exclusive_push.clone() {
                     if subscriber.client_id == *client_id && subscriber.sub_path == *path {
                         if let Some(sx) = subscribe_manager.exclusive_push_thread.get(&key) {
-                            match sx.send(true) {
-                                Ok(_) => {}
-                                Err(e) => error!("{}", e),
-                            }
+                            sx.send(true)?;
                             subscribe_manager.exclusive_push.remove(&key);
                         }
                         subscribe_manager.remove_topic_subscribe_by_path(
@@ -102,4 +124,5 @@ pub fn unsubscribe_by_path(
             }
         }
     }
+    Ok(())
 }
diff --git a/src/mqtt-broker/src/handler/validator.rs b/src/mqtt-broker/src/handler/validator.rs
index 8fdfdc628..1d297463e 100644
--- a/src/mqtt-broker/src/handler/validator.rs
+++ b/src/mqtt-broker/src/handler/validator.rs
@@ -504,15 +504,6 @@ pub async fn un_subscribe_validator(
         ));
     }
 
-    if authentication_acl() {
-        return Some(response_packet_mqtt_unsuback(
-            connection,
-            un_subscribe.pkid,
-            vec![UnsubAckReason::NotAuthorized],
-            None,
-        ));
-    }
-
     for path in un_subscribe.filters.clone() {
         if subscribe_manager.get_subscribe(client_id, &path).is_none() {
             return Some(response_packet_mqtt_unsuback(
diff --git a/src/mqtt-broker/src/lib.rs b/src/mqtt-broker/src/lib.rs
index c697f2b74..f1641b8e2 100644
--- a/src/mqtt-broker/src/lib.rs
+++ b/src/mqtt-broker/src/lib.rs
@@ -310,7 +310,6 @@ where
     fn start_keep_alive_thread(&self, stop_send: broadcast::Sender<bool>) {
         let mut keep_alive = ClientKeepAlive::new(
             self.client_pool.clone(),
-            self.subscribe_manager.clone(),
             self.connection_manager.clone(),
             self.cache_manager.clone(),
             stop_send,
diff --git a/src/mqtt-broker/src/observability/metrics/packets.rs b/src/mqtt-broker/src/observability/metrics/packets.rs
index 7346c206f..f1ab37131 100644
--- a/src/mqtt-broker/src/observability/metrics/packets.rs
+++ b/src/mqtt-broker/src/observability/metrics/packets.rs
@@ -244,6 +244,13 @@ common_base::register_gauge_metric!(
     QosLabel
 );
 
+common_base::register_gauge_metric!(
+    MESSAGES_DROPPED_NO_SUBSCRIBERS,
+    "messages.dropped.no_subscribers",
+    "Number of messages dropped due to no subscribers",
+    QosLabel
+);
+
 // Record the packet-related metrics received by the server for failed resolution
 pub fn record_received_error_metrics(network_type: NetworkConnectionType) {
     let labe = NetworkLabel {
@@ -367,6 +374,12 @@ pub fn record_retain_sent_metrics(qos: QoS) {
     common_base::gauge_metric_inc!(RETAIN_PACKETS_SEND, label);
 }
 
+pub fn record_messages_dropped_no_subscribers_metrics(qos: QoS) {
+    let qos_str = (qos as u8).to_string();
+    let label = QosLabel { qos: qos_str };
+    common_base::gauge_metric_inc!(MESSAGES_DROPPED_NO_SUBSCRIBERS, label);
+}
+
 #[cfg(test)]
 mod test {
     use super::*;
diff --git a/src/mqtt-broker/src/server/grpc/inner.rs b/src/mqtt-broker/src/server/grpc/inner.rs
index aa8fb3ade..d3e39a47b 100644
--- a/src/mqtt-broker/src/server/grpc/inner.rs
+++ b/src/mqtt-broker/src/server/grpc/inner.rs
@@ -29,8 +29,6 @@ use tonic::{Request, Response, Status};
 use crate::handler::cache::CacheManager;
 use crate::handler::cache_update::update_cache_metadata;
 use crate::handler::lastwill::send_last_will_message;
-use crate::handler::sub_exclusive::try_remove_exclusive_subscribe_by_client_id;
-use crate::handler::unsubscribe::stop_push_by_client_id;
 use crate::subscribe::subscribe_manager::SubscribeManager;
 
 pub struct GrpcInnerServices<S> {
@@ -87,10 +85,10 @@ where
         if req.client_id.is_empty() {
             return Err(Status::cancelled("Client ID cannot be empty".to_string()));
         }
+
         for client_id in req.client_id {
-            try_remove_exclusive_subscribe_by_client_id(&self.subscribe_manager, &client_id);
+            self.subscribe_manager.remove_client_id(&client_id);
             self.cache_manager.remove_session(&client_id);
-            stop_push_by_client_id(&self.subscribe_manager, &client_id);
         }
 
         return Ok(Response::new(DeleteSessionReply::default()));
diff --git a/src/mqtt-broker/src/server/tcp/response.rs b/src/mqtt-broker/src/server/tcp/response.rs
index 1086704cf..a19f7eeee 100644
--- a/src/mqtt-broker/src/server/tcp/response.rs
+++ b/src/mqtt-broker/src/server/tcp/response.rs
@@ -28,13 +28,11 @@ use crate::handler::connection::disconnect_connection;
 use crate::observability::metrics::server::{metrics_request_queue, metrics_response_queue};
 use crate::server::connection_manager::ConnectionManager;
 use crate::server::packet::ResponsePackage;
-use crate::subscribe::subscribe_manager::SubscribeManager;
 
 pub(crate) async fn response_process(
     response_process_num: usize,
     connection_manager: Arc<ConnectionManager>,
     cache_manager: Arc<CacheManager>,
-    subscribe_manager: Arc<SubscribeManager>,
     mut response_queue_rx: Receiver<ResponsePackage>,
     client_pool: Arc<ClientPool>,
     stop_sx: broadcast::Sender<bool>,
@@ -48,7 +46,6 @@ pub(crate) async fn response_process(
             stop_sx.clone(),
             connection_manager,
             cache_manager,
-            subscribe_manager,
             client_pool,
         );
 
@@ -102,7 +99,6 @@ pub(crate) fn response_child_process(
     stop_sx: broadcast::Sender<bool>,
     connection_manager: Arc<ConnectionManager>,
     cache_manager: Arc<CacheManager>,
-    subscribe_manager: Arc<SubscribeManager>,
     client_pool: Arc<ClientPool>,
 ) {
     for index in 1..=response_process_num {
@@ -113,7 +109,6 @@ pub(crate) fn response_child_process(
         let raw_connect_manager = connection_manager.clone();
         let raw_cache_manager = cache_manager.clone();
         let raw_client_pool = client_pool.clone();
-        let raw_subscribe_manager = subscribe_manager.clone();
         tokio::spawn(async move {
             debug!("TCP Server response process thread {index} start successfully.");
 
@@ -158,7 +153,6 @@ pub(crate) fn response_child_process(
                                         &raw_cache_manager,
                                         &raw_client_pool,
                                         &raw_connect_manager,
-                                        &raw_subscribe_manager,
                                     ).await{
                                         Ok(()) => {},
                                         Err(e) => error!("{}",e)
diff --git a/src/mqtt-broker/src/server/tcp/server.rs b/src/mqtt-broker/src/server/tcp/server.rs
index ebe204ec4..39c02b622 100644
--- a/src/mqtt-broker/src/server/tcp/server.rs
+++ b/src/mqtt-broker/src/server/tcp/server.rs
@@ -65,7 +65,6 @@ pub async fn start_tcp_server<S>(
         proc_config,
         stop_sx.clone(),
         connection_manager.clone(),
-        subscribe_manager.clone(),
         cache_manager.clone(),
         client_pool.clone(),
     );
@@ -76,7 +75,6 @@ pub async fn start_tcp_server<S>(
         proc_config,
         stop_sx.clone(),
         connection_manager,
-        subscribe_manager.clone(),
         cache_manager,
         client_pool,
     );
@@ -89,7 +87,6 @@ struct TcpServer<S> {
     command: Command<S>,
     connection_manager: Arc<ConnectionManager>,
     cache_manager: Arc<CacheManager>,
-    subscribe_manager: Arc<SubscribeManager>,
     client_pool: Arc<ClientPool>,
     accept_thread_num: usize,
     handler_process_num: usize,
@@ -114,13 +111,11 @@ where
         proc_config: ProcessorConfig,
         stop_sx: broadcast::Sender<bool>,
         connection_manager: Arc<ConnectionManager>,
-        subscribe_manager: Arc<SubscribeManager>,
         cache_manager: Arc<CacheManager>,
         client_pool: Arc<ClientPool>,
     ) -> Self {
         Self {
             command,
-            subscribe_manager,
             cache_manager,
             client_pool,
             connection_manager,
@@ -169,7 +164,6 @@ where
             self.response_process_num,
             self.connection_manager.clone(),
             self.cache_manager.clone(),
-            self.subscribe_manager.clone(),
             response_queue_rx,
             self.client_pool.clone(),
             self.stop_sx.clone(),
@@ -216,7 +210,6 @@ where
             self.response_process_num,
             self.connection_manager.clone(),
             self.cache_manager.clone(),
-            self.subscribe_manager.clone(),
             response_queue_rx,
             self.client_pool.clone(),
             self.stop_sx.clone(),
diff --git a/src/mqtt-broker/src/subscribe/share_follower_resub.rs b/src/mqtt-broker/src/subscribe/share_follower_resub.rs
index 53f679ce8..c3b536598 100644
--- a/src/mqtt-broker/src/subscribe/share_follower_resub.rs
+++ b/src/mqtt-broker/src/subscribe/share_follower_resub.rs
@@ -84,8 +84,7 @@ impl ShareFollowerResub {
             let metadata_cache = self.cache_manager.clone();
             let connection_manager = self.connection_manager.clone();
 
-            match get_share_sub_leader(self.client_pool.clone(), share_sub.group_name.clone()).await
-            {
+            match get_share_sub_leader(&self.client_pool, &share_sub.group_name).await {
                 Ok(reply) => {
                     let conf = broker_mqtt_conf();
                     if conf.broker_id == reply.broker_id {
diff --git a/src/mqtt-broker/src/subscribe/sub_common.rs b/src/mqtt-broker/src/subscribe/sub_common.rs
index 2b08601d8..b0c0ca751 100644
--- a/src/mqtt-broker/src/subscribe/sub_common.rs
+++ b/src/mqtt-broker/src/subscribe/sub_common.rs
@@ -69,23 +69,23 @@ pub fn sub_path_validator(sub_path: String) -> bool {
     true
 }
 
-pub fn path_regex_match(topic_name: String, sub_path: String) -> bool {
-    let path = if is_share_sub(sub_path.clone()) {
-        let (_, group_path) = decode_share_info(sub_path.clone());
+pub fn path_regex_match(topic_name: &str, sub_path: &str) -> bool {
+    let path = if is_share_sub(sub_path) {
+        let (_, group_path) = decode_share_info(sub_path);
         group_path
-    } else if is_queue_sub(sub_path.clone()) {
-        decode_queue_info(sub_path.clone())
+    } else if is_queue_sub(sub_path) {
+        decode_queue_info(sub_path)
     } else {
-        sub_path.clone()
+        sub_path.to_owned()
     };
 
-    let topic = if is_share_sub(topic_name.clone()) {
-        let (_, group_path) = decode_share_info(topic_name.clone());
+    let topic = if is_share_sub(topic_name) {
+        let (_, group_path) = decode_share_info(topic_name);
         group_path
-    } else if is_queue_sub(topic_name.clone()) {
-        decode_queue_info(topic_name.clone())
+    } else if is_queue_sub(topic_name) {
+        decode_queue_info(topic_name)
     } else {
-        topic_name.clone()
+        topic_name.to_owned()
     };
 
     // Path perfect matching
@@ -124,22 +124,22 @@ pub async fn get_sub_topic_id_list(
 ) -> Vec<String> {
     let mut result = Vec::new();
     for (topic_id, topic_name) in metadata_cache.topic_id_name.clone() {
-        if path_regex_match(topic_name.clone(), sub_path.to_string()) {
+        if path_regex_match(&topic_name, sub_path) {
             result.push(topic_id);
         }
     }
     result
 }
 
-pub fn is_share_sub(sub_name: String) -> bool {
+pub fn is_share_sub(sub_name: &str) -> bool {
     sub_name.starts_with(SHARE_SUB_PREFIX)
 }
 
-pub fn is_queue_sub(sub_name: String) -> bool {
+pub fn is_queue_sub(sub_name: &str) -> bool {
     sub_name.starts_with(QUEUE_SUB_PREFIX)
 }
 
-pub fn decode_share_info(sub_name: String) -> (String, String) {
+pub fn decode_share_info(sub_name: &str) -> (String, String) {
     let mut str_slice: Vec<&str> = sub_name.split("/").collect();
     str_slice.remove(0);
     let group_name = str_slice.remove(0).to_string();
@@ -147,22 +147,22 @@ pub fn decode_share_info(sub_name: String) -> (String, String) {
     (group_name, sub_name)
 }
 
-pub fn decode_queue_info(sub_name: String) -> String {
+pub fn decode_queue_info(sub_name: &str) -> String {
     let mut str_slice: Vec<&str> = sub_name.split("/").collect();
     str_slice.remove(0);
     format!("/{}", str_slice.join("/"))
 }
 
 pub async fn get_share_sub_leader(
-    client_pool: Arc<ClientPool>,
-    group_name: String,
+    client_pool: &Arc<ClientPool>,
+    group_name: &String,
 ) -> Result<GetShareSubLeaderReply, CommonError> {
     let conf = broker_mqtt_conf();
     let req = GetShareSubLeaderRequest {
-        cluster_name: conf.cluster_name.clone(),
-        group_name,
+        cluster_name: conf.cluster_name.to_owned(),
+        group_name: group_name.to_owned(),
     };
-    match placement_get_share_sub_leader(&client_pool, &conf.placement_center, req).await {
+    match placement_get_share_sub_leader(client_pool, &conf.placement_center, req).await {
         Ok(reply) => Ok(reply),
         Err(e) => Err(e),
     }
@@ -478,16 +478,16 @@ mod tests {
         let sub3 = "$share/consumer1/sport/#".to_string();
         let sub4 = "$share/comsumer1/finance/#".to_string();
 
-        assert!(is_share_sub(sub1));
-        assert!(is_share_sub(sub2));
-        assert!(is_share_sub(sub3));
-        assert!(is_share_sub(sub4));
+        assert!(is_share_sub(&sub1));
+        assert!(is_share_sub(&sub2));
+        assert!(is_share_sub(&sub3));
+        assert!(is_share_sub(&sub4));
 
         let sub5 = "/comsumer1/$share/finance/#".to_string();
         let sub6 = "/comsumer1/$share/finance/$share".to_string();
 
-        assert!(!is_share_sub(sub5));
-        assert!(!is_share_sub(sub6));
+        assert!(!is_share_sub(&sub5));
+        assert!(!is_share_sub(&sub6));
     }
 
     #[tokio::test]
@@ -498,19 +498,19 @@ mod tests {
         let sub3 = "$share/consumer1/sport/#".to_string();
         let sub4 = "$share/comsumer1/finance/#".to_string();
 
-        let (group_name, topic_name) = decode_share_info(sub1);
+        let (group_name, topic_name) = decode_share_info(&sub1);
         assert_eq!(group_name, "consumer1".to_string());
         assert_eq!(topic_name, "/sport/tennis/+".to_string());
 
-        let (group_name, topic_name) = decode_share_info(sub2);
+        let (group_name, topic_name) = decode_share_info(&sub2);
         assert_eq!(group_name, "consumer2".to_string());
         assert_eq!(topic_name, "/sport/tennis/+".to_string());
 
-        let (group_name, topic_name) = decode_share_info(sub3);
+        let (group_name, topic_name) = decode_share_info(&sub3);
         assert_eq!(group_name, "consumer1".to_string());
         assert_eq!(topic_name, "/sport/#".to_string());
 
-        let (group_name, topic_name) = decode_share_info(sub4);
+        let (group_name, topic_name) = decode_share_info(&sub4);
         assert_eq!(group_name, "consumer1".to_string());
         assert_eq!(topic_name, "/finance/#".to_string());
     }
@@ -518,55 +518,55 @@ mod tests {
     fn path_regex_match_test() {
         let topic_name = "/loboxu/test".to_string();
         let sub_regex = "/loboxu/#".to_string();
-        assert!(path_regex_match(topic_name, sub_regex));
+        assert!(path_regex_match(&topic_name, &sub_regex));
 
         let topic_name = "/topic/test".to_string();
         let sub_regex = "/topic/test".to_string();
-        assert!(path_regex_match(topic_name, sub_regex));
+        assert!(path_regex_match(&topic_name, &sub_regex));
 
         let topic_name = r"/sensor/1/temperature".to_string();
         let sub_regex = r"/sensor/+/temperature".to_string();
-        assert!(path_regex_match(topic_name, sub_regex));
+        assert!(path_regex_match(&topic_name, &sub_regex));
 
         let topic_name = r"/sensor/1/2/temperature3".to_string();
         let sub_regex = r"/sensor/+/temperature".to_string();
-        assert!(!path_regex_match(topic_name, sub_regex));
+        assert!(!path_regex_match(&topic_name, &sub_regex));
 
         let topic_name = r"/sensor/temperature3".to_string();
         let sub_regex = r"/sensor/+/temperature".to_string();
-        assert!(!path_regex_match(topic_name, sub_regex));
+        assert!(!path_regex_match(&topic_name, &sub_regex));
 
         let topic_name = r"/sensor/temperature3".to_string();
         let sub_regex = r"/sensor/+".to_string();
-        assert!(path_regex_match(topic_name, sub_regex));
+        assert!(path_regex_match(&topic_name, &sub_regex));
 
         let topic_name = r"/sensor/temperature3/tmpq".to_string();
         let sub_regex = r"/sensor/#".to_string();
-        assert!(path_regex_match(topic_name, sub_regex));
+        assert!(path_regex_match(&topic_name, &sub_regex));
 
         let topic_name = "/topic/test".to_string();
         let sub_regex = "$share/groupname/topic/test".to_string();
-        assert!(path_regex_match(topic_name, sub_regex));
+        assert!(path_regex_match(&topic_name, &sub_regex));
 
         let topic_name = r"/sensor/1/temperature".to_string();
         let sub_regex = r"$share/groupname/sensor/+/temperature".to_string();
-        assert!(path_regex_match(topic_name, sub_regex));
+        assert!(path_regex_match(&topic_name, &sub_regex));
 
         let topic_name = r"/sensor/1/2/temperature3".to_string();
         let sub_regex = r"$share/groupname/sensor/+/temperature".to_string();
-        assert!(!path_regex_match(topic_name, sub_regex));
+        assert!(!path_regex_match(&topic_name, &sub_regex));
 
         let topic_name = r"/sensor/temperature3".to_string();
         let sub_regex = r"$share/groupname/sensor/+/temperature".to_string();
-        assert!(!path_regex_match(topic_name, sub_regex));
+        assert!(!path_regex_match(&topic_name, &sub_regex));
 
         let topic_name = r"/sensor/temperature3".to_string();
         let sub_regex = r"$share/groupname/sensor/+".to_string();
-        assert!(path_regex_match(topic_name, sub_regex));
+        assert!(path_regex_match(&topic_name, &sub_regex));
 
         let topic_name = r"/sensor/temperature3/tmpq".to_string();
         let sub_regex = r"$share/groupname/sensor/#".to_string();
-        assert!(path_regex_match(topic_name, sub_regex));
+        assert!(path_regex_match(&topic_name, &sub_regex));
     }
 
     #[test]
diff --git a/src/mqtt-broker/src/subscribe/subscribe_manager.rs b/src/mqtt-broker/src/subscribe/subscribe_manager.rs
index f78ce280b..2658419c3 100644
--- a/src/mqtt-broker/src/subscribe/subscribe_manager.rs
+++ b/src/mqtt-broker/src/subscribe/subscribe_manager.rs
@@ -120,7 +120,7 @@ impl SubscribeManager {
         self.exclusive_push.insert(key, sub);
     }
 
-    pub fn remove_exclusive_push_by_client_id(&self, client_id: &str) {
+    fn remove_exclusive_push_by_client_id(&self, client_id: &str) {
         for (key, subscriber) in self.exclusive_push.clone() {
             if subscriber.client_id == *client_id {
                 self.exclusive_push.remove(&key);
@@ -132,7 +132,7 @@ impl SubscribeManager {
         }
     }
 
-    // push by share subscribe
+    // Leader push by share subscribe
     pub fn add_share_subscribe_leader(&self, sub_name: &str, sub: Subscriber) {
         let group_name = sub.group_name.clone().unwrap();
         let share_leader_key = self.share_leader_key(&group_name, sub_name, &sub.topic_id);
@@ -157,7 +157,7 @@ impl SubscribeManager {
         }
     }
 
-    pub fn remove_share_subscribe_leader_by_client_id(&self, client_id: &str) {
+    fn remove_share_subscribe_leader_by_client_id(&self, client_id: &str) {
         for (key, share_sub) in self.share_leader_push.clone() {
             for (sub_key, subscriber) in share_sub.sub_list {
                 if subscriber.client_id == *client_id {
@@ -172,6 +172,7 @@ impl SubscribeManager {
         }
     }
 
+    // Follower resub by share subscribe
     pub fn add_share_subscribe_follower(
         &self,
         client_id: &str,
@@ -183,7 +184,7 @@ impl SubscribeManager {
         self.share_follower_resub.insert(key, share_sub);
     }
 
-    pub fn remove_share_subscribe_follower_by_client_id(&self, client_id: &str) {
+    fn remove_share_subscribe_follower_by_client_id(&self, client_id: &str) {
         for (key, share_sub) in self.share_follower_resub.clone() {
             if share_sub.client_id == *client_id {
                 self.share_follower_resub.remove(&key);
@@ -242,7 +243,7 @@ impl SubscribeManager {
         self.exclusive_subscribe.remove(topic);
     }
 
-    pub fn remove_exclusive_subscribe_by_client_id(&self, client_id: &str) {
+    fn remove_exclusive_subscribe_by_client_id(&self, client_id: &str) {
         for (topic, cid) in self.exclusive_subscribe.clone() {
             if cid == *client_id {
                 self.exclusive_subscribe.remove(&topic);
@@ -250,6 +251,13 @@ impl SubscribeManager {
         }
     }
 
+    pub fn remove_client_id(&self, client_id: &str) {
+        self.remove_exclusive_push_by_client_id(client_id);
+        self.remove_share_subscribe_leader_by_client_id(client_id);
+        self.remove_share_subscribe_follower_by_client_id(client_id);
+        self.remove_exclusive_subscribe_by_client_id(client_id);
+    }
+
     // key
     fn subscribe_key(&self, client_id: &str, path: &str) -> String {
         format!("{}_{}", client_id, path)


Test patch: