Repository: robustmq/robustmq (language: rust)

Issue:
MINOR: Optimization of Placement Center Leader switching logic
### What type of enhancement is this?

Refactor

### What does the enhancement do?

Optimization of Placement Center Leader switching logic

### Implementation challenges

_No response_


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 e60802df8..d090c974f 100644
--- a/docs/.design/plan/mqtt-stable-to-do.md
+++ b/docs/.design/plan/mqtt-stable-to-do.md
@@ -3,23 +3,19 @@
 2. 订阅逻辑优化【done】
 3. MQTT 测试用例失败 case 修复
 4. 自动订阅
-5. 数据集成
+5. 延时发布
+6. 消息验证：确保消息的完整性和合法性
+7. 数据集成
    1. sink local file
-   3. Kafka 集成
-6. 限流
+   2. Kafka 集成
+8. 限流
    1. 速率限制
    2. QOS 限制
-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. 多协议网关
+9. 可观测体系搭建 【done】 需要继续丰富指标
+10. ACL/User 【done】
+11. Schema: Schema Registry、灵活地消息格式转换：JSON、Avro、Protobuf、Custom codec (HTTP/gRPC)
+12. Cloud-Native & K8s
+13. MQTT Over QUIC
 
 ## 测试用例覆盖度
 1. 代码测试覆盖度
diff --git a/src/placement-center/src/lib.rs b/src/placement-center/src/lib.rs
index 6ad26fe37..30ff23162 100644
--- a/src/placement-center/src/lib.rs
+++ b/src/placement-center/src/lib.rs
@@ -17,7 +17,7 @@ use std::time::Duration;
 
 use common_base::config::placement_center::placement_center_conf;
 use grpc_clients::pool::ClientPool;
-use log::{error, info};
+use log::info;
 use mqtt::controller::call_broker::{mqtt_call_thread_manager, MQTTInnerCallManager};
 use openraft::Raft;
 use protocol::placement_center::placement_center_inner::placement_center_service_server::PlacementCenterServiceServer;
@@ -25,6 +25,7 @@ use protocol::placement_center::placement_center_journal::engine_service_server:
 use protocol::placement_center::placement_center_kv::kv_service_server::KvServiceServer;
 use protocol::placement_center::placement_center_mqtt::mqtt_service_server::MqttServiceServer;
 use protocol::placement_center::placement_center_openraft::open_raft_service_server::OpenRaftServiceServer;
+use raft::leadership::monitoring_leader_transition;
 use server::grpc::service_inner::GrpcPlacementService;
 use server::grpc::service_journal::GrpcEngineService;
 use server::grpc::service_kv::GrpcKvService;
@@ -40,9 +41,7 @@ use crate::core::cache::PlacementCacheManager;
 use crate::core::controller::ClusterController;
 use crate::journal::cache::{load_journal_cache, JournalCacheManager};
 use crate::journal::controller::call_node::{journal_call_thread_manager, JournalInnerCallManager};
-use crate::journal::controller::StorageEngineController;
 use crate::mqtt::cache::MqttCacheManager;
-use crate::mqtt::controller::MqttController;
 use crate::raft::raft_node::{create_raft_node, start_openraft_node};
 use crate::raft::typeconfig::TypeConfig;
 use crate::route::apply::RaftMachineApply;
@@ -124,7 +123,7 @@ impl PlacementCenter {
 
         let placement_center_storage = Arc::new(RaftMachineApply::new(openraft_node.clone()));
 
-        self.start_controller(placement_center_storage.clone(), stop_send.clone());
+        self.start_heartbeat(placement_center_storage.clone(), stop_send.clone());
 
         self.start_raft_machine(openraft_node.clone());
 
@@ -132,43 +131,26 @@ impl PlacementCenter {
 
         self.start_grpc_server(placement_center_storage.clone());
 
-        self.monitoring_leader_transition(openraft_node.clone());
+        self.monitoring_leader_transition(openraft_node.clone(), placement_center_storage.clone());
 
         self.awaiting_stop(stop_send).await;
     }
 
-    pub fn monitoring_leader_transition(&self, raft: Raft<TypeConfig>) {
+    pub fn monitoring_leader_transition(
+        &self,
+        raft: Raft<TypeConfig>,
+        raft_machine_apply: Arc<RaftMachineApply>,
+    ) {
         info!("Initiate Monitoring of Raft Leader Transitions");
-        let mut metrics_rx = raft.metrics();
-        tokio::spawn(async move {
-            let mut last_leader: Option<u64> = None;
-            loop {
-                match metrics_rx.changed().await {
-                    Ok(_) => {
-                        let mm = metrics_rx.borrow().clone();
-
-                        if let Some(current_leader) = mm.current_leader {
-                            if last_leader != Some(current_leader) && last_leader.is_some() {
-                                info!(
-                                    "The leader transition has occurred. The current leader is Node {}. Previous leader was Node {}.",
-                                    current_leader, last_leader.unwrap()
-                                );
-                                if mm.id == current_leader {
-                                    info!("The current node transition to the Leader, starts controller thread.");
-                                    // TODO Start the control thread
-                                }
-                            }
-                            last_leader = Some(current_leader);
-                        }
-                    }
-                    Err(changed_err) => {
-                        error!(
-                        "Error while watching metrics_rx: {}; quitting monitoring_leader_transition() loop",
-                        changed_err);
-                    }
-                }
-            }
-        });
+        monitoring_leader_transition(
+            &raft,
+            self.rocksdb_engine_handler.clone(),
+            self.cluster_cache.clone(),
+            self.mqtt_cache.clone(),
+            self.engine_cache.clone(),
+            self.client_pool.clone(),
+            raft_machine_apply,
+        );
     }
 
     // Start HTTP Server
@@ -237,8 +219,7 @@ impl PlacementCenter {
         });
     }
 
