[ISSUE #4059]Notify broker role changed(RequstCode:1008) (#4058)

* add notify min broker id change

* add notify min broker id change

* Broker id Change function external logic implementation

* Broker id Change function external logic implementation

* Broker id Change function external logic implementation

* Implementation Notification Broker Id Change Logic

* Implementation Notification Broker Id Change Logic

* Implementation Notification Broker Id Change Logic

* Broker id Change function external logic implementation

* Broker id Change function external logic implementation

* Broker id Change function external logic implementation

* Broker id Change function external logic implementation

* update broker ha info

* Broker id Change function external logic implementation

* update broker ha info

* update logic

* update logic

* reset master flush offset

* broker epoch cache handler

* broker epoch cache handler

* back ffi

* Comment out entry encode

* Comment out entry encode

* add object encode

* update unwrap function

* back ffi push

* notify broker role changed

* Delete the imported module

* Optimization logic

---------

Co-authored-by: jon <zjon234@gmail.cm>

Author: watchgou

rocketmq-broker/src/broker_runtime.rs

@@ -1848,6 +1848,11 @@ impl<MS: MessageStore> BrokerRuntimeInner<MS> {
         &self.replicas_manager
     }
 
+    #[inline]
+    pub fn replicas_manager_mut(&mut self) -> &mut Option<ReplicasManager> {
+        &mut self.replicas_manager
+    }
+
     #[inline]
     pub fn store_host(&self) -> SocketAddr {
         self.store_host

rocketmq-broker/src/controller/replicas_manager.rs

@@ -1,4 +1,3 @@
-use rocketmq_remoting::protocol::body::epoch_entry_cache::EpochEntry;
 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
@@ -15,7 +14,13 @@ use rocketmq_remoting::protocol::body::epoch_entry_cache::EpochEntry;
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
+use std::collections::HashSet;
+
+use cheetah_string::CheetahString;
+use rocketmq_remoting::protocol::body::epoch_entry_cache::EpochEntry;
 use tracing::warn;
+
 #[derive(Default)]
 pub struct ReplicasManager {}
 
@@ -31,4 +36,15 @@ impl ReplicasManager {
     pub fn get_epoch_entries(&self) -> Vec<EpochEntry> {
         unimplemented!("")
     }
+
+    pub async fn change_broker_role(
+        &mut self,
+        _new_master_broker_id: Option<u64>,
+        _new_master_address: Option<CheetahString>,
+        _new_master_epoch: Option<i32>,
+        _sync_state_set_epoch: Option<i32>,
+        _sync_state_set: Option<&HashSet<i64>>,
+    ) -> rocketmq_error::RocketMQResult<()> {
+        Ok(())
+    }
 }

rocketmq-broker/src/processor/admin_broker_processor.rs

@@ -30,6 +30,7 @@ use crate::processor::admin_broker_processor::batch_mq_handler::BatchMqHandler;
 use crate::processor::admin_broker_processor::broker_config_request_handler::BrokerConfigRequestHandler;
 use crate::processor::admin_broker_processor::broker_epoch_cache_handler::BrokerEpochCacheHandler;
 use crate::processor::admin_broker_processor::consumer_request_handler::ConsumerRequestHandler;
+use crate::processor::admin_broker_processor::notify_broker_role_change_handler::NotifyBrokerRoleChangeHandler;
 use crate::processor::admin_broker_processor::notify_min_broker_id_handler::NotifyMinBrokerChangeIdHandler;
 use crate::processor::admin_broker_processor::offset_request_handler::OffsetRequestHandler;
 use crate::processor::admin_broker_processor::reset_master_flusg_offset_handler::ResetMasterFlushOffsetHandler;
@@ -41,6 +42,7 @@ mod batch_mq_handler;
 mod broker_config_request_handler;
 mod broker_epoch_cache_handler;
 mod consumer_request_handler;
+mod notify_broker_role_change_handler;
 mod notify_min_broker_id_handler;
 mod offset_request_handler;
 mod reset_master_flusg_offset_handler;
@@ -62,6 +64,7 @@ pub struct AdminBrokerProcessor<MS: MessageStore> {
     update_broker_ha_handler: UpdateBrokerHaHandler<MS>,
     reset_master_flusg_offset_handler: ResetMasterFlushOffsetHandler<MS>,
     broker_epoch_cache_handler: BrokerEpochCacheHandler<MS>,
+    notify_broker_role_change_handler: NotifyBrokerRoleChangeHandler<MS>,
 }
 
 impl<MS> RequestProcessor for AdminBrokerProcessor<MS>
@@ -102,6 +105,9 @@ impl<MS: MessageStore> AdminBrokerProcessor<MS> {
 
         let broker_epoch_cache_handler = BrokerEpochCacheHandler::new(broker_runtime_inner.clone());
 
+        let notify_broker_role_change_handler =
+            NotifyBrokerRoleChangeHandler::new(broker_runtime_inner.clone());
+
         AdminBrokerProcessor {
             topic_request_handler,
             broker_config_request_handler,
@@ -114,6 +120,7 @@ impl<MS: MessageStore> AdminBrokerProcessor<MS> {
             update_broker_ha_handler,
             reset_master_flusg_offset_handler,
             broker_epoch_cache_handler,
+            notify_broker_role_change_handler,
         }
     }
 }
@@ -260,6 +267,11 @@ impl<MS: MessageStore> AdminBrokerProcessor<MS> {
                     .get_broker_epoch_cache(channel, ctx, request_code, request)
                     .await
             }
+            RequestCode::NotifyBrokerRoleChanged => {
+                self.notify_broker_role_change_handler
+                    .notify_broker_role_changed(channel, ctx, request_code, request)
+                    .await
+            }
             _ => Some(get_unknown_cmd_response(request_code)),
         }
     }

