Reset master offset (#4043)

Author: watchgou

rocketmq-broker/src/processor/admin_broker_processor.rs

@@ -31,6 +31,7 @@ use crate::processor::admin_broker_processor::broker_config_request_handler::Bro
 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;
+use crate::processor::admin_broker_processor::reset_master_flusg_offset_handler::ResetMasterFlushOffsetHandler;
 use crate::processor::admin_broker_processor::subscription_group_handler::SubscriptionGroupHandler;
 use crate::processor::admin_broker_processor::topic_request_handler::TopicRequestHandler;
 use crate::processor::admin_broker_processor::update_broker_ha_handler::UpdateBrokerHaHandler;
@@ -40,6 +41,7 @@ mod broker_config_request_handler;
 mod consumer_request_handler;
 mod notify_min_broker_id_handler;
 mod offset_request_handler;
+mod reset_master_flusg_offset_handler;
 mod subscription_group_handler;
 mod topic_request_handler;
 mod update_broker_ha_handler;
@@ -56,6 +58,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>,
 }
 
 impl<MS> RequestProcessor for AdminBrokerProcessor<MS>
@@ -91,6 +94,9 @@ impl<MS: MessageStore> AdminBrokerProcessor<MS> {
 
         let update_broker_ha_handler = UpdateBrokerHaHandler::new(broker_runtime_inner.clone());
 
+        let reset_master_flusg_offset_handler =
+            ResetMasterFlushOffsetHandler::new(broker_runtime_inner.clone());
+
         AdminBrokerProcessor {
             topic_request_handler,
             broker_config_request_handler,
@@ -101,6 +107,7 @@ impl<MS: MessageStore> AdminBrokerProcessor<MS> {
             broker_runtime_inner,
             notify_min_broker_handler,
             update_broker_ha_handler,
+            reset_master_flusg_offset_handler,
         }
     }
 }
@@ -237,6 +244,11 @@ impl<MS: MessageStore> AdminBrokerProcessor<MS> {
                     .update_broker_ha_info(channel, ctx, request_code, request)
                     .await
             }
+            RequestCode::ResetMasterFlushOffset => {
+                self.reset_master_flusg_offset_handler
+                    .reset_master_flush_offset(channel, ctx, request_code, request)
+                    .await
+            }
             _ => Some(get_unknown_cmd_response(request_code)),
         }
     }

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

@@ -0,0 +1,73 @@
+/*
+ * 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 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;
+use rocketmq_remoting::net::channel::Channel;
+use rocketmq_remoting::protocol::header::reset_master_flush_offset_header::ResetMasterFlushOffsetHeader;
+use rocketmq_remoting::protocol::remoting_command::RemotingCommand;
+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 ResetMasterFlushOffsetHandler<MS: MessageStore> {
+    broker_runtime_inner: ArcMut<BrokerRuntimeInner<MS>>,
+}
+
+impl<MS: MessageStore> ResetMasterFlushOffsetHandler<MS> {
+    pub fn new(broker_runtime_inner: ArcMut<BrokerRuntimeInner<MS>>) -> Self {
+        Self {
+            broker_runtime_inner,
+        }
+    }
+
+    pub async fn reset_master_flush_offset(
+        &mut self,
+        _channel: Channel,
+        _ctx: ConnectionHandlerContext,
+        _request_code: RequestCode,
+        request: &mut RemotingCommand,
+    ) -> Option<RemotingCommand> {
+        let mut response = RemotingCommand::default();
+
+        let broker_id = self
+            .broker_runtime_inner
+            .broker_config()
+            .broker_identity
+            .broker_id;
+        if broker_id != MASTER_ID {
+            let request_header = request
+                .decode_command_custom_header::<ResetMasterFlushOffsetHeader>()
+                .unwrap();
+
+            if let Some(maset_flush_offset) = request_header.master_flush_offset {
+                if let Some(message_store) = self.broker_runtime_inner.message_store() {
+                    message_store.set_master_flushed_offset(maset_flush_offset);
+                }
+            }
+        }
+
+        response.set_code_ref(ResponseCode::Success);
+        response.set_remark_mut(CheetahString::empty());
+        Some(response)
+    }
+}

rocketmq-remoting/src/protocol/header.rs

@@ -66,6 +66,7 @@ pub mod query_subscription_by_consumer_request_header;
 pub mod query_topic_consume_by_who_request_header;
 pub mod query_topics_by_consumer_request_header;
 pub mod reply_message_request_header;
+pub mod reset_master_flush_offset_header;
 pub mod reset_offset_request_header;
 pub mod search_offset_response_header;
 pub mod unlock_batch_mq_request_header;

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

@@ -0,0 +1,49 @@
+/*
+ * 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_macros::RequestHeaderCodec;
+use serde::Deserialize;
+use serde::Serialize;
+
+#[derive(Clone, Debug, Serialize, Deserialize, RequestHeaderCodec)]
+#[serde(rename_all = "camelCase")]
+pub struct ResetMasterFlushOffsetHeader {
+    pub master_flush_offset: Option<i64>,
+}
+
+#[cfg(test)]
+mod tests {
+    use crate::protocol::header::reset_master_flush_offset_header::ResetMasterFlushOffsetHeader;
+
+    #[test]
+    fn reset_master_flush_offset_header_serializes_correctly() {
+        let header = ResetMasterFlushOffsetHeader {
+            master_flush_offset: Some(4231),
+        };
+
+        let serialized = serde_json::to_string(&header).unwrap();
+        let expected = r#"{"masterFlushOffset":4231}"#;
+        assert_eq!(serialized, expected);
+    }
+
+    #[test]
+    fn reset_master_flush_offset_header_deserializes_correctly() {
+        let data = r#"{"masterFlushOffset":9527}"#;
+        let header: ResetMasterFlushOffsetHeader = serde_json::from_str(data).unwrap();
+        assert_eq!(header.master_flush_offset, Some(9527));
+    }
+}