-    // Start Storage Engine Cluster Controller
-    pub fn start_controller(
+    pub fn start_heartbeat(
         &self,
         raft_machine_apply: Arc<RaftMachineApply>,
         stop_send: Sender<bool>,
@@ -254,27 +235,6 @@ impl PlacementCenter {
         tokio::spawn(async move {
             ctrl.start_node_heartbeat_check().await;
         });
-
-        let mqtt_controller = MqttController::new(
-            self.rocksdb_engine_handler.clone(),
-            self.cluster_cache.clone(),
-            self.mqtt_cache.clone(),
-            self.client_pool.clone(),
-            stop_send.clone(),
-        );
-        tokio::spawn(async move {
-            mqtt_controller.start().await;
-        });
-
-        let journal_controller = StorageEngineController::new(
-            raft_machine_apply.clone(),
-            self.engine_cache.clone(),
-            self.cluster_cache.clone(),
-            self.client_pool.clone(),
-        );
-        tokio::spawn(async move {
-            journal_controller.start().await;
-        });
     }
 
     // Start Raft Status Machine
diff --git a/src/placement-center/src/mqtt/controller/mod.rs b/src/placement-center/src/mqtt/controller/mod.rs
index 147fa70e2..d8d1340c6 100644
--- a/src/placement-center/src/mqtt/controller/mod.rs
+++ b/src/placement-center/src/mqtt/controller/mod.rs
@@ -17,6 +17,7 @@ use std::time::Duration;
 
 use dashmap::DashMap;
 use grpc_clients::pool::ClientPool;
+use log::info;
 use message_expire::MessageExpire;
 use session_expire::SessionExpire;
 use tokio::select;
@@ -65,6 +66,7 @@ impl MqttController {
                 val = stop_recv.recv() =>{
                     if let Ok(flag) = val {
                         if flag {
+                            info!("Mqtt controller thread stopped successfully");
                             break;
                         }
                     }
diff --git a/src/placement-center/src/mqtt/controller/session_expire.rs b/src/placement-center/src/mqtt/controller/session_expire.rs
index a147c334a..32c4f04b9 100644
--- a/src/placement-center/src/mqtt/controller/session_expire.rs
+++ b/src/placement-center/src/mqtt/controller/session_expire.rs
@@ -287,7 +287,7 @@ pub async fn send_last_will(
     let node_addr = placement_cache_manager.get_broker_node_addr_by_cluster(&cluster_name);
 
     if node_addr.is_empty() {
-        warn!("Get cluster {} Node access address is empty, there is no cluster node address available.",cluster_name);
+        debug!("Get cluster {} Node access address is empty, there is no cluster node address available.",cluster_name);
         mqtt_cache_manager.remove_expire_last_will(&cluster_name, &client_id);
         return;
     }
diff --git a/src/placement-center/src/raft/leadership.rs b/src/placement-center/src/raft/leadership.rs
new file mode 100644
index 000000000..89d9484ee
--- /dev/null
+++ b/src/placement-center/src/raft/leadership.rs
@@ -0,0 +1,122 @@
+// 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::sync::Arc;
+
+use crate::{
+    core::cache::PlacementCacheManager,
+    journal::{cache::JournalCacheManager, controller::StorageEngineController},
+    mqtt::{cache::MqttCacheManager, controller::MqttController},
+    route::apply::RaftMachineApply,
+};
+
+use super::typeconfig::TypeConfig;
+use grpc_clients::pool::ClientPool;
+use log::{error, info};
+use openraft::Raft;
+use rocksdb_engine::RocksDBEngine;
+use tokio::sync::broadcast::{self, Sender};
+
+pub fn monitoring_leader_transition(
+    raft: &Raft<TypeConfig>,
+    rocksdb_engine_handler: Arc<RocksDBEngine>,
+    cluster_cache: Arc<PlacementCacheManager>,
+    mqtt_cache: Arc<MqttCacheManager>,
+    engine_cache: Arc<JournalCacheManager>,
+    client_pool: Arc<ClientPool>,
+    raft_machine_apply: Arc<RaftMachineApply>,
+) {
+    let mut metrics_rx = raft.metrics();
+    let (stop_send, _) = broadcast::channel::<bool>(2);
+    let mut controller_running = false;
+    tokio::spawn(async move {
+        let mut last_leader: Option<u64> = None;
+        loop {
+            match metrics_rx.changed().await {
+                Ok(_) => {
+                    let mm = metrics_rx.borrow().clone();
+                    if let Some(current_leader) = mm.current_leader {
+                        if last_leader != Some(current_leader) {
+                            if mm.id == current_leader {
+                                info!(
+                                    "Leader transition has occurred. current leader is Node {:?}. Previous leader was Node {:?}.",
+                                    mm.current_leader, last_leader
+                                );
+                                start_controller(
+                                    &rocksdb_engine_handler,
+                                    &cluster_cache,
+                                    &mqtt_cache,
+                                    &engine_cache,
+                                    &client_pool,
+                                    &raft_machine_apply,
+                                    stop_send.clone(),
+                                );
+                                controller_running = true;
+                            } else if controller_running {
+                                stop_controller(stop_send.clone());
+                                controller_running = false
+                            }
+                            last_leader = Some(current_leader);
+                        }
+                    }
+                }
+                Err(changed_err) => {
+                    error!(
+                    "Error while watching metrics_rx: {}; quitting monitoring_leader_transition() loop",
+                    changed_err);
+                }
+            }
+        }
+    });
+}
+
+pub fn start_controller(
+    rocksdb_engine_handler: &Arc<RocksDBEngine>,
+    cluster_cache: &Arc<PlacementCacheManager>,
+    mqtt_cache: &Arc<MqttCacheManager>,
+    engine_cache: &Arc<JournalCacheManager>,
+    client_pool: &Arc<ClientPool>,
+    raft_machine_apply: &Arc<RaftMachineApply>,
+    stop_send: Sender<bool>,
+) {
+    let mqtt_controller = MqttController::new(
+        rocksdb_engine_handler.clone(),
+        cluster_cache.clone(),
+        mqtt_cache.clone(),
+        client_pool.clone(),
+        stop_send.clone(),
+    );
+    tokio::spawn(async move {
+        mqtt_controller.start().await;
+    });
+
+    let journal_controller = StorageEngineController::new(
+        raft_machine_apply.clone(),
+        engine_cache.clone(),
+        cluster_cache.clone(),
+        client_pool.clone(),
+    );
+    tokio::spawn(async move {
+        journal_controller.start().await;
+    });
+}
+
+pub fn stop_controller(stop_send: Sender<bool>) {
+    if let Err(e) = stop_send.send(true) {
+        error!(
+            "Failed to send stop signal, Failure to stop controller,Error message:{}",
+            e
+        );
+    }
+}
diff --git a/src/placement-center/src/raft/mod.rs b/src/placement-center/src/raft/mod.rs
index a2f607908..fba4b4768 100644
--- a/src/placement-center/src/raft/mod.rs
+++ b/src/placement-center/src/raft/mod.rs
@@ -13,6 +13,7 @@
 // limitations under the License.
 
 pub mod error;
+pub mod leadership;
 #[allow(clippy::module_inception)]
 pub mod network;
 pub mod raft_node;
diff --git a/src/storage-adapter/src/lib.rs b/src/storage-adapter/src/lib.rs
index 50ff97a2d..675258a03 100644
--- a/src/storage-adapter/src/lib.rs
+++ b/src/storage-adapter/src/lib.rs
@@ -19,6 +19,7 @@ pub mod memory;
 pub mod minio;
 pub mod mysql;
 pub mod rocksdb;
+pub mod s3;
 pub mod storage;
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
diff --git a/src/storage-adapter/src/s3/mod.rs b/src/storage-adapter/src/s3/mod.rs
new file mode 100644
index 000000000..88041165f
--- /dev/null
+++ b/src/storage-adapter/src/s3/mod.rs
@@ -0,0 +1,13 @@
+// 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.


Test patch: