[ISSUE ##1052] Support request code LOCK_BATCH_MQ(41)

Author: mxsm

rocketmq-broker/src/out_api/broker_outer_api.rs

@@ -31,7 +31,6 @@ use rocketmq_remoting::code::request_code::RequestCode;
 use rocketmq_remoting::code::response_code::ResponseCode;
 use rocketmq_remoting::protocol::body::broker_body::register_broker_body::RegisterBrokerBody;
 use rocketmq_remoting::protocol::body::kv_table::KVTable;
-use rocketmq_remoting::protocol::body::request::lock_batch_request_body::LockBatchRequestBody;
 use rocketmq_remoting::protocol::body::response::lock_batch_response_body::LockBatchResponseBody;
 use rocketmq_remoting::protocol::body::topic_info_wrapper::topic_config_wrapper::TopicConfigAndMappingSerializeWrapper;
 use rocketmq_remoting::protocol::header::lock_batch_mq_request_header::LockBatchMqRequestHeader;
@@ -351,7 +350,7 @@ impl BrokerOuterAPI {
                     Ok(lock_batch_response_body.lock_ok_mq_set)
                 } else {
                     Err(BrokerError::MQBrokerError(
-                        response.code() as i32,
+                        response.code(),
                         response.remark().cloned().unwrap_or("".to_string()),
                         "".to_string(),
                     ))

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

@@ -23,6 +23,7 @@ use bytes::Bytes;
 use rocketmq_remoting::code::request_code::RequestCode;
 use rocketmq_remoting::net::channel::Channel;
 use rocketmq_remoting::protocol::body::request::lock_batch_request_body::LockBatchRequestBody;
+use rocketmq_remoting::protocol::body::response::lock_batch_response_body::LockBatchResponseBody;
 use rocketmq_remoting::protocol::remoting_command::RemotingCommand;
 use rocketmq_remoting::protocol::RemotingDeserializable;
 use rocketmq_remoting::protocol::RemotingSerializable;
@@ -74,18 +75,17 @@ impl BatchMqHandler {
                 addr_map.extend(self.inner.broker_member_group.broker_addrs.clone());
                 addr_map.remove(&self.inner.broker_config.broker_identity.broker_id);
 
-                let mut tasks = vec![];
-
                 let count_down_latch = CountDownLatch::new(addr_map.len() as u32);
                 let request_body = Bytes::from(request_body.encode());
                 let mq_lock_map_arc = Arc::new(Mutex::new(mq_lock_map.clone()));
                 for (_, broker_addr) in addr_map {
                     let count_down_latch = count_down_latch.clone();
                     let broker_outer_api = self.inner.broker_out_api.clone();
                     let mq_lock_map = mq_lock_map_arc.clone();
-                    tasks.push(tokio::spawn(async move {
+                    let request_body_cloned = request_body.clone();
+                    tokio::spawn(async move {
                         let result = broker_outer_api
-                            .lock_batch_mq_async(broker_addr, request_body.clone(), 1000)
+                            .lock_batch_mq_async(broker_addr, request_body_cloned, 1000)
                             .await;
                         match result {
                             Ok(lock_ok_mqs) => {
@@ -99,7 +99,7 @@ impl BatchMqHandler {
                             }
                         }
                         count_down_latch.count_down().await;
-                    }));
+                    });
                 }
                 count_down_latch.wait_timeout(Duration::from_secs(2)).await;
 
@@ -111,15 +111,18 @@ impl BatchMqHandler {
                 }
             }
         }
-        unimplemented!("lock_natch_mq")
+        let response_body = LockBatchResponseBody {
+            lock_ok_mq_set: lock_ok_mqset,
+        };
+        Some(RemotingCommand::create_response_command().set_body(Some(response_body.encode())))
     }
 
     pub async fn unlock_batch_mq(
         &mut self,
         _channel: Channel,
         _ctx: ConnectionHandlerContext,
         _request_code: RequestCode,
-        request: RemotingCommand,
+        _request: RemotingCommand,
     ) -> Option<RemotingCommand> {
         unimplemented!("unlockBatchMQ")
     }

rocketmq-namesrv/src/route/route_info_manager.rs

@@ -652,7 +652,9 @@ impl RouteInfoManager {
             BrokerMemberGroup::new(cluster_name.to_string(), broker_name.to_string());
         if let Some(broker_data) = self.broker_addr_table.get(broker_name) {
             let map = broker_data.broker_addrs().clone();
-            group_member.broker_addrs.extend(map);
+            for (key, value) in map {
+                group_member.broker_addrs.insert(key as u64, value);
+            }
         }
         Some(group_member)
     }