Repository: robustmq/robustmq (language: rust)

Issue:
refactor: Replace CommonError in mqtt broker
### What type of enhancement is this?

Refactor

### What does the enhancement do?

Replace all CommonErrors in MQTT Broker with MqttBrokerError.
![image](https://github.com/user-attachments/assets/673ee45a-1ee9-44ea-8c56-8d7282a2399c)


### Implementation challenges

_No response_


Discussion / hints:

Fix patch:
diff --git a/src/mqtt-broker/src/handler/connection.rs b/src/mqtt-broker/src/handler/connection.rs
index da5e7bd33..169dae26b 100644
--- a/src/mqtt-broker/src/handler/connection.rs
+++ b/src/mqtt-broker/src/handler/connection.rs
@@ -15,7 +15,6 @@
 use std::net::SocketAddr;
 use std::sync::Arc;
 
-use common_base::error::common::CommonError;
 use common_base::tools::{now_second, unique_id};
 use grpc_clients::pool::ClientPool;
 use metadata_struct::mqtt::cluster::MqttClusterDynamicConfig;
@@ -23,6 +22,7 @@ use metadata_struct::mqtt::connection::{ConnectionConfig, MQTTConnection};
 use protocol::mqtt::common::{Connect, ConnectProperties};
 
 use super::cache::CacheManager;
+use super::error::MqttBrokerError;
 use super::keep_alive::client_keep_live_time;
 use crate::server::connection_manager::ConnectionManager;
 use crate::storage::session::SessionStorage;
@@ -116,7 +116,7 @@ pub async fn disconnect_connection(
     client_pool: &Arc<ClientPool>,
     connection_manager: &Arc<ConnectionManager>,
     subscribe_manager: &Arc<SubscribeManager>,
-) -> Result<(), CommonError> {
+) -> Result<(), MqttBrokerError> {
     subscribe_manager
         .remove_exclusive_subscribe_by_client_id(client_id)
         .await?;
@@ -135,7 +135,7 @@ pub async fn disconnect_connection(
     {
         Ok(_) => {}
         Err(e) => {
-            return Err(e);
+            return Err(MqttBrokerError::CommonError(e.to_string()));
         }
     }
 
diff --git a/src/mqtt-broker/src/handler/pkid.rs b/src/mqtt-broker/src/handler/pkid.rs
index 80315630b..25a27f9a9 100644
--- a/src/mqtt-broker/src/handler/pkid.rs
+++ b/src/mqtt-broker/src/handler/pkid.rs
@@ -15,7 +15,6 @@
 use std::sync::Arc;
 
 use common_base::config::broker_mqtt::broker_mqtt_conf;
-use common_base::error::common::CommonError;
 use grpc_clients::placement::inner::call::{
     delete_idempotent_data, exists_idempotent_data, set_idempotent_data,
 };
@@ -25,13 +24,14 @@ use protocol::placement_center::placement_center_inner::{
 };
 
 use super::cache::CacheManager;
+use super::error::MqttBrokerError;
 
 pub async fn pkid_save(
     cache_manager: &Arc<CacheManager>,
     client_pool: &Arc<ClientPool>,
     client_id: &str,
     pkid: u16,
-) -> Result<(), CommonError> {
+) -> Result<(), MqttBrokerError> {
     if cache_manager
         .get_cluster_info()
         .protocol
@@ -48,7 +48,7 @@ pub async fn pkid_save(
                 return Ok(());
             }
             Err(e) => {
-                return Err(e);
+                return Err(MqttBrokerError::CommonError(e.to_string()));
             }
         }
     } else {
@@ -62,7 +62,7 @@ pub async fn pkid_exists(
     client_pool: &Arc<ClientPool>,
     client_id: &str,
     pkid: u16,
-) -> Result<bool, CommonError> {
+) -> Result<bool, MqttBrokerError> {
     if cache_manager
         .get_cluster_info()
         .protocol
@@ -76,7 +76,7 @@ pub async fn pkid_exists(
         };
         match exists_idempotent_data(client_pool, &conf.placement_center, request).await {
             Ok(reply) => Ok(reply.exists),
-            Err(e) => Err(e),
+            Err(e) => Err(MqttBrokerError::CommonError(e.to_string())),
         }
     } else {
         Ok(cache_manager.get_client_pkid(client_id, pkid).is_some())
@@ -88,7 +88,7 @@ pub async fn pkid_delete(
     client_pool: &Arc<ClientPool>,
     client_id: &str,
     pkid: u16,
-) -> Result<(), CommonError> {
+) -> Result<(), MqttBrokerError> {
     if cache_manager
         .get_cluster_info()
         .protocol
@@ -105,7 +105,7 @@ pub async fn pkid_delete(
                 return Ok(());
             }
             Err(e) => {
-                return Err(e);
+                return Err(MqttBrokerError::CommonError(e.to_string()));
             }
         }
     } else {
diff --git a/src/mqtt-broker/src/handler/session.rs b/src/mqtt-broker/src/handler/session.rs
index 5ff9b89ef..4c0565f43 100644
--- a/src/mqtt-broker/src/handler/session.rs
+++ b/src/mqtt-broker/src/handler/session.rs
@@ -15,13 +15,13 @@
 use std::sync::Arc;
 
 use common_base::config::broker_mqtt::broker_mqtt_conf;
-use common_base::error::common::CommonError;
 use common_base::tools::now_second;
 use grpc_clients::pool::ClientPool;
 use metadata_struct::mqtt::session::MqttSession;
 use protocol::mqtt::common::{Connect, ConnectProperties, LastWill, LastWillProperties};
 
 use super::cache::CacheManager;
+use super::error::MqttBrokerError;
 use super::lastwill::last_will_delay_interval;
 use crate::storage::session::SessionStorage;
 
@@ -35,7 +35,7 @@ pub async fn build_session(
     last_will_properties: &Option<LastWillProperties>,
     client_pool: &Arc<ClientPool>,
     cache_manager: &Arc<CacheManager>,
-) -> Result<(MqttSession, bool), CommonError> {
+) -> Result<(MqttSession, bool), MqttBrokerError> {
     let session_expiry = session_expiry_interval(cache_manager, connect_properties);
     let is_contain_last_will = !last_will.is_none();
     let last_will_delay_interval = last_will_delay_interval(last_will_properties);
@@ -54,7 +54,7 @@ pub async fn build_session(
                 true,
             ),
             Err(e) => {
-                return Err(e);
+                return Err(MqttBrokerError::CommonError(e.to_string()));
             }
         }
     } else {
@@ -82,14 +82,14 @@ pub async fn save_session(
     new_session: bool,
     client_id: String,
     client_pool: &Arc<ClientPool>,
-) -> Result<(), CommonError> {
+) -> Result<(), MqttBrokerError> {
     let conf = broker_mqtt_conf();
     let session_storage = SessionStorage::new(client_pool.clone());
     if new_session {
         match session_storage.set_session(client_id, &session).await {
             Ok(_) => {}
             Err(e) => {
-                return Err(e);
+                return Err(MqttBrokerError::CommonError(e.to_string()));
             }
         }
     } else {
@@ -99,7 +99,7 @@ pub async fn save_session(
         {
             Ok(_) => {}
             Err(e) => {
-                return Err(e);
+                return Err(MqttBrokerError::CommonError(e.to_string()));
             }
         }
     }


Test patch: