[ISSUE #4057]Broker epoch cache handler (#4056)

* 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

---------

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

Author: watchgou

rocketmq-broker/src/broker_runtime.rs

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

rocketmq-broker/src/controller/replicas_manager.rs

@@ -1,3 +1,4 @@
+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 +16,6 @@
  * limitations under the License.
  */
 use tracing::warn;
-
 #[derive(Default)]
 pub struct ReplicasManager {}
 
@@ -27,4 +27,8 @@ impl ReplicasManager {
     pub fn shutdown(&mut self) {
         warn!("ReplicasManager shutdown not implemented");
     }
+
+    pub fn get_epoch_entries(&self) -> Vec<EpochEntry> {
+        unimplemented!("")
+    }
 }

rocketmq-broker/src/processor/admin_broker_processor.rs

@@ -28,6 +28,7 @@ use tracing::warn;
 use crate::broker_runtime::BrokerRuntimeInner;
 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_min_broker_id_handler::NotifyMinBrokerChangeIdHandler;
 use crate::processor::admin_broker_processor::offset_request_handler::OffsetRequestHandler;
@@ -38,6 +39,7 @@ use crate::processor::admin_broker_processor::update_broker_ha_handler::UpdateBr
 
 mod batch_mq_handler;
 mod broker_config_request_handler;
+mod broker_epoch_cache_handler;
 mod consumer_request_handler;
 mod notify_min_broker_id_handler;
 mod offset_request_handler;
@@ -59,6 +61,7 @@ pub struct AdminBrokerProcessor<MS: MessageStore> {
     notify_min_broker_handler: NotifyMinBrokerChangeIdHandler<MS>,
     update_broker_ha_handler: UpdateBrokerHaHandler<MS>,
     reset_master_flusg_offset_handler: ResetMasterFlushOffsetHandler<MS>,
+    broker_epoch_cache_handler: BrokerEpochCacheHandler<MS>,
 }
 
 impl<MS> RequestProcessor for AdminBrokerProcessor<MS>
@@ -97,6 +100,8 @@ impl<MS: MessageStore> AdminBrokerProcessor<MS> {
         let reset_master_flusg_offset_handler =
             ResetMasterFlushOffsetHandler::new(broker_runtime_inner.clone());
 
+        let broker_epoch_cache_handler = BrokerEpochCacheHandler::new(broker_runtime_inner.clone());
+
         AdminBrokerProcessor {
             topic_request_handler,
             broker_config_request_handler,
@@ -108,6 +113,7 @@ impl<MS: MessageStore> AdminBrokerProcessor<MS> {
             notify_min_broker_handler,
             update_broker_ha_handler,
             reset_master_flusg_offset_handler,
+            broker_epoch_cache_handler,
         }
     }
 }
@@ -249,6 +255,11 @@ impl<MS: MessageStore> AdminBrokerProcessor<MS> {
                     .reset_master_flush_offset(channel, ctx, request_code, request)
                     .await
             }
+            RequestCode::GetBrokerEpochCache => {
+                self.broker_epoch_cache_handler
+                    .get_broker_epoch_cache(channel, ctx, request_code, request)
+                    .await
+            }
             _ => Some(get_unknown_cmd_response(request_code)),
         }
     }

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

@@ -0,0 +1,90 @@
+/*
+ * 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::epoch_entry_cache::EpochEntryCache;
+use rocketmq_remoting::protocol::remoting_command::RemotingCommand;
+use rocketmq_remoting::protocol::RemotingSerializable;
+use rocketmq_remoting::runtime::connection_handler_context::ConnectionHandlerContext;
+use rocketmq_rust::ArcMut;
+use rocketmq_store::base::message_store::MessageStore;
+
+use crate::broker_runtime::BrokerRuntimeInner;
+
+#[derive(Clone)]
+pub struct BrokerEpochCacheHandler<MS: MessageStore> {
+    broker_runtime_inner: ArcMut<BrokerRuntimeInner<MS>>,
+}
+
+impl<MS: MessageStore> BrokerEpochCacheHandler<MS> {
+    pub fn new(broker_runtime_inner: ArcMut<BrokerRuntimeInner<MS>>) -> Self {
+        Self {
+            broker_runtime_inner,
+        }
+    }
+
+    pub async fn get_broker_epoch_cache(
+        &mut self,
+        _channel: Channel,
+        _ctx: ConnectionHandlerContext,
+        _request_code: RequestCode,
+        _request: &mut RemotingCommand,
+    ) -> Option<RemotingCommand> {
+        let broker_runtime_inner = self.broker_runtime_inner.as_mut();
+
+        let replicas_manage = if let Some(replicas_manage) = broker_runtime_inner.replicas_manager()
+        {
+            replicas_manage
+        } else {
+            panic!("`replicas_manage` object is empty")
+        };
+
+        let broker_config = broker_runtime_inner.broker_config();
+        let response = RemotingCommand::create_response_command();
+
+        if !broker_config.enable_controller_mode {
+            return Some(
+                response
+                    .set_code(ResponseCode::SystemError)
+                    .set_remark("this request only for controllerMode"),
+            );
+        }
+
+        let broker_identity = &broker_config.broker_identity;
+        let broker_cluster_name = &broker_identity.broker_cluster_name;
+        let broker_name = broker_config.broker_name();
+        let broker_id = broker_identity.broker_id;
+
+        let epoch_list = replicas_manage.get_epoch_entries();
+
+        let message_store = broker_runtime_inner.message_store().as_ref().unwrap();
+        let max_offset = message_store.get_max_phy_offset() as u64;
+
+        let entry_code = EpochEntryCache::new(
+            broker_cluster_name,
+            broker_name,
+            broker_id,
+            epoch_list,
+            max_offset,
+        );
+
+        let cache = entry_code.encode().unwrap_or_default();
+        Some(response.set_body(cache).set_code(ResponseCode::Success))
+    }
+}

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

@@ -89,9 +89,8 @@ impl<MS: MessageStore> NotifyMinBrokerChangeIdHandler<MS> {
 
         self.update_min_broker(change_header).await;
 
-        let mut response = RemotingCommand::default();
-        response.set_code_ref(ResponseCode::Success);
-        Some(response)
+        let response = RemotingCommand::create_response_command();
+        Some(response.set_code(ResponseCode::Success))
     }
 
     async fn update_min_broker(&mut self, change_header: NotifyMinBrokerIdChangeRequestHeader) {

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

@@ -15,7 +15,6 @@
  * limitations under the License.
  */
 
-use cheetah_string::CheetahString;
 use rocketmq_common::common::mix_all::MASTER_ID;
 use rocketmq_remoting::code::request_code::RequestCode;
 use rocketmq_remoting::code::response_code::ResponseCode;
@@ -47,7 +46,7 @@ impl<MS: MessageStore> ResetMasterFlushOffsetHandler<MS> {
         _request_code: RequestCode,
         request: &mut RemotingCommand,
     ) -> Option<RemotingCommand> {
-        let mut response = RemotingCommand::default();
+        let response = RemotingCommand::create_response_command();
 
         let broker_id = self
             .broker_runtime_inner
@@ -66,8 +65,6 @@ impl<MS: MessageStore> ResetMasterFlushOffsetHandler<MS> {
             }
         }
 
-        response.set_code_ref(ResponseCode::Success);
-        response.set_remark_mut(CheetahString::empty());
-        Some(response)
+        Some(response.set_code(ResponseCode::Success))
     }
 }