rocketmq-broker/src/processor/admin_broker_processor/notify_broker_role_change_handler.rs

@@ -0,0 +1,86 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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 rocketmq_remoting::code::request_code::RequestCode;
+use rocketmq_remoting::code::response_code::ResponseCode;
+use rocketmq_remoting::net::channel::Channel;
+use rocketmq_remoting::protocol::body::sync_state_set_body::SyncStateSet;
+use rocketmq_remoting::protocol::header::notify_broker_role_change_request_header::NotifyBrokerRoleChangedRequestHeader;
+use rocketmq_remoting::protocol::remoting_command::RemotingCommand;
+use rocketmq_remoting::protocol::RemotingDeserializable;
+use rocketmq_remoting::runtime::connection_handler_context::ConnectionHandlerContext;
+use rocketmq_rust::ArcMut;
+use rocketmq_store::base::message_store::MessageStore;
+use tracing::info;
+
+use crate::broker_runtime::BrokerRuntimeInner;
+
+#[derive(Clone)]
+pub struct NotifyBrokerRoleChangeHandler<MS: MessageStore> {
+    broker_runtime_inner: ArcMut<BrokerRuntimeInner<MS>>,
+}
+
+impl<MS: MessageStore> NotifyBrokerRoleChangeHandler<MS> {
+    pub fn new(broker_runtime_inner: ArcMut<BrokerRuntimeInner<MS>>) -> Self {
+        Self {
+            broker_runtime_inner,
+        }
+    }
+
+    pub async fn notify_broker_role_changed(
+        &mut self,
+        _channel: Channel,
+        _ctx: ConnectionHandlerContext,
+        _request_code: RequestCode,
+        request: &mut RemotingCommand,
+    ) -> Option<RemotingCommand> {
+        let request_header =
+            request.decode_command_custom_header::<NotifyBrokerRoleChangedRequestHeader>();
+
+        let sync_state_set_info =
+            SyncStateSet::decode(request.get_body().unwrap()).unwrap_or_default();
+
+        let response = RemotingCommand::create_response_command();
+
+        info!(
+            "Receive notifyBrokerRoleChanged request, try to change brokerRole, request:{}",
+            request_header.as_ref().expect("null")
+        );
+
+        if let Some(replicas_mangesr) = self.broker_runtime_inner.replicas_manager_mut() {
+            if let Ok(request_header) = request_header {
+                match replicas_mangesr
+                    .change_broker_role(
+                        request_header.master_broker_id,
+                        request_header.master_address,
+                        request_header.master_epoch,
+                        request_header.sync_state_set_epoch,
+                        sync_state_set_info.get_sync_state_set(),
+                    )
+                    .await
+                {
+                    Ok(_) => {}
+                    Err(e) => {
+                        panic!("Failed to call method change_broker_role: {}", e);
+                    }
+                }
+            }
+        }
+
+        Some(response.set_code(ResponseCode::Success))
+    }
+}

rocketmq-remoting/src/protocol/body.rs

@@ -56,6 +56,7 @@ pub mod request;
 pub mod response;
 pub mod set_message_request_mode_request_body;
 pub mod subscription_group_wrapper;
+pub mod sync_state_set_body;
 pub mod timer_metrics_serialize_wrapper;
 pub mod topic;
 pub mod topic_info_wrapper;

rocketmq-remoting/src/protocol/body/sync_state_set_body.rs

@@ -0,0 +1,38 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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::collections::HashSet;
+
+use serde::Deserialize;
+use serde::Serialize;
+
+#[derive(Debug, Serialize, Deserialize, Default)]
+#[serde(rename_all = "camelCase")]
+pub struct SyncStateSet {
+    sync_state_set: Option<HashSet<i64>>,
+    sync_state_set_epoch: i32,
+}
+
+impl SyncStateSet {
+    pub fn get_sync_state_set(&self) -> Option<&HashSet<i64>> {
+        self.sync_state_set.as_ref()
+    }
+
+    pub fn get_sync_state_set_epoch(&self) -> i32 {
+        self.sync_state_set_epoch
+    }
+}

rocketmq-remoting/src/protocol/header.rs

@@ -52,6 +52,7 @@ pub mod message_operation_header;
 pub mod namesrv;
 pub mod notification_request_header;
 pub mod notification_response_header;
+pub mod notify_broker_role_change_request_header;
 pub mod notify_consumer_ids_changed_request_header;
 pub mod pop_message_request_header;
 pub mod pop_message_response_header;

rocketmq-remoting/src/protocol/header/notify_broker_role_change_request_header.rs

@@ -0,0 +1,53 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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::fmt::Display;
+
+use cheetah_string::CheetahString;
+use rocketmq_macros::RequestHeaderCodec;
+use serde::Deserialize;
+use serde::Serialize;
+
+#[derive(Serialize, Deserialize, Debug, RequestHeaderCodec)]
+#[serde(rename_all = "camelCase")]
+pub struct NotifyBrokerRoleChangedRequestHeader {
+    #[required]
+    pub master_address: Option<CheetahString>,
+
+    #[required]
+    pub master_epoch: Option<i32>,
+
+    #[required]
+    pub sync_state_set_epoch: Option<i32>,
+
+    #[required]
+    pub master_broker_id: Option<u64>,
+}
+
+impl Display for NotifyBrokerRoleChangedRequestHeader {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        write!(
+            f,
+            "(master_address={:?}, master_epoch={:?}, sync_state_set_epoch={:?}, \
+             master_broker_id={:?})",
+            self.master_address,
+            self.master_epoch,
+            self.sync_state_set_epoch,
+            self.master_broker_id
+        )
+    }
+}