rocketmq_client_rust/implementation/mq_client_api_impl/
admin.rs

1// Copyright 2023 The RocketMQ Rust Authors
2//
3// Licensed under the Apache License, Version 2.0 (the "License");
4// you may not use this file except in compliance with the License.
5// You may obtain a copy of the License at
6//
7//     http://www.apache.org/licenses/LICENSE-2.0
8//
9// Unless required by applicable law or agreed to in writing, software
10// distributed under the License is distributed on an "AS IS" BASIS,
11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12// See the License for the specific language governing permissions and
13// limitations under the License.
14
15use super::request_builder::*;
16use super::response_decoder::*;
17use super::*;
18use rocketmq_error::fields;
19use rocketmq_error::Error;
20use rocketmq_error::ErrorContext;
21use rocketmq_model::topic::TopicConfig;
22use rocketmq_protocol::protocol::subscription::subscription_group_config::SubscriptionGroupConfig;
23use rocketmq_transport::api::OutboundRequestContractReason;
24use rocketmq_transport::api::OutboundRequestRejectionReason;
25use rocketmq_transport::api::OutboundRequestStage;
26use serde::de::DeserializeOwned;
27
28#[cfg(feature = "admin-full")]
29mod compatibility;
30#[cfg(feature = "admin-read")]
31mod controller_rollout;
32mod metadata;
33#[cfg(feature = "admin-mutation")]
34mod supervised_mutation;
35#[cfg(feature = "admin-mutation")]
36mod supervised_mutation_decode;
37mod versioned_config;
38
39fn admin_request_error(operation: &'static str, input: RetryInput) -> ClientError {
40    match input {
41        RetryInput::Transport(error) => ClientError::from_shared(error.into_shared_error()),
42        RetryInput::Rejected(rejection) => match rejection.reason() {
43            OutboundRequestRejectionReason::DeadlineExpired => {
44                let mut context = ErrorContext::new().with_text(fields::OPERATION_DIAGNOSTIC, operation);
45                if let Some(timeout_millis) = rejection.timeout_millis() {
46                    context = context.with_u64(fields::TIMEOUT_MS, timeout_millis);
47                }
48                ClientError::from_shared(Arc::new(
49                    Error::new(&rocketmq_error::CORE_OPERATION_TIMED_OUT).with_context(context),
50                ))
51            }
52            OutboundRequestRejectionReason::ClientStopping => ClientError::not_started(),
53            OutboundRequestRejectionReason::QueueSaturated => ClientError::from_shared(Arc::new(Error::new(
54                &rocketmq_error::TRANSPORT_ADMISSION_QUEUE_SATURATED,
55            ))),
56            OutboundRequestRejectionReason::Cancelled
57            | OutboundRequestRejectionReason::SessionClosed
58            | OutboundRequestRejectionReason::EndpointUnavailable => {
59                let phase = match rejection.reason() {
60                    OutboundRequestRejectionReason::SessionClosed => "closed",
61                    _ => admin_request_stage_phase(rejection.stage()),
62                };
63                let mut context = ErrorContext::new().with_text(fields::PHASE, phase);
64                if rejection.remote_addr_present() {
65                    context = context.with_secret_presence(fields::REMOTE_ADDR_PRESENT);
66                }
67                ClientError::from_shared(Arc::new(
68                    Error::new(&rocketmq_error::TRANSPORT_CONNECTION_FAILED).with_context(context),
69                ))
70            }
71        },
72        RetryInput::Contract(contract) => match contract.reason() {
73            OutboundRequestContractReason::NameServerEndpointMissing => {
74                let mut context = ErrorContext::new().with_text(fields::PHASE, "connect");
75                if contract.remote_addr_present() {
76                    context = context.with_secret_presence(fields::REMOTE_ADDR_PRESENT);
77                }
78                ClientError::from_shared(Arc::new(
79                    Error::new(&rocketmq_error::TRANSPORT_CONNECTION_FAILED).with_context(context),
80                ))
81            }
82        },
83        RetryInput::Response { terminal_error, .. } | RetryInput::BusinessError(terminal_error) => terminal_error,
84        // Admin request attempts cannot construct producer-only send status or route-policy inputs.
85        RetryInput::SendStatus(_) | RetryInput::RouteUnavailable => {
86            unreachable!("admin request returned an unrelated retry input")
87        }
88    }
89}
90
91const fn admin_request_stage_phase(stage: OutboundRequestStage) -> &'static str {
92    match stage {
93        OutboundRequestStage::BeforeWrite => "before_write",
94        OutboundRequestStage::Writing => "writing",
95        OutboundRequestStage::AwaitingResponse => "awaiting_response",
96        OutboundRequestStage::ResponseReceived => "response_received",
97    }
98}
99
100fn decode_admin_json<T: DeserializeOwned>(bytes: &[u8]) -> ClientResult<T> {
101    serde_json::from_slice(bytes).map_err(|source| {
102        let context = ErrorContext::new()
103            .with_text(fields::OPERATION_DIAGNOSTIC, "deserialize")
104            .with_text(fields::FORMAT, "JSON")
105            .with_secret_presence(fields::SOURCE_PRESENT);
106        ClientError::from_error(
107            Error::caused_by(&rocketmq_error::CORE_SERIALIZATION_FAILED, source).with_context(context),
108        )
109    })
110}
111
112pub struct AdminClient<'a> {
113    api: &'a MQClientAPIImpl,
114}
115
116impl AdminClient<'_> {
117    pub async fn broker_cluster_info(&self, timeout_millis: u64) -> ClientResult<ClusterInfo> {
118        self.api.get_broker_cluster_info(timeout_millis).await
119    }
120}
121
122impl MQClientAPIImpl {
123    #[must_use]
124    pub fn admin_client(&self) -> AdminClient<'_> {
125        AdminClient { api: self }
126    }
127}
128
129impl MQClientAPIImpl {
130    pub(crate) async fn get_kvconfig_value(
131        &self,
132        namespace: CheetahString,
133        key: CheetahString,
134        timeout_millis: u64,
135    ) -> ClientResult<Option<CheetahString>> {
136        let request_header = GetKVConfigRequestHeader::new(namespace, key);
137        let request = self.create_request_command(RequestCode::GetKvConfig, request_header);
138
139        let name_server_address_list = self.remoting_client.get_name_server_address_list();
140        let max_attempts = u32::try_from(name_server_address_list.len()).unwrap_or(u32::MAX);
141        let deadline = RequestDeadline::from_timeout_millis(timeout_millis);
142        let mut last_error = None;
143        for (index, name_srv_addr) in name_server_address_list.into_iter().enumerate() {
144            if deadline.is_expired() {
145                if let Some(error) = last_error.take() {
146                    return Err(error);
147                }
148            }
149            let attempt = u32::try_from(index).unwrap_or(u32::MAX).saturating_add(1);
150            let outcome = self
151                .remoting_client
152                .invoke_request_with_deadline(Some(&name_srv_addr), request.clone(), deadline)
153                .await;
154            let input = match outcome {
155                Ok(OutboundRequestOutcome::Response(response)) => match ResponseCode::from(response.code()) {
156                    ResponseCode::Success => {
157                        let response_header = response
158                            .decode_command_custom_header::<GetKVConfigResponseHeader>()
159                            .map_err(|error| {
160                                ClientError::response_process_source("decode GetKVConfigResponseHeader", error)
161                            })?;
162                        return Ok(response_header.value);
163                    }
164                    ResponseCode::QueryNotFound => return Ok(None),
165                    _ => {
166                        let code = response.code();
167                        RetryInput::Response {
168                            code,
169                            retry_after: None,
170                            terminal_error: mq_client_err!(
171                                code,
172                                response.remark().map_or("".to_string(), |s| s.to_string())
173                            ),
174                        }
175                    }
176                },
177                Ok(OutboundRequestOutcome::Rejected(rejection)) => RetryInput::Rejected(rejection),
178                Ok(OutboundRequestOutcome::Contract(contract)) => RetryInput::Contract(contract),
179                Err(error) => RetryInput::Transport(error),
180            };
181            let action = RetryPolicy::decide(
182                RetryContext {
183                    operation: RetryOperation::NameServerKv,
184                    idempotency: RetryIdempotency::Idempotent,
185                    attempt,
186                    max_attempts,
187                    remaining: deadline.remaining(),
188                    producer_retry_response_codes: None,
189                    retry_not_store_ok: false,
190                },
191                &input,
192            );
193            let terminal_error = admin_request_error("get_kvconfig_value", input);
194            match action {
195                RetryAction::Stop => return Err(terminal_error),
196                RetryAction::RetryAfter(delay) => {
197                    if tokio::time::timeout_at(deadline.instant(), tokio::time::sleep(delay))
198                        .await
199                        .is_err()
200                    {
201                        return Err(terminal_error);
202                    }
203                    last_error = Some(terminal_error);
204                }
205                RetryAction::RetryNow | RetryAction::SwitchBroker => {
206                    last_error = Some(terminal_error);
207                }
208                RetryAction::RefreshRoute | RetryAction::RefreshLeader => {
209                    return Err(terminal_error);
210                }
211            }
212            if deadline.is_expired() {
213                if let Some(error) = last_error.take() {
214                    return Err(error);
215                }
216            }
217        }
218
219        match last_error {
220            Some(error) => Err(error),
221            None => Ok(None),
222        }
223    }
224
225    pub(crate) async fn get_kv_config_value(
226        &self,
227        namespace: CheetahString,
228        key: CheetahString,
229        timeout_millis: u64,
230    ) -> ClientResult<Option<CheetahString>> {
231        self.get_kvconfig_value(namespace, key, timeout_millis).await
232    }
233
234    #[cfg(feature = "admin-mutation")]
235    pub(crate) async fn delete_kvconfig_value(
236        &self,
237        namespace: CheetahString,
238        key: CheetahString,
239        timeout_millis: u64,
240    ) -> ClientResult<()> {
241        let request_header = DeleteKVConfigRequestHeader::new(namespace, key);
242        let request = self.create_request_command(RequestCode::DeleteKvConfig, request_header);
243
244        let name_server_address_list = self.remoting_client.get_name_server_address_list();
245        let mut err_response = None;
246        for name_srv_addr in name_server_address_list {
247            let outcome = self
248                .remoting_client
249                .invoke_request(Some(&name_srv_addr), request.clone(), timeout_millis)
250                .await;
251            let response = match outcome {
252                Ok(OutboundRequestOutcome::Response(response)) => response,
253                Ok(OutboundRequestOutcome::Rejected(rejection)) => {
254                    return Err(admin_request_error(
255                        "delete_kvconfig_value",
256                        RetryInput::Rejected(rejection),
257                    ));
258                }
259                Ok(OutboundRequestOutcome::Contract(contract)) => {
260                    return Err(admin_request_error(
261                        "delete_kvconfig_value",
262                        RetryInput::Contract(contract),
263                    ));
264                }
265                Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
266            };
267            match ResponseCode::from(response.code()) {
268                ResponseCode::Success => {}
269                _ => err_response = Some(response),
270            }
271        }
272
273        if let Some(err_response) = err_response {
274            return Err(mq_client_err!(
275                err_response.code(),
276                err_response.remark().map_or("".to_string(), |s| s.to_string())
277            ));
278        }
279        Ok(())
280    }
281
282    #[cfg(feature = "admin-mutation")]
283    pub(crate) async fn delete_kv_config_value(
284        &self,
285        namespace: CheetahString,
286        key: CheetahString,
287        timeout_millis: u64,
288    ) -> ClientResult<()> {
289        self.delete_kvconfig_value(namespace, key, timeout_millis).await
290    }
291
292    pub(crate) async fn get_kvlist_by_namespace(
293        &self,
294        namespace: CheetahString,
295        timeout_millis: u64,
296    ) -> ClientResult<rocketmq_protocol::protocol::body::kv_table::KVTable> {
297        let request_header = GetKVListByNamespaceRequestHeader::new(namespace);
298        let request = self.create_request_command(RequestCode::GetKvlistByNamespace, request_header);
299
300        let outcome = self.remoting_client.invoke_request(None, request, timeout_millis).await;
301        let response = match outcome {
302            Ok(OutboundRequestOutcome::Response(response)) => response,
303            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
304                return Err(admin_request_error(
305                    "get_kvlist_by_namespace",
306                    RetryInput::Rejected(rejection),
307                ));
308            }
309            Ok(OutboundRequestOutcome::Contract(contract)) => {
310                return Err(admin_request_error(
311                    "get_kvlist_by_namespace",
312                    RetryInput::Contract(contract),
313                ));
314            }
315            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
316        };
317        if ResponseCode::from(response.code()) == ResponseCode::Success {
318            if let Some(body) = response.get_body() {
319                return rocketmq_protocol::protocol::body::kv_table::KVTable::decode(body.as_ref())
320                    .map_err(ClientError::from);
321            }
322        }
323
324        Err(mq_client_err!(
325            response.code(),
326            response.remark().map_or("".to_string(), |s| s.to_string())
327        ))
328    }
329
330    pub(crate) async fn get_kv_list_by_namespace(
331        &self,
332        namespace: CheetahString,
333        timeout_millis: u64,
334    ) -> ClientResult<rocketmq_protocol::protocol::body::kv_table::KVTable> {
335        self.get_kvlist_by_namespace(namespace, timeout_millis).await
336    }
337
338    #[cfg(feature = "admin-mutation")]
339    pub(crate) async fn put_kvconfig_value(
340        &self,
341        namespace: CheetahString,
342        key: CheetahString,
343        value: CheetahString,
344        timeout_millis: u64,
345    ) -> ClientResult<()> {
346        let request_header = PutKVConfigRequestHeader::new(namespace, key, value);
347        let request = self.create_request_command(RequestCode::PutKvConfig, request_header);
348
349        let name_server_address_list = self.remoting_client.get_name_server_address_list();
350        let mut err_response = None;
351        for name_srv_addr in name_server_address_list {
352            let outcome = self
353                .remoting_client
354                .invoke_request(Some(&name_srv_addr), request.clone(), timeout_millis)
355                .await;
356            let response = match outcome {
357                Ok(OutboundRequestOutcome::Response(response)) => response,
358                Ok(OutboundRequestOutcome::Rejected(rejection)) => {
359                    return Err(admin_request_error(
360                        "put_kvconfig_value",
361                        RetryInput::Rejected(rejection),
362                    ));
363                }
364                Ok(OutboundRequestOutcome::Contract(contract)) => {
365                    return Err(admin_request_error(
366                        "put_kvconfig_value",
367                        RetryInput::Contract(contract),
368                    ));
369                }
370                Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
371            };
372            match ResponseCode::from(response.code()) {
373                ResponseCode::Success => {}
374                _ => err_response = Some(response),
375            }
376        }
377
378        if let Some(err_response) = err_response {
379            return Err(mq_client_err!(
380                err_response.code(),
381                err_response.remark().map_or("".to_string(), |s| s.to_string())
382            ));
383        }
384        Ok(())
385    }
386
387    #[cfg(feature = "admin-mutation")]
388    pub(crate) async fn put_kv_config_value(
389        &self,
390        namespace: CheetahString,
391        key: CheetahString,
392        value: CheetahString,
393        timeout_millis: u64,
394    ) -> ClientResult<()> {
395        self.put_kvconfig_value(namespace, key, value, timeout_millis).await
396    }
397
398    #[cfg(feature = "admin-mutation")]
399    pub(crate) async fn create_user(
400        &self,
401        broker_address: CheetahString,
402        user_info: &UserInfo,
403        timeout_millis: u64,
404    ) -> ClientResult<()> {
405        let mut request_header = CreateUserRequestHeader::default();
406        let username = user_info
407            .username
408            .clone()
409            .ok_or_else(|| mq_client_err!(-1, "username is required".to_string()))?;
410        request_header.set_username(username);
411        let mut request = self.create_request_command(RequestCode::AuthCreateUser, request_header);
412        request = request.set_body(user_info.encode()?);
413
414        let outcome = self
415            .remoting_client
416            .invoke_request(Some(&broker_address), request.clone(), timeout_millis)
417            .await;
418        let response = match outcome {
419            Ok(OutboundRequestOutcome::Response(response)) => response,
420            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
421                return Err(admin_request_error("create_user", RetryInput::Rejected(rejection)));
422            }
423            Ok(OutboundRequestOutcome::Contract(contract)) => {
424                return Err(admin_request_error("create_user", RetryInput::Contract(contract)));
425            }
426            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
427        };
428
429        let mut err_response = None;
430        match ResponseCode::from(response.code()) {
431            ResponseCode::Success => {}
432            _ => err_response = Some(response),
433        }
434
435        if let Some(err_response) = err_response {
436            return Err(mq_client_err!(
437                err_response.code(),
438                err_response.remark().map_or("".to_string(), |s| s.to_string())
439            ));
440        }
441        Ok(())
442    }
443
444    #[cfg(feature = "admin-mutation")]
445    pub(crate) async fn update_user(
446        &self,
447        broker_address: CheetahString,
448        user_info: &UserInfo,
449        timeout_millis: u64,
450    ) -> ClientResult<()> {
451        let mut request_header = UpdateUserRequestHeader::default();
452        let username = user_info
453            .username
454            .clone()
455            .ok_or_else(|| mq_client_err!(-1, "username is required".to_string()))?;
456        request_header.set_username(username);
457        let mut request = self.create_request_command(RequestCode::AuthUpdateUser, request_header);
458        request = request.set_body(user_info.encode()?);
459
460        let outcome = self
461            .remoting_client
462            .invoke_request(Some(&broker_address), request.clone(), timeout_millis)
463            .await;
464        let response = match outcome {
465            Ok(OutboundRequestOutcome::Response(response)) => response,
466            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
467                return Err(admin_request_error("update_user", RetryInput::Rejected(rejection)));
468            }
469            Ok(OutboundRequestOutcome::Contract(contract)) => {
470                return Err(admin_request_error("update_user", RetryInput::Contract(contract)));
471            }
472            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
473        };
474
475        let mut err_response = None;
476        match ResponseCode::from(response.code()) {
477            ResponseCode::Success => {}
478            _ => err_response = Some(response),
479        }
480
481        if let Some(err_response) = err_response {
482            return Err(mq_client_err!(
483                err_response.code(),
484                err_response.remark().map_or("".to_string(), |s| s.to_string())
485            ));
486        }
487        Ok(())
488    }
489
490    #[cfg(feature = "admin-mutation")]
491    pub(crate) async fn create_acl(
492        &self,
493        broker_address: CheetahString,
494        acl_info: &AclInfo,
495        timeout_millis: u64,
496    ) -> ClientResult<()> {
497        let subject = acl_info
498            .subject
499            .clone()
500            .ok_or_else(|| mq_client_err!(-1, "ACL subject is required".to_string()))?;
501        let request_header = CreateAclRequestHeader { subject };
502        let request = self
503            .create_request_command(RequestCode::AuthCreateAcl, request_header)
504            .set_body(acl_info.encode()?);
505
506        let outcome = self
507            .remoting_client
508            .invoke_request(Some(&broker_address), request, timeout_millis)
509            .await;
510        let response = match outcome {
511            Ok(OutboundRequestOutcome::Response(response)) => response,
512            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
513                return Err(admin_request_error("create_acl", RetryInput::Rejected(rejection)));
514            }
515            Ok(OutboundRequestOutcome::Contract(contract)) => {
516                return Err(admin_request_error("create_acl", RetryInput::Contract(contract)));
517            }
518            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
519        };
520
521        match ResponseCode::from(response.code()) {
522            ResponseCode::Success => Ok(()),
523            _ => Err(mq_client_err!(
524                response.code(),
525                response.remark().map_or("".to_string(), |s| s.to_string())
526            )),
527        }
528    }
529
530    #[cfg(feature = "admin-mutation")]
531    pub(crate) async fn update_acl(
532        &self,
533        broker_address: CheetahString,
534        acl_info: &AclInfo,
535        timeout_millis: u64,
536    ) -> ClientResult<()> {
537        let subject = acl_info
538            .subject
539            .clone()
540            .ok_or_else(|| mq_client_err!(-1, "ACL subject is required".to_string()))?;
541        let request_header = UpdateAclRequestHeader { subject };
542        let request = self
543            .create_request_command(RequestCode::AuthUpdateAcl, request_header)
544            .set_body(acl_info.encode()?);
545
546        let outcome = self
547            .remoting_client
548            .invoke_request(Some(&broker_address), request, timeout_millis)
549            .await;
550        let response = match outcome {
551            Ok(OutboundRequestOutcome::Response(response)) => response,
552            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
553                return Err(admin_request_error("update_acl", RetryInput::Rejected(rejection)));
554            }
555            Ok(OutboundRequestOutcome::Contract(contract)) => {
556                return Err(admin_request_error("update_acl", RetryInput::Contract(contract)));
557            }
558            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
559        };
560
561        match ResponseCode::from(response.code()) {
562            ResponseCode::Success => Ok(()),
563            _ => Err(mq_client_err!(
564                response.code(),
565                response.remark().map_or("".to_string(), |s| s.to_string())
566            )),
567        }
568    }
569
570    #[cfg(feature = "admin-mutation")]
571    pub(crate) async fn create_and_update_plain_access_config(
572        &self,
573        broker_address: CheetahString,
574        plain_access_config: &PlainAccessConfig,
575        timeout_millis: u64,
576    ) -> ClientResult<()> {
577        let request = create_and_update_plain_access_config_request(&self.command_factory, plain_access_config)?;
578
579        let outcome = self
580            .remoting_client
581            .invoke_request(Some(&broker_address), request, timeout_millis)
582            .await;
583        let response = match outcome {
584            Ok(OutboundRequestOutcome::Response(response)) => response,
585            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
586                return Err(admin_request_error(
587                    "create_and_update_plain_access_config",
588                    RetryInput::Rejected(rejection),
589                ));
590            }
591            Ok(OutboundRequestOutcome::Contract(contract)) => {
592                return Err(admin_request_error(
593                    "create_and_update_plain_access_config",
594                    RetryInput::Contract(contract),
595                ));
596            }
597            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
598        };
599
600        match ResponseCode::from(response.code()) {
601            ResponseCode::Success => Ok(()),
602            _ => Err(mq_client_err!(
603                response.code(),
604                response.remark().map_or_else(String::new, ToString::to_string)
605            )),
606        }
607    }
608
609    #[cfg(feature = "admin-mutation")]
610    pub(crate) async fn delete_plain_access_config(
611        &self,
612        broker_address: CheetahString,
613        access_key: CheetahString,
614        timeout_millis: u64,
615    ) -> ClientResult<()> {
616        let request = delete_plain_access_config_request(&self.command_factory, &access_key);
617
618        let outcome = self
619            .remoting_client
620            .invoke_request(Some(&broker_address), request, timeout_millis)
621            .await;
622        let response = match outcome {
623            Ok(OutboundRequestOutcome::Response(response)) => response,
624            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
625                return Err(admin_request_error(
626                    "delete_plain_access_config",
627                    RetryInput::Rejected(rejection),
628                ));
629            }
630            Ok(OutboundRequestOutcome::Contract(contract)) => {
631                return Err(admin_request_error(
632                    "delete_plain_access_config",
633                    RetryInput::Contract(contract),
634                ));
635            }
636            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
637        };
638
639        match ResponseCode::from(response.code()) {
640            ResponseCode::Success => Ok(()),
641            _ => Err(mq_client_err!(
642                response.code(),
643                response.remark().map_or_else(String::new, ToString::to_string)
644            )),
645        }
646    }
647
648    #[cfg(feature = "admin-mutation")]
649    pub(crate) async fn update_global_white_addrs_config(
650        &self,
651        broker_address: CheetahString,
652        global_white_addrs: CheetahString,
653        timeout_millis: u64,
654    ) -> ClientResult<()> {
655        let request_header = UpdateGlobalWhiteAddrsConfigRequestHeader { global_white_addrs };
656        let request = self.create_request_command(RequestCode::UpdateGlobalWhiteAddrsConfig, request_header);
657
658        let outcome = self
659            .remoting_client
660            .invoke_request(Some(&broker_address), request, timeout_millis)
661            .await;
662        let response = match outcome {
663            Ok(OutboundRequestOutcome::Response(response)) => response,
664            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
665                return Err(admin_request_error(
666                    "update_global_white_addrs_config",
667                    RetryInput::Rejected(rejection),
668                ));
669            }
670            Ok(OutboundRequestOutcome::Contract(contract)) => {
671                return Err(admin_request_error(
672                    "update_global_white_addrs_config",
673                    RetryInput::Contract(contract),
674                ));
675            }
676            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
677        };
678
679        match ResponseCode::from(response.code()) {
680            ResponseCode::Success => Ok(()),
681            _ => Err(mq_client_err!(
682                response.code(),
683                response.remark().map_or("".to_string(), |s| s.to_string())
684            )),
685        }
686    }
687
688    pub(crate) async fn get_broker_cluster_acl_version_info(
689        &self,
690        broker_address: CheetahString,
691        timeout_millis: u64,
692    ) -> ClientResult<ClusterAclVersionInfo> {
693        let request = self.create_remoting_command(RequestCode::GetBrokerClusterAclInfo);
694
695        let outcome = self
696            .remoting_client
697            .invoke_request(Some(&broker_address), request, timeout_millis)
698            .await;
699        let response = match outcome {
700            Ok(OutboundRequestOutcome::Response(response)) => response,
701            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
702                return Err(admin_request_error(
703                    "get_broker_cluster_acl_version_info",
704                    RetryInput::Rejected(rejection),
705                ));
706            }
707            Ok(OutboundRequestOutcome::Contract(contract)) => {
708                return Err(admin_request_error(
709                    "get_broker_cluster_acl_version_info",
710                    RetryInput::Contract(contract),
711                ));
712            }
713            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
714        };
715
716        match ResponseCode::from(response.code()) {
717            ResponseCode::Success => decode_cluster_acl_version_info_response_body(response.body()),
718            _ => Err(mq_client_err!(
719                response.code(),
720                response.remark().map_or_else(String::new, |remark| remark.to_string())
721            )),
722        }
723    }
724
725    #[cfg(feature = "admin-mutation")]
726    pub(crate) async fn delete_acl(
727        &self,
728        broker_address: CheetahString,
729        subject: CheetahString,
730        resource: CheetahString,
731        timeout_millis: u64,
732    ) -> ClientResult<()> {
733        let resource_option = if resource.is_empty() { None } else { Some(resource) };
734        let request_header = DeleteAclRequestHeader::new(subject, resource_option);
735        self.delete_acl_request(broker_address, request_header, timeout_millis)
736            .await
737    }
738
739    #[cfg(feature = "admin-mutation")]
740    pub(crate) async fn delete_acl_entry(
741        &self,
742        broker_address: CheetahString,
743        subject: CheetahString,
744        policy_type: CheetahString,
745        resource: CheetahString,
746        timeout_millis: u64,
747    ) -> ClientResult<()> {
748        let header = acl_entry_delete_header(subject, policy_type, resource)?;
749        self.delete_acl_request(broker_address, header, timeout_millis).await
750    }
751
752    #[cfg(feature = "admin-mutation")]
753    async fn delete_acl_request(
754        &self,
755        broker_address: CheetahString,
756        request_header: DeleteAclRequestHeader,
757        timeout_millis: u64,
758    ) -> ClientResult<()> {
759        let request = self.create_request_command(RequestCode::AuthDeleteAcl, request_header);
760
761        let outcome = self
762            .remoting_client
763            .invoke_request(Some(&broker_address), request, timeout_millis)
764            .await;
765        let response = match outcome {
766            Ok(OutboundRequestOutcome::Response(response)) => response,
767            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
768                return Err(admin_request_error("delete_acl", RetryInput::Rejected(rejection)));
769            }
770            Ok(OutboundRequestOutcome::Contract(contract)) => {
771                return Err(admin_request_error("delete_acl", RetryInput::Contract(contract)));
772            }
773            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
774        };
775
776        match ResponseCode::from(response.code()) {
777            ResponseCode::Success => Ok(()),
778            _ => Err(mq_client_err!(
779                response.code(),
780                response.remark().map_or("".to_string(), |s| s.to_string())
781            )),
782        }
783    }
784
785    pub(crate) async fn list_acl(
786        &self,
787        broker_address: CheetahString,
788        subject_filter: CheetahString,
789        resource_filter: CheetahString,
790        timeout_millis: u64,
791    ) -> ClientResult<Vec<AclInfo>> {
792        let request_header = ListAclRequestHeader {
793            subject_filter,
794            resource_filter,
795        };
796        let request = self.create_request_command(RequestCode::AuthListAcl, request_header);
797
798        let outcome = self
799            .remoting_client
800            .invoke_request(Some(&broker_address), request, timeout_millis)
801            .await;
802        let response = match outcome {
803            Ok(OutboundRequestOutcome::Response(response)) => response,
804            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
805                return Err(admin_request_error("list_acl", RetryInput::Rejected(rejection)));
806            }
807            Ok(OutboundRequestOutcome::Contract(contract)) => {
808                return Err(admin_request_error("list_acl", RetryInput::Contract(contract)));
809            }
810            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
811        };
812
813        match ResponseCode::from(response.code()) {
814            ResponseCode::Success => {
815                if let Some(body) = response.get_body() {
816                    Vec::<AclInfo>::decode(body.as_ref()).map_err(ClientError::from)
817                } else {
818                    Ok(Vec::new())
819                }
820            }
821            _ => Err(mq_client_err!(
822                response.code(),
823                response.remark().map_or("".to_string(), |s| s.to_string())
824            )),
825        }
826    }
827
828    pub(crate) async fn get_acl(
829        &self,
830        broker_address: CheetahString,
831        subject: CheetahString,
832        timeout_millis: u64,
833    ) -> ClientResult<AclInfo> {
834        let request = get_acl_request(&self.command_factory, subject);
835        let outcome = self
836            .remoting_client
837            .invoke_request(Some(&broker_address), request, timeout_millis)
838            .await;
839        let response = match outcome {
840            Ok(OutboundRequestOutcome::Response(response)) => response,
841            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
842                return Err(admin_request_error("get_acl", RetryInput::Rejected(rejection)));
843            }
844            Ok(OutboundRequestOutcome::Contract(contract)) => {
845                return Err(admin_request_error("get_acl", RetryInput::Contract(contract)));
846            }
847            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
848        };
849
850        match ResponseCode::from(response.code()) {
851            ResponseCode::Success => {
852                let body = response
853                    .get_body()
854                    .ok_or_else(|| mq_client_err!("get_acl response body is empty".to_string()))?;
855                AclInfo::decode(body.as_ref()).map_err(ClientError::from)
856            }
857            _ => Err(mq_client_err!(
858                response.code(),
859                response.remark().map_or("".to_string(), |s| s.to_string())
860            )),
861        }
862    }
863
864    pub(crate) async fn get_user(
865        &self,
866        broker_address: CheetahString,
867        username: CheetahString,
868        timeout_millis: u64,
869    ) -> ClientResult<Option<UserInfo>> {
870        let request_header = GetUserRequestHeader {
871            username: username.clone(),
872        };
873        let request = self.create_request_command(RequestCode::AuthGetUser, request_header);
874
875        let outcome = self
876            .remoting_client
877            .invoke_request(Some(&broker_address), request, timeout_millis)
878            .await;
879        let response = match outcome {
880            Ok(OutboundRequestOutcome::Response(response)) => response,
881            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
882                return Err(admin_request_error("get_user", RetryInput::Rejected(rejection)));
883            }
884            Ok(OutboundRequestOutcome::Contract(contract)) => {
885                return Err(admin_request_error("get_user", RetryInput::Contract(contract)));
886            }
887            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
888        };
889
890        match ResponseCode::from(response.code()) {
891            ResponseCode::Success => {
892                let _body = response.get_body();
893                if let Some(body) = response.get_body() {
894                    let user_info = UserInfo::decode(body)?;
895                    Ok(Some(user_info))
896                } else {
897                    Ok(None)
898                }
899            }
900            _ => Err(mq_client_err!(
901                response.code(),
902                response.remark().map(|s| s.to_string()).unwrap_or_default()
903            )),
904        }
905    }
906
907    pub(crate) async fn list_users(
908        &self,
909        broker_address: CheetahString,
910        filter: CheetahString,
911        timeout_millis: u64,
912    ) -> ClientResult<Vec<UserInfo>> {
913        let request_header = ListUsersRequestHeader { filter };
914        let request = self.create_request_command(RequestCode::AuthListUsers, request_header);
915
916        let outcome = self
917            .remoting_client
918            .invoke_request(Some(&broker_address), request, timeout_millis)
919            .await;
920        let response = match outcome {
921            Ok(OutboundRequestOutcome::Response(response)) => response,
922            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
923                return Err(admin_request_error("list_users", RetryInput::Rejected(rejection)));
924            }
925            Ok(OutboundRequestOutcome::Contract(contract)) => {
926                return Err(admin_request_error("list_users", RetryInput::Contract(contract)));
927            }
928            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
929        };
930
931        match ResponseCode::from(response.code()) {
932            ResponseCode::Success => {
933                if let Some(body) = response.get_body() {
934                    Vec::<UserInfo>::decode(body.as_ref()).map_err(ClientError::from)
935                } else {
936                    Ok(Vec::new())
937                }
938            }
939            _ => Err(mq_client_err!(
940                response.code(),
941                response.remark().map_or("".to_string(), |s| s.to_string())
942            )),
943        }
944    }
945
946    pub(crate) async fn list_user(
947        &self,
948        broker_address: CheetahString,
949        filter: CheetahString,
950        timeout_millis: u64,
951    ) -> ClientResult<Vec<UserInfo>> {
952        self.list_users(broker_address, filter, timeout_millis).await
953    }
954
955    #[cfg(feature = "admin-mutation")]
956    pub(crate) async fn delete_user(
957        &self,
958        broker_address: CheetahString,
959        username: CheetahString,
960        timeout_millis: u64,
961    ) -> ClientResult<()> {
962        let mut request_header = DeleteUserRequestHeader::default();
963        request_header.set_username(username);
964        let request = self.create_request_command(RequestCode::AuthDeleteUser, request_header);
965
966        let outcome = self
967            .remoting_client
968            .invoke_request(Some(&broker_address), request.clone(), timeout_millis)
969            .await;
970        let response = match outcome {
971            Ok(OutboundRequestOutcome::Response(response)) => response,
972            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
973                return Err(admin_request_error("delete_user", RetryInput::Rejected(rejection)));
974            }
975            Ok(OutboundRequestOutcome::Contract(contract)) => {
976                return Err(admin_request_error("delete_user", RetryInput::Contract(contract)));
977            }
978            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
979        };
980
981        let mut err_response = None;
982        match ResponseCode::from(response.code()) {
983            ResponseCode::Success => {}
984            _ => err_response = Some(response),
985        }
986
987        if let Some(err_response) = err_response {
988            return Err(mq_client_err!(
989                err_response.code(),
990                err_response.remark().map_or("".to_string(), |s| s.to_string())
991            ));
992        }
993        Ok(())
994    }
995
996    #[cfg(feature = "admin-mutation")]
997    pub(crate) async fn update_name_server_config(
998        &self,
999        properties: HashMap<CheetahString, CheetahString>,
1000        special_name_servers: Option<Vec<CheetahString>>,
1001        timeout_millis: u64,
1002    ) -> ClientResult<()> {
1003        let body = mix_all::properties_to_string(&properties);
1004        if body.is_empty() {
1005            return Ok(());
1006        }
1007        let invoke_name_servers = if let Some(name_servers) = special_name_servers {
1008            if !name_servers.is_empty() {
1009                name_servers
1010            } else {
1011                self.get_name_server_address_list()
1012            }
1013        } else {
1014            self.get_name_server_address_list()
1015        };
1016        if invoke_name_servers.is_empty() {
1017            return Ok(());
1018        }
1019        let empty_header = EmptyHeader {};
1020        let mut request = self.create_request_command(RequestCode::UpdateNamesrvConfig, empty_header);
1021
1022        request = request.set_body(body.to_string());
1023        let mut err_response = None;
1024        for name_srv_addr in invoke_name_servers {
1025            let outcome = self
1026                .remoting_client
1027                .invoke_request(Some(&name_srv_addr), request.clone(), timeout_millis)
1028                .await;
1029            let response = match outcome {
1030                Ok(OutboundRequestOutcome::Response(response)) => response,
1031                Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1032                    return Err(admin_request_error(
1033                        "update_name_server_config",
1034                        RetryInput::Rejected(rejection),
1035                    ));
1036                }
1037                Ok(OutboundRequestOutcome::Contract(contract)) => {
1038                    return Err(admin_request_error(
1039                        "update_name_server_config",
1040                        RetryInput::Contract(contract),
1041                    ));
1042                }
1043                Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1044            };
1045            match ResponseCode::from(response.code()) {
1046                ResponseCode::Success => {}
1047                _ => err_response = Some(response),
1048            }
1049        }
1050
1051        if let Some(err_response) = err_response {
1052            return Err(mq_client_err!(
1053                err_response.code(),
1054                err_response.remark().map_or("".to_string(), |s| s.to_string())
1055            ));
1056        }
1057        Ok(())
1058    }
1059
1060    #[cfg(feature = "admin-mutation")]
1061    pub(crate) async fn add_write_perm_of_broker(
1062        &self,
1063        namesrv_addr: CheetahString,
1064        broker_name: CheetahString,
1065        timeout_millis: u64,
1066    ) -> ClientResult<i32> {
1067        let request_header = AddWritePermOfBrokerRequestHeader::new(broker_name);
1068        let request = self.create_request_command(RequestCode::AddWritePermOfBroker, request_header);
1069
1070        let outcome = self
1071            .remoting_client
1072            .invoke_request(Some(&namesrv_addr), request, timeout_millis)
1073            .await;
1074        let response = match outcome {
1075            Ok(OutboundRequestOutcome::Response(response)) => response,
1076            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1077                return Err(admin_request_error(
1078                    "add_write_perm_of_broker",
1079                    RetryInput::Rejected(rejection),
1080                ));
1081            }
1082            Ok(OutboundRequestOutcome::Contract(contract)) => {
1083                return Err(admin_request_error(
1084                    "add_write_perm_of_broker",
1085                    RetryInput::Contract(contract),
1086                ));
1087            }
1088            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1089        };
1090        if ResponseCode::from(response.code()) == ResponseCode::Success {
1091            let request_header = response.decode_command_custom_header_fast::<AddWritePermOfBrokerResponseHeader>()?;
1092            return Ok(request_header.get_add_topic_count());
1093        }
1094        Err(mq_client_err!(
1095            response.code(),
1096            response.remark().map_or("".to_string(), |s| s.to_string())
1097        ))
1098    }
1099
1100    #[cfg(feature = "admin-mutation")]
1101    pub(crate) async fn wipe_write_perm_of_broker(
1102        &self,
1103        namesrv_addr: CheetahString,
1104        broker_name: CheetahString,
1105        timeout_millis: u64,
1106    ) -> ClientResult<i32> {
1107        let request_header = WipeWritePermOfBrokerRequestHeader::new(broker_name);
1108        let request = self.create_request_command(RequestCode::WipeWritePermOfBroker, request_header);
1109
1110        let outcome = self
1111            .remoting_client
1112            .invoke_request(Some(&namesrv_addr), request, timeout_millis)
1113            .await;
1114        let response = match outcome {
1115            Ok(OutboundRequestOutcome::Response(response)) => response,
1116            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1117                return Err(admin_request_error(
1118                    "wipe_write_perm_of_broker",
1119                    RetryInput::Rejected(rejection),
1120                ));
1121            }
1122            Ok(OutboundRequestOutcome::Contract(contract)) => {
1123                return Err(admin_request_error(
1124                    "wipe_write_perm_of_broker",
1125                    RetryInput::Contract(contract),
1126                ));
1127            }
1128            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1129        };
1130        if ResponseCode::from(response.code()) == ResponseCode::Success {
1131            let request_header = response.decode_command_custom_header_fast::<WipeWritePermOfBrokerResponseHeader>()?;
1132            return Ok(request_header.get_wipe_topic_count());
1133        }
1134        Err(mq_client_err!(
1135            response.code(),
1136            response.remark().map_or("".to_string(), |s| s.to_string())
1137        ))
1138    }
1139
1140    pub(crate) async fn get_broker_cluster_info(&self, timeout_millis: u64) -> ClientResult<ClusterInfo> {
1141        let request = self.create_request_command(RequestCode::GetBrokerClusterInfo, EmptyHeader {});
1142        let outcome = self.remoting_client.invoke_request(None, request, timeout_millis).await;
1143        let response = match outcome {
1144            Ok(OutboundRequestOutcome::Response(response)) => response,
1145            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1146                return Err(admin_request_error(
1147                    "get_broker_cluster_info",
1148                    RetryInput::Rejected(rejection),
1149                ));
1150            }
1151            Ok(OutboundRequestOutcome::Contract(contract)) => {
1152                return Err(admin_request_error(
1153                    "get_broker_cluster_info",
1154                    RetryInput::Contract(contract),
1155                ));
1156            }
1157            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1158        };
1159        if ResponseCode::from(response.code()) == ResponseCode::Success {
1160            if let Some(body) = response.get_body() {
1161                return ClusterInfo::decode(body.as_ref()).map_err(ClientError::from);
1162            }
1163        }
1164        Err(mq_client_err!(
1165            response.code(),
1166            response.remark().map_or("".to_string(), |s| s.to_string())
1167        ))
1168    }
1169
1170    pub(crate) async fn get_broker_runtime_info(
1171        &self,
1172        addr: &CheetahString,
1173        timeout_millis: u64,
1174    ) -> ClientResult<rocketmq_protocol::protocol::body::kv_table::KVTable> {
1175        let request = self.create_request_command(RequestCode::GetBrokerRuntimeInfo, EmptyHeader {});
1176        let outcome = self
1177            .remoting_client
1178            .invoke_request(Some(addr), request, timeout_millis)
1179            .await;
1180        let response = match outcome {
1181            Ok(OutboundRequestOutcome::Response(response)) => response,
1182            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1183                return Err(admin_request_error(
1184                    "get_broker_runtime_info",
1185                    RetryInput::Rejected(rejection),
1186                ));
1187            }
1188            Ok(OutboundRequestOutcome::Contract(contract)) => {
1189                return Err(admin_request_error(
1190                    "get_broker_runtime_info",
1191                    RetryInput::Contract(contract),
1192                ));
1193            }
1194            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1195        };
1196        if ResponseCode::from(response.code()) == ResponseCode::Success {
1197            if let Some(body) = response.get_body() {
1198                return rocketmq_protocol::protocol::body::kv_table::KVTable::decode(body.as_ref())
1199                    .map_err(ClientError::from);
1200            }
1201        }
1202        Err(mq_client_err!(
1203            response.code(),
1204            response.remark().map_or("".to_string(), |s| s.to_string())
1205        ))
1206    }
1207
1208    #[cfg(feature = "admin-mutation")]
1209    pub(crate) async fn sync_broker_member_group(
1210        &self,
1211        cluster_name: &CheetahString,
1212        broker_name: &CheetahString,
1213        is_compatible_with_old_name_srv: bool,
1214    ) -> ClientResult<Option<BrokerMemberGroup>> {
1215        if is_compatible_with_old_name_srv {
1216            self.get_broker_member_group_compatible(cluster_name, broker_name).await
1217        } else {
1218            self.get_broker_member_group(cluster_name, broker_name).await
1219        }
1220    }
1221
1222    pub(super) async fn get_broker_member_group(
1223        &self,
1224        cluster_name: &CheetahString,
1225        broker_name: &CheetahString,
1226    ) -> ClientResult<Option<BrokerMemberGroup>> {
1227        let request_header = GetBrokerMemberGroupRequestHeader::new(cluster_name.clone(), broker_name.clone());
1228        let request = self.create_request_command(RequestCode::GetBrokerMemberGroup, request_header);
1229        let outcome = self.remoting_client.invoke_request(None, request, 3000).await;
1230        let mut response = match outcome {
1231            Ok(OutboundRequestOutcome::Response(response)) => response,
1232            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1233                return Err(admin_request_error(
1234                    "get_broker_member_group",
1235                    RetryInput::Rejected(rejection),
1236                ));
1237            }
1238            Ok(OutboundRequestOutcome::Contract(contract)) => {
1239                return Err(admin_request_error(
1240                    "get_broker_member_group",
1241                    RetryInput::Contract(contract),
1242                ));
1243            }
1244            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1245        };
1246        if ResponseCode::from(response.code()) == ResponseCode::Success {
1247            if let Some(body) = response.take_body() {
1248                let response_body = GetBrokerMemberGroupResponseBody::decode(body.as_ref())?;
1249                return Ok(Some(
1250                    response_body
1251                        .broker_member_group
1252                        .unwrap_or_else(|| empty_broker_member_group(cluster_name, broker_name)),
1253                ));
1254            }
1255        }
1256        Ok(Some(empty_broker_member_group(cluster_name, broker_name)))
1257    }
1258
1259    pub(super) async fn get_broker_member_group_compatible(
1260        &self,
1261        cluster_name: &CheetahString,
1262        broker_name: &CheetahString,
1263    ) -> ClientResult<Option<BrokerMemberGroup>> {
1264        let request_header = GetRouteInfoRequestHeader {
1265            topic: CheetahString::from_string(format!(
1266                "{}{}",
1267                TopicValidator::SYNC_BROKER_MEMBER_GROUP_PREFIX,
1268                broker_name
1269            )),
1270            ..Default::default()
1271        };
1272        let request = self.create_request_command(RequestCode::GetRouteinfoByTopic, request_header);
1273        let outcome = self.remoting_client.invoke_request(None, request, 3000).await;
1274        let mut response = match outcome {
1275            Ok(OutboundRequestOutcome::Response(response)) => response,
1276            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1277                return Err(admin_request_error(
1278                    "get_broker_member_group_compatible",
1279                    RetryInput::Rejected(rejection),
1280                ));
1281            }
1282            Ok(OutboundRequestOutcome::Contract(contract)) => {
1283                return Err(admin_request_error(
1284                    "get_broker_member_group_compatible",
1285                    RetryInput::Contract(contract),
1286                ));
1287            }
1288            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1289        };
1290        if ResponseCode::from(response.code()) == ResponseCode::Success {
1291            if let Some(body) = response.take_body() {
1292                let topic_route_data = TopicRouteData::decode(body.as_ref())?;
1293                return Ok(Some(broker_member_group_from_route_data(
1294                    cluster_name,
1295                    broker_name,
1296                    &topic_route_data,
1297                )));
1298            }
1299        }
1300        Ok(Some(empty_broker_member_group(cluster_name, broker_name)))
1301    }
1302
1303    pub(crate) async fn get_broker_lite_info(
1304        &self,
1305        addr: &CheetahString,
1306        timeout_millis: u64,
1307    ) -> ClientResult<GetBrokerLiteInfoResponseBody> {
1308        let request = self.create_request_command(RequestCode::GetBrokerLiteInfo, EmptyHeader {});
1309        let outcome = self
1310            .remoting_client
1311            .invoke_request(Some(addr), request, timeout_millis)
1312            .await;
1313        let response = match outcome {
1314            Ok(OutboundRequestOutcome::Response(response)) => response,
1315            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1316                return Err(admin_request_error(
1317                    "get_broker_lite_info",
1318                    RetryInput::Rejected(rejection),
1319                ));
1320            }
1321            Ok(OutboundRequestOutcome::Contract(contract)) => {
1322                return Err(admin_request_error(
1323                    "get_broker_lite_info",
1324                    RetryInput::Contract(contract),
1325                ));
1326            }
1327            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1328        };
1329        if ResponseCode::from(response.code()) == ResponseCode::Success {
1330            if let Some(body) = response.get_body() {
1331                return GetBrokerLiteInfoResponseBody::decode(body.as_ref()).map_err(ClientError::from);
1332            }
1333        }
1334        Err(mq_client_err!(
1335            response.code(),
1336            response.remark().map_or("".to_string(), |s| s.to_string())
1337        ))
1338    }
1339
1340    pub(crate) async fn check_rocksdb_cq_write_progress(
1341        &self,
1342        addr: &CheetahString,
1343        topic: CheetahString,
1344        check_store_time: i64,
1345        timeout_millis: u64,
1346    ) -> ClientResult<CheckRocksdbCqWriteResult> {
1347        let request_header = CheckRocksdbCqWriteProgressRequestHeader {
1348            topic,
1349            check_store_time,
1350            rpc: None,
1351        };
1352        let request = self.create_request_command(RequestCode::CheckRocksdbCqWriteProgress, request_header);
1353        let outcome = self
1354            .remoting_client
1355            .invoke_request(Some(addr), request, timeout_millis)
1356            .await;
1357        let response = match outcome {
1358            Ok(OutboundRequestOutcome::Response(response)) => response,
1359            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1360                return Err(admin_request_error(
1361                    "check_rocksdb_cq_write_progress",
1362                    RetryInput::Rejected(rejection),
1363                ));
1364            }
1365            Ok(OutboundRequestOutcome::Contract(contract)) => {
1366                return Err(admin_request_error(
1367                    "check_rocksdb_cq_write_progress",
1368                    RetryInput::Contract(contract),
1369                ));
1370            }
1371            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1372        };
1373        if ResponseCode::from(response.code()) == ResponseCode::Success {
1374            if let Some(body) = response.get_body() {
1375                let result: CheckRocksdbCqWriteResult = serde_json::from_slice(body.as_ref()).map_err(|e| {
1376                    mq_client_err!(-1, format!("Failed to deserialize CheckRocksdbCqWriteResult: {}", e))
1377                })?;
1378                return Ok(result);
1379            }
1380        }
1381        Err(mq_client_err!(
1382            response.code(),
1383            response.remark().map_or("".to_string(), |s| s.to_string())
1384        ))
1385    }
1386
1387    pub(crate) async fn query_consume_queue(
1388        &self,
1389        addr: &CheetahString,
1390        topic: CheetahString,
1391        queue_id: i32,
1392        index: i64,
1393        count: i32,
1394        consumer_group: CheetahString,
1395        timeout_millis: u64,
1396    ) -> ClientResult<QueryConsumeQueueResponseBody> {
1397        let request_header = QueryConsumeQueueRequestHeader {
1398            topic,
1399            queue_id,
1400            index,
1401            count,
1402            consumer_group: if consumer_group.is_empty() {
1403                None
1404            } else {
1405                Some(consumer_group)
1406            },
1407            topic_request_header: None,
1408        };
1409        let request = self.create_request_command(RequestCode::QueryConsumeQueue, request_header);
1410        let outcome = self
1411            .remoting_client
1412            .invoke_request(Some(addr), request, timeout_millis)
1413            .await;
1414        let response = match outcome {
1415            Ok(OutboundRequestOutcome::Response(response)) => response,
1416            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1417                return Err(admin_request_error(
1418                    "query_consume_queue",
1419                    RetryInput::Rejected(rejection),
1420                ));
1421            }
1422            Ok(OutboundRequestOutcome::Contract(contract)) => {
1423                return Err(admin_request_error(
1424                    "query_consume_queue",
1425                    RetryInput::Contract(contract),
1426                ));
1427            }
1428            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1429        };
1430        if ResponseCode::from(response.code()) == ResponseCode::Success {
1431            if let Some(body) = response.get_body() {
1432                let result: QueryConsumeQueueResponseBody = serde_json::from_slice(body.as_ref()).map_err(|e| {
1433                    mq_client_err!(
1434                        -1,
1435                        format!("Failed to deserialize QueryConsumeQueueResponseBody: {}", e)
1436                    )
1437                })?;
1438                return Ok(result);
1439            }
1440        }
1441        Err(mq_client_err!(
1442            response.code(),
1443            response.remark().map_or("".to_string(), |s| s.to_string())
1444        ))
1445    }
1446
1447    pub(crate) async fn get_lite_group_info(
1448        &self,
1449        addr: &CheetahString,
1450        group: CheetahString,
1451        lite_topic: CheetahString,
1452        top_k: i32,
1453        timeout_millis: u64,
1454    ) -> ClientResult<GetLiteGroupInfoResponseBody> {
1455        let request_header = GetLiteGroupInfoRequestHeader {
1456            group,
1457            lite_topic,
1458            top_k,
1459            rpc: None,
1460        };
1461        let request = self.create_request_command(RequestCode::GetLiteGroupInfo, request_header);
1462        let outcome = self
1463            .remoting_client
1464            .invoke_request(Some(addr), request, timeout_millis)
1465            .await;
1466        let response = match outcome {
1467            Ok(OutboundRequestOutcome::Response(response)) => response,
1468            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1469                return Err(admin_request_error(
1470                    "get_lite_group_info",
1471                    RetryInput::Rejected(rejection),
1472                ));
1473            }
1474            Ok(OutboundRequestOutcome::Contract(contract)) => {
1475                return Err(admin_request_error(
1476                    "get_lite_group_info",
1477                    RetryInput::Contract(contract),
1478                ));
1479            }
1480            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1481        };
1482        if ResponseCode::from(response.code()) == ResponseCode::Success {
1483            if let Some(body) = response.get_body() {
1484                return GetLiteGroupInfoResponseBody::decode(body.as_ref()).map_err(ClientError::from);
1485            }
1486        }
1487        Err(mq_client_err!(
1488            response.code(),
1489            response.remark().map_or("".to_string(), |s| s.to_string())
1490        ))
1491    }
1492
1493    pub(crate) async fn get_lite_client_info(
1494        &self,
1495        addr: &CheetahString,
1496        parent_topic: CheetahString,
1497        group: CheetahString,
1498        client_id: CheetahString,
1499        timeout_millis: u64,
1500    ) -> ClientResult<GetLiteClientInfoResponseBody> {
1501        let request_header = GetLiteClientInfoRequestHeader {
1502            parent_topic: Some(parent_topic),
1503            group: Some(group),
1504            client_id: Some(client_id),
1505            max_count: 1000,
1506        };
1507        let request = self.create_request_command(RequestCode::GetLiteClientInfo, request_header);
1508        let outcome = self
1509            .remoting_client
1510            .invoke_request(Some(addr), request, timeout_millis)
1511            .await;
1512        let response = match outcome {
1513            Ok(OutboundRequestOutcome::Response(response)) => response,
1514            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1515                return Err(admin_request_error(
1516                    "get_lite_client_info",
1517                    RetryInput::Rejected(rejection),
1518                ));
1519            }
1520            Ok(OutboundRequestOutcome::Contract(contract)) => {
1521                return Err(admin_request_error(
1522                    "get_lite_client_info",
1523                    RetryInput::Contract(contract),
1524                ));
1525            }
1526            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1527        };
1528        if ResponseCode::from(response.code()) == ResponseCode::Success {
1529            if let Some(body) = response.get_body() {
1530                return GetLiteClientInfoResponseBody::decode(body.as_ref()).map_err(ClientError::from);
1531            }
1532        }
1533        Err(mq_client_err!(
1534            response.code(),
1535            response.remark().map_or("".to_string(), |s| s.to_string())
1536        ))
1537    }
1538
1539    #[cfg(feature = "admin-mutation")]
1540    pub(crate) async fn trigger_lite_dispatch(
1541        &self,
1542        addr: &CheetahString,
1543        group: CheetahString,
1544        client_id: CheetahString,
1545        timeout_millis: u64,
1546    ) -> ClientResult<()> {
1547        let request_header = TriggerLiteDispatchRequestHeader {
1548            group,
1549            client_id: if client_id.is_empty() { None } else { Some(client_id) },
1550        };
1551        let request = self.create_request_command(RequestCode::TriggerLiteDispatch, request_header);
1552        let outcome = self
1553            .remoting_client
1554            .invoke_request(Some(addr), request, timeout_millis)
1555            .await;
1556        let response = match outcome {
1557            Ok(OutboundRequestOutcome::Response(response)) => response,
1558            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1559                return Err(admin_request_error(
1560                    "trigger_lite_dispatch",
1561                    RetryInput::Rejected(rejection),
1562                ));
1563            }
1564            Ok(OutboundRequestOutcome::Contract(contract)) => {
1565                return Err(admin_request_error(
1566                    "trigger_lite_dispatch",
1567                    RetryInput::Contract(contract),
1568                ));
1569            }
1570            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1571        };
1572        match ResponseCode::from(response.code()) {
1573            ResponseCode::Success => Ok(()),
1574            _ => Err(mq_client_err!(
1575                response.code(),
1576                response.remark().map_or("".to_string(), |s| s.to_string())
1577            )),
1578        }
1579    }
1580
1581    #[cfg(feature = "admin-mutation")]
1582    pub(crate) async fn sync_lite_subscription(
1583        &self,
1584        broker_addr: &CheetahString,
1585        lite_subscription_dto: LiteSubscriptionDTO,
1586        timeout_millis: u64,
1587    ) -> ClientResult<()> {
1588        let request = lite_subscription_ctl_request(&self.command_factory, lite_subscription_dto)?;
1589        let outcome = self
1590            .remoting_client
1591            .invoke_request(Some(broker_addr), request, timeout_millis)
1592            .await;
1593        let response = match outcome {
1594            Ok(OutboundRequestOutcome::Response(response)) => response,
1595            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1596                return Err(admin_request_error(
1597                    "sync_lite_subscription",
1598                    RetryInput::Rejected(rejection),
1599                ));
1600            }
1601            Ok(OutboundRequestOutcome::Contract(contract)) => {
1602                return Err(admin_request_error(
1603                    "sync_lite_subscription",
1604                    RetryInput::Contract(contract),
1605                ));
1606            }
1607            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1608        };
1609
1610        match ResponseCode::from(response.code()) {
1611            ResponseCode::Success => Ok(()),
1612            _ => Err(client_broker_err!(
1613                response.code(),
1614                response.remark().map_or_else(String::new, |remark| remark.to_string()),
1615                broker_addr
1616            )),
1617        }
1618    }
1619
1620    #[cfg(feature = "admin-mutation")]
1621    pub async fn sync_lite_subscription_async(
1622        &self,
1623        broker_addr: &CheetahString,
1624        lite_subscription_dto: LiteSubscriptionDTO,
1625        timeout_millis: u64,
1626    ) -> ClientResult<()> {
1627        self.sync_lite_subscription(broker_addr, lite_subscription_dto, timeout_millis)
1628            .await
1629    }
1630
1631    pub(crate) async fn get_lite_topic_info(
1632        &self,
1633        addr: &CheetahString,
1634        parent_topic: &CheetahString,
1635        lite_topic: &CheetahString,
1636        timeout_millis: u64,
1637    ) -> ClientResult<GetLiteTopicInfoResponseBody> {
1638        let request_header = GetLiteTopicInfoRequestHeader {
1639            parent_topic: parent_topic.clone(),
1640            lite_topic: lite_topic.clone(),
1641        };
1642        let request = self.create_request_command(RequestCode::GetLiteTopicInfo, request_header);
1643        let outcome = self
1644            .remoting_client
1645            .invoke_request(Some(addr), request, timeout_millis)
1646            .await;
1647        let response = match outcome {
1648            Ok(OutboundRequestOutcome::Response(response)) => response,
1649            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1650                return Err(admin_request_error(
1651                    "get_lite_topic_info",
1652                    RetryInput::Rejected(rejection),
1653                ));
1654            }
1655            Ok(OutboundRequestOutcome::Contract(contract)) => {
1656                return Err(admin_request_error(
1657                    "get_lite_topic_info",
1658                    RetryInput::Contract(contract),
1659                ));
1660            }
1661            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1662        };
1663        if ResponseCode::from(response.code()) == ResponseCode::Success {
1664            if let Some(body) = response.get_body() {
1665                return GetLiteTopicInfoResponseBody::decode(body.as_ref()).map_err(ClientError::from);
1666            }
1667        }
1668        Err(mq_client_err!(
1669            response.code(),
1670            response.remark().map_or("".to_string(), |s| s.to_string())
1671        ))
1672    }
1673    pub(crate) async fn get_parent_topic_info(
1674        &self,
1675        addr: &CheetahString,
1676        topic: CheetahString,
1677        timeout_millis: u64,
1678    ) -> ClientResult<GetParentTopicInfoResponseBody> {
1679        let request_header = GetParentTopicInfoRequestHeader { topic, rpc: None };
1680        let request = self.create_request_command(RequestCode::GetParentTopicInfo, request_header);
1681        let outcome = self
1682            .remoting_client
1683            .invoke_request(Some(addr), request, timeout_millis)
1684            .await;
1685        let response = match outcome {
1686            Ok(OutboundRequestOutcome::Response(response)) => response,
1687            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1688                return Err(admin_request_error(
1689                    "get_parent_topic_info",
1690                    RetryInput::Rejected(rejection),
1691                ));
1692            }
1693            Ok(OutboundRequestOutcome::Contract(contract)) => {
1694                return Err(admin_request_error(
1695                    "get_parent_topic_info",
1696                    RetryInput::Contract(contract),
1697                ));
1698            }
1699            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1700        };
1701        if ResponseCode::from(response.code()) == ResponseCode::Success {
1702            if let Some(body) = response.get_body() {
1703                return GetParentTopicInfoResponseBody::decode(body.as_ref()).map_err(ClientError::from);
1704            }
1705        }
1706        Err(mq_client_err!(
1707            response.code(),
1708            response.remark().map_or("".to_string(), |s| s.to_string())
1709        ))
1710    }
1711
1712    #[cfg(feature = "admin-mutation")]
1713    pub async fn delete_subscription_group(
1714        &self,
1715        addr: &CheetahString,
1716        group_name: CheetahString,
1717        clean_offset: bool,
1718        timeout_millis: u64,
1719    ) -> ClientResult<()> {
1720        let request_header = DeleteSubscriptionGroupRequestHeader {
1721            group_name,
1722            clean_offset,
1723            rpc_request_header: None,
1724        };
1725
1726        let request = self.create_request_command(RequestCode::DeleteSubscriptionGroup, request_header);
1727
1728        let outcome = self
1729            .remoting_client
1730            .invoke_request(Some(addr), request, timeout_millis)
1731            .await;
1732        let response = match outcome {
1733            Ok(OutboundRequestOutcome::Response(response)) => response,
1734            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1735                return Err(admin_request_error(
1736                    "delete_subscription_group",
1737                    RetryInput::Rejected(rejection),
1738                ));
1739            }
1740            Ok(OutboundRequestOutcome::Contract(contract)) => {
1741                return Err(admin_request_error(
1742                    "delete_subscription_group",
1743                    RetryInput::Contract(contract),
1744                ));
1745            }
1746            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1747        };
1748
1749        if ResponseCode::from(response.code()) == ResponseCode::Success {
1750            return Ok(());
1751        }
1752
1753        Err(mq_client_err!(
1754            response.code(),
1755            response.remark().map_or("".to_string(), |s| s.to_string())
1756        ))
1757    }
1758
1759    /// Deletes multiple subscription groups with one broker request.
1760    #[cfg(feature = "admin-mutation")]
1761    pub async fn delete_subscription_group_list(
1762        &self,
1763        addr: &CheetahString,
1764        group_name_list: Vec<CheetahString>,
1765        clean_offset: bool,
1766        timeout_millis: u64,
1767    ) -> ClientResult<()> {
1768        let request = delete_subscription_group_list_request(&self.command_factory, group_name_list, clean_offset)?;
1769        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
1770        let outcome = self
1771            .remoting_client
1772            .invoke_request(Some(&broker_addr), request, timeout_millis)
1773            .await;
1774        let response = match outcome {
1775            Ok(OutboundRequestOutcome::Response(response)) => response,
1776            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1777                return Err(admin_request_error(
1778                    "delete_subscription_group_list",
1779                    RetryInput::Rejected(rejection),
1780                ));
1781            }
1782            Ok(OutboundRequestOutcome::Contract(contract)) => {
1783                return Err(admin_request_error(
1784                    "delete_subscription_group_list",
1785                    RetryInput::Contract(contract),
1786                ));
1787            }
1788            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1789        };
1790
1791        match ResponseCode::from(response.code()) {
1792            ResponseCode::Success => Ok(()),
1793            _ => Err(mq_client_err!(
1794                response.code(),
1795                response.remark().map_or_else(String::new, |remark| remark.to_string())
1796            )),
1797        }
1798    }
1799
1800    #[cfg(feature = "admin-mutation")]
1801    pub async fn reset_master_flush_offset(
1802        &self,
1803        broker_addr: &CheetahString,
1804        master_flush_offset: i64,
1805    ) -> ClientResult<()> {
1806        let request_header = ResetMasterFlushOffsetHeader {
1807            master_flush_offset: Some(master_flush_offset),
1808        };
1809
1810        let request = self.create_request_command(RequestCode::ResetMasterFlushOffset, request_header);
1811
1812        let outcome = self
1813            .remoting_client
1814            .invoke_request(Some(broker_addr), request, 3000)
1815            .await;
1816        let response = match outcome {
1817            Ok(OutboundRequestOutcome::Response(response)) => response,
1818            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1819                return Err(admin_request_error(
1820                    "reset_master_flush_offset",
1821                    RetryInput::Rejected(rejection),
1822                ));
1823            }
1824            Ok(OutboundRequestOutcome::Contract(contract)) => {
1825                return Err(admin_request_error(
1826                    "reset_master_flush_offset",
1827                    RetryInput::Contract(contract),
1828                ));
1829            }
1830            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1831        };
1832
1833        if ResponseCode::from(response.code()) == ResponseCode::Success {
1834            return Ok(());
1835        }
1836
1837        Err(mq_client_err!(
1838            response.code(),
1839            response.remark().map_or("".to_string(), |s| s.to_string())
1840        ))
1841    }
1842
1843    pub(super) async fn get_topic_list_from_name_server_by_code(
1844        &self,
1845        request_code: RequestCode,
1846        timeout_millis: u64,
1847    ) -> ClientResult<TopicList> {
1848        let request = self.create_request_command(request_code, EmptyHeader {});
1849        let outcome = self.remoting_client.invoke_request(None, request, timeout_millis).await;
1850        let response = match outcome {
1851            Ok(OutboundRequestOutcome::Response(response)) => response,
1852            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1853                return Err(admin_request_error(
1854                    "get_topic_list_from_name_server_by_code",
1855                    RetryInput::Rejected(rejection),
1856                ));
1857            }
1858            Ok(OutboundRequestOutcome::Contract(contract)) => {
1859                return Err(admin_request_error(
1860                    "get_topic_list_from_name_server_by_code",
1861                    RetryInput::Contract(contract),
1862                ));
1863            }
1864            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1865        };
1866        if ResponseCode::from(response.code()) == ResponseCode::Success {
1867            if let Some(body) = response.get_body() {
1868                return TopicList::decode(body.as_ref()).map_err(ClientError::from);
1869            }
1870        }
1871        Err(mq_client_err!(
1872            response.code(),
1873            response.remark().map_or("".to_string(), |s| s.to_string())
1874        ))
1875    }
1876
1877    pub(crate) async fn get_topic_list_from_name_server(&self, timeout_millis: u64) -> ClientResult<TopicList> {
1878        self.get_topic_list_from_name_server_by_code(RequestCode::GetAllTopicListFromNameserver, timeout_millis)
1879            .await
1880    }
1881
1882    pub(crate) async fn get_all_topic_list_from_name_server(&self, timeout_millis: u64) -> ClientResult<TopicList> {
1883        self.get_topic_list_from_name_server(timeout_millis).await
1884    }
1885
1886    pub(crate) async fn get_system_topic_list(&self, timeout_millis: u64) -> ClientResult<TopicList> {
1887        let mut topic_list = self
1888            .get_topic_list_from_name_server_by_code(RequestCode::GetSystemTopicListFromNs, timeout_millis)
1889            .await?;
1890        merge_system_topic_list_from_broker(self, &mut topic_list, timeout_millis).await?;
1891        Ok(topic_list)
1892    }
1893
1894    pub(crate) async fn get_unit_topic_list(
1895        &self,
1896        contain_retry: bool,
1897        timeout_millis: u64,
1898    ) -> ClientResult<TopicList> {
1899        let mut topic_list = self
1900            .get_topic_list_from_name_server_by_code(RequestCode::GetUnitTopicList, timeout_millis)
1901            .await?;
1902        filter_retry_topics_like_java(&mut topic_list, contain_retry);
1903        Ok(topic_list)
1904    }
1905
1906    pub(crate) async fn get_has_unit_sub_topic_list(
1907        &self,
1908        contain_retry: bool,
1909        timeout_millis: u64,
1910    ) -> ClientResult<TopicList> {
1911        let mut topic_list = self
1912            .get_topic_list_from_name_server_by_code(RequestCode::GetHasUnitSubTopicList, timeout_millis)
1913            .await?;
1914        filter_retry_topics_like_java(&mut topic_list, contain_retry);
1915        Ok(topic_list)
1916    }
1917
1918    pub(crate) async fn get_has_unit_sub_un_unit_topic_list(
1919        &self,
1920        contain_retry: bool,
1921        timeout_millis: u64,
1922    ) -> ClientResult<TopicList> {
1923        let mut topic_list = self
1924            .get_topic_list_from_name_server_by_code(RequestCode::GetHasUnitSubUnunitTopicList, timeout_millis)
1925            .await?;
1926        filter_retry_topics_like_java(&mut topic_list, contain_retry);
1927        Ok(topic_list)
1928    }
1929
1930    pub(crate) async fn get_topics_by_cluster(
1931        &self,
1932        cluster: CheetahString,
1933        timeout_millis: u64,
1934    ) -> ClientResult<rocketmq_protocol::protocol::body::topic::topic_list::TopicList> {
1935        let request_header = GetTopicsByClusterRequestHeader::new(cluster);
1936        let request = self.create_request_command(RequestCode::GetTopicsByCluster, request_header);
1937        let outcome = self.remoting_client.invoke_request(None, request, timeout_millis).await;
1938        let response = match outcome {
1939            Ok(OutboundRequestOutcome::Response(response)) => response,
1940            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1941                return Err(admin_request_error(
1942                    "get_topics_by_cluster",
1943                    RetryInput::Rejected(rejection),
1944                ));
1945            }
1946            Ok(OutboundRequestOutcome::Contract(contract)) => {
1947                return Err(admin_request_error(
1948                    "get_topics_by_cluster",
1949                    RetryInput::Contract(contract),
1950                ));
1951            }
1952            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
1953        };
1954        if ResponseCode::from(response.code()) == ResponseCode::Success {
1955            if let Some(body) = response.get_body() {
1956                return rocketmq_protocol::protocol::body::topic::topic_list::TopicList::decode(body.as_ref())
1957                    .map_err(ClientError::from);
1958            }
1959        }
1960        Err(mq_client_err!(
1961            response.code(),
1962            response.remark().map_or("".to_string(), |s| s.to_string())
1963        ))
1964    }
1965
1966    pub(crate) async fn get_cluster_list(
1967        &self,
1968        topic: CheetahString,
1969        timeout_millis: u64,
1970    ) -> ClientResult<HashSet<CheetahString>> {
1971        let topic_route_data = self
1972            .get_topic_route_info_from_name_server(&topic, timeout_millis)
1973            .await?
1974            .ok_or_else(|| mq_client_err!(format!("Topic route not found for: {topic}")))?;
1975        let cluster_info = self.get_broker_cluster_info(timeout_millis).await?;
1976        Ok(cluster_names_for_topic_route(&cluster_info, &topic_route_data))
1977    }
1978
1979    pub(crate) async fn get_system_topic_list_from_broker(
1980        &self,
1981        addr: &CheetahString,
1982        timeout_millis: u64,
1983    ) -> ClientResult<rocketmq_protocol::protocol::body::topic::topic_list::TopicList> {
1984        let request = self.create_request_command(RequestCode::GetSystemTopicListFromBroker, EmptyHeader {});
1985        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
1986        let outcome = self
1987            .remoting_client
1988            .invoke_request(Some(&broker_addr), request, timeout_millis)
1989            .await;
1990        let response = match outcome {
1991            Ok(OutboundRequestOutcome::Response(response)) => response,
1992            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
1993                return Err(admin_request_error(
1994                    "get_system_topic_list_from_broker",
1995                    RetryInput::Rejected(rejection),
1996                ));
1997            }
1998            Ok(OutboundRequestOutcome::Contract(contract)) => {
1999                return Err(admin_request_error(
2000                    "get_system_topic_list_from_broker",
2001                    RetryInput::Contract(contract),
2002                ));
2003            }
2004            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2005        };
2006        if ResponseCode::from(response.code()) == ResponseCode::Success {
2007            if let Some(body) = response.get_body() {
2008                return rocketmq_protocol::protocol::body::topic::topic_list::TopicList::decode(body.as_ref())
2009                    .map_err(ClientError::from);
2010            }
2011        }
2012        Err(mq_client_err!(
2013            response.code(),
2014            response.remark().map_or("".to_string(), |s| s.to_string())
2015        ))
2016    }
2017
2018    pub(crate) async fn get_consume_stats(
2019        &self,
2020        addr: &CheetahString,
2021        request_header: rocketmq_protocol::protocol::header::get_consume_stats_request_header::GetConsumeStatsRequestHeader,
2022        timeout_millis: u64,
2023    ) -> ClientResult<rocketmq_protocol::protocol::admin::consume_stats::ConsumeStats> {
2024        let request = self.create_request_command(RequestCode::GetConsumeStats, request_header);
2025        let outcome = self
2026            .remoting_client
2027            .invoke_request(Some(addr), request, timeout_millis)
2028            .await;
2029        let response = match outcome {
2030            Ok(OutboundRequestOutcome::Response(response)) => response,
2031            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2032                return Err(admin_request_error(
2033                    "get_consume_stats",
2034                    RetryInput::Rejected(rejection),
2035                ));
2036            }
2037            Ok(OutboundRequestOutcome::Contract(contract)) => {
2038                return Err(admin_request_error("get_consume_stats", RetryInput::Contract(contract)));
2039            }
2040            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2041        };
2042        if ResponseCode::from(response.code()) == ResponseCode::Success {
2043            if let Some(body) = response.get_body() {
2044                return rocketmq_protocol::protocol::admin::consume_stats::ConsumeStats::decode(body.as_ref())
2045                    .map_err(ClientError::from);
2046            }
2047        }
2048        Err(client_broker_err!(
2049            response.code(),
2050            response.remark().map_or("".to_string(), |s| s.to_string()),
2051            addr
2052        ))
2053    }
2054
2055    pub(crate) async fn query_consume_time_span(
2056        &self,
2057        addr: &CheetahString,
2058        request_header: QueryConsumeTimeSpanRequestHeader,
2059        timeout_millis: u64,
2060    ) -> ClientResult<Vec<rocketmq_protocol::protocol::body::queue_time_span::QueueTimeSpan>> {
2061        let request = self.create_request_command(RequestCode::QueryConsumeTimeSpan, request_header);
2062        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
2063        let outcome = self
2064            .remoting_client
2065            .invoke_request(Some(&broker_addr), request, timeout_millis)
2066            .await;
2067        let response = match outcome {
2068            Ok(OutboundRequestOutcome::Response(response)) => response,
2069            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2070                return Err(admin_request_error(
2071                    "query_consume_time_span",
2072                    RetryInput::Rejected(rejection),
2073                ));
2074            }
2075            Ok(OutboundRequestOutcome::Contract(contract)) => {
2076                return Err(admin_request_error(
2077                    "query_consume_time_span",
2078                    RetryInput::Contract(contract),
2079                ));
2080            }
2081            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2082        };
2083        if ResponseCode::from(response.code()) == ResponseCode::Success {
2084            if let Some(body) = response.get_body() {
2085                let body: QueryConsumeTimeSpanBody = decode_admin_json(body.as_ref())?;
2086                return Ok(body.consume_time_span_set);
2087            }
2088        }
2089        Err(mq_client_err!(
2090            response.code(),
2091            response.remark().map_or("".to_string(), |s| s.to_string())
2092        ))
2093    }
2094
2095    pub(crate) async fn get_topic_stats_info(
2096        &self,
2097        addr: &CheetahString,
2098        request_header: GetTopicStatsInfoRequestHeader,
2099        timeout_millis: u64,
2100    ) -> ClientResult<rocketmq_protocol::protocol::admin::topic_stats_table::TopicStatsTable> {
2101        let request = self.create_request_command(RequestCode::GetTopicStatsInfo, request_header);
2102        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
2103        let outcome = self
2104            .remoting_client
2105            .invoke_request(Some(&broker_addr), request, timeout_millis)
2106            .await;
2107        let response = match outcome {
2108            Ok(OutboundRequestOutcome::Response(response)) => response,
2109            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2110                return Err(admin_request_error(
2111                    "get_topic_stats_info",
2112                    RetryInput::Rejected(rejection),
2113                ));
2114            }
2115            Ok(OutboundRequestOutcome::Contract(contract)) => {
2116                return Err(admin_request_error(
2117                    "get_topic_stats_info",
2118                    RetryInput::Contract(contract),
2119                ));
2120            }
2121            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2122        };
2123        if ResponseCode::from(response.code()) == ResponseCode::Success {
2124            if let Some(body) = response.get_body() {
2125                return rocketmq_protocol::protocol::admin::topic_stats_table::TopicStatsTable::decode(body.as_ref())
2126                    .map_err(ClientError::from);
2127            }
2128        }
2129        Err(mq_client_err!(
2130            response.code(),
2131            response.remark().map_or("".to_string(), |s| s.to_string())
2132        ))
2133    }
2134
2135    pub(crate) async fn get_topic_config(
2136        &self,
2137        addr: &CheetahString,
2138        request_header: GetTopicConfigRequestHeader,
2139        timeout_millis: u64,
2140    ) -> ClientResult<TopicConfigAndQueueMapping> {
2141        let request = self.create_request_command(RequestCode::GetTopicConfig, request_header);
2142        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
2143        let outcome = self
2144            .remoting_client
2145            .invoke_request(Some(&broker_addr), request, timeout_millis)
2146            .await;
2147        let response = match outcome {
2148            Ok(OutboundRequestOutcome::Response(response)) => response,
2149            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2150                return Err(admin_request_error("get_topic_config", RetryInput::Rejected(rejection)));
2151            }
2152            Ok(OutboundRequestOutcome::Contract(contract)) => {
2153                return Err(admin_request_error("get_topic_config", RetryInput::Contract(contract)));
2154            }
2155            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2156        };
2157        if ResponseCode::from(response.code()) == ResponseCode::Success {
2158            if let Some(body) = response.get_body() {
2159                return decode_admin_json(body.as_ref());
2160            }
2161        }
2162        Err(mq_client_err!(
2163            response.code(),
2164            response.remark().map_or("".to_string(), |s| s.to_string())
2165        ))
2166    }
2167
2168    #[cfg(feature = "admin-mutation")]
2169    pub(crate) async fn create_static_topic(
2170        &self,
2171        addr: &CheetahString,
2172        default_topic: CheetahString,
2173        topic_config: TopicConfig,
2174        mapping_detail: TopicQueueMappingDetail,
2175        force: bool,
2176        timeout_millis: u64,
2177    ) -> ClientResult<()> {
2178        let topic = topic_config
2179            .topic_name
2180            .clone()
2181            .filter(|topic| !topic.is_empty())
2182            .ok_or_else(|| crate::ClientError::illegal_argument("Topic name is required"))?;
2183        let read_queue_nums = i32::try_from(topic_config.read_queue_nums)
2184            .map_err(|_| crate::ClientError::illegal_argument("readQueueNums exceeds Java int range"))?;
2185        let write_queue_nums = i32::try_from(topic_config.write_queue_nums)
2186            .map_err(|_| crate::ClientError::illegal_argument("writeQueueNums exceeds Java int range"))?;
2187        let perm = i32::try_from(topic_config.perm)
2188            .map_err(|_| crate::ClientError::illegal_argument("perm exceeds Java int range"))?;
2189        let topic_sys_flag = i32::try_from(topic_config.topic_sys_flag)
2190            .map_err(|_| crate::ClientError::illegal_argument("topicSysFlag exceeds Java int range"))?;
2191
2192        let request_header = CreateTopicRequestHeader {
2193            topic,
2194            default_topic,
2195            read_queue_nums,
2196            write_queue_nums,
2197            perm,
2198            topic_filter_type: CheetahString::from_static_str(topic_config.topic_filter_type.as_str()),
2199            topic_sys_flag: Some(topic_sys_flag),
2200            order: topic_config.order,
2201            attributes: None,
2202            force: Some(force),
2203            topic_request_header: None,
2204        };
2205        let request = self
2206            .create_request_command(RequestCode::UpdateAndCreateStaticTopic, request_header)
2207            .set_body(mapping_detail.encode()?);
2208        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
2209        let outcome = self
2210            .remoting_client
2211            .invoke_request(Some(&broker_addr), request, timeout_millis)
2212            .await;
2213        let response = match outcome {
2214            Ok(OutboundRequestOutcome::Response(response)) => response,
2215            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2216                return Err(admin_request_error(
2217                    "create_static_topic",
2218                    RetryInput::Rejected(rejection),
2219                ));
2220            }
2221            Ok(OutboundRequestOutcome::Contract(contract)) => {
2222                return Err(admin_request_error(
2223                    "create_static_topic",
2224                    RetryInput::Contract(contract),
2225                ));
2226            }
2227            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2228        };
2229
2230        if ResponseCode::from(response.code()) == ResponseCode::Success {
2231            return Ok(());
2232        }
2233
2234        Err(client_broker_err!(
2235            response.code(),
2236            response.remark().map_or("".to_string(), |s| s.to_string()),
2237            addr
2238        ))
2239    }
2240
2241    pub(crate) async fn query_topic_consume_by_who(
2242        &self,
2243        addr: &CheetahString,
2244        request_header: rocketmq_protocol::protocol::header::query_topic_consume_by_who_request_header::QueryTopicConsumeByWhoRequestHeader,
2245        timeout_millis: u64,
2246    ) -> ClientResult<rocketmq_protocol::protocol::body::group_list::GroupList> {
2247        let request = self.create_request_command(RequestCode::QueryTopicConsumeByWho, request_header);
2248        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
2249        let outcome = self
2250            .remoting_client
2251            .invoke_request(Some(&broker_addr), request, timeout_millis)
2252            .await;
2253        let response = match outcome {
2254            Ok(OutboundRequestOutcome::Response(response)) => response,
2255            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2256                return Err(admin_request_error(
2257                    "query_topic_consume_by_who",
2258                    RetryInput::Rejected(rejection),
2259                ));
2260            }
2261            Ok(OutboundRequestOutcome::Contract(contract)) => {
2262                return Err(admin_request_error(
2263                    "query_topic_consume_by_who",
2264                    RetryInput::Contract(contract),
2265                ));
2266            }
2267            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2268        };
2269        if ResponseCode::from(response.code()) == ResponseCode::Success {
2270            if let Some(body) = response.get_body() {
2271                return rocketmq_protocol::protocol::body::group_list::GroupList::decode(body.as_ref())
2272                    .map_err(ClientError::from);
2273            }
2274        }
2275        Err(mq_client_err!(
2276            response.code(),
2277            response.remark().map_or("".to_string(), |s| s.to_string())
2278        ))
2279    }
2280
2281    pub(crate) async fn query_topics_by_consumer(
2282        &self,
2283        addr: &CheetahString,
2284        request_header: QueryTopicsByConsumerRequestHeader,
2285        timeout_millis: u64,
2286    ) -> ClientResult<rocketmq_protocol::protocol::body::topic::topic_list::TopicList> {
2287        let request = self.create_request_command(RequestCode::QueryTopicsByConsumer, request_header);
2288        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
2289        let outcome = self
2290            .remoting_client
2291            .invoke_request(Some(&broker_addr), request, timeout_millis)
2292            .await;
2293        let response = match outcome {
2294            Ok(OutboundRequestOutcome::Response(response)) => response,
2295            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2296                return Err(admin_request_error(
2297                    "query_topics_by_consumer",
2298                    RetryInput::Rejected(rejection),
2299                ));
2300            }
2301            Ok(OutboundRequestOutcome::Contract(contract)) => {
2302                return Err(admin_request_error(
2303                    "query_topics_by_consumer",
2304                    RetryInput::Contract(contract),
2305                ));
2306            }
2307            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2308        };
2309        if ResponseCode::from(response.code()) == ResponseCode::Success {
2310            if let Some(body) = response.get_body() {
2311                return rocketmq_protocol::protocol::body::topic::topic_list::TopicList::decode(body.as_ref())
2312                    .map_err(ClientError::from);
2313            }
2314        }
2315        Err(mq_client_err!(
2316            response.code(),
2317            response.remark().map_or("".to_string(), |s| s.to_string())
2318        ))
2319    }
2320
2321    pub(crate) async fn query_subscription_by_consumer(
2322        &self,
2323        addr: &CheetahString,
2324        request_header: QuerySubscriptionByConsumerRequestHeader,
2325        timeout_millis: u64,
2326    ) -> ClientResult<SubscriptionData> {
2327        let request = self.create_request_command(RequestCode::QuerySubscriptionByConsumer, request_header);
2328        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
2329        let outcome = self
2330            .remoting_client
2331            .invoke_request(Some(&broker_addr), request, timeout_millis)
2332            .await;
2333        let response = match outcome {
2334            Ok(OutboundRequestOutcome::Response(response)) => response,
2335            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2336                return Err(admin_request_error(
2337                    "query_subscription_by_consumer",
2338                    RetryInput::Rejected(rejection),
2339                ));
2340            }
2341            Ok(OutboundRequestOutcome::Contract(contract)) => {
2342                return Err(admin_request_error(
2343                    "query_subscription_by_consumer",
2344                    RetryInput::Contract(contract),
2345                ));
2346            }
2347            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2348        };
2349        if ResponseCode::from(response.code()) == ResponseCode::Success {
2350            let body = response
2351                .get_body()
2352                .ok_or_else(|| mq_client_err!("query_subscription_by_consumer response body is empty".to_string()))?;
2353            let response_body: QuerySubscriptionResponseBody = decode_admin_json(body.as_ref())?;
2354            return response_body.subscription_data.ok_or_else(|| {
2355                mq_client_err!("query_subscription_by_consumer response subscriptionData is empty".to_string())
2356            });
2357        }
2358        Err(client_broker_err!(
2359            response.code(),
2360            response.remark().map_or("".to_string(), |s| s.to_string()),
2361            addr
2362        ))
2363    }
2364
2365    #[cfg(feature = "admin-mutation")]
2366    pub(super) async fn invoke_broker_success_request(
2367        &self,
2368        addr: &CheetahString,
2369        request_code: RequestCode,
2370        timeout_millis: u64,
2371    ) -> ClientResult<bool> {
2372        let request = self.create_request_command(request_code, EmptyHeader {});
2373        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
2374        let outcome = self
2375            .remoting_client
2376            .invoke_request(Some(&broker_addr), request, timeout_millis)
2377            .await;
2378        let response = match outcome {
2379            Ok(OutboundRequestOutcome::Response(response)) => response,
2380            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2381                return Err(admin_request_error(
2382                    "invoke_broker_success_request",
2383                    RetryInput::Rejected(rejection),
2384                ));
2385            }
2386            Ok(OutboundRequestOutcome::Contract(contract)) => {
2387                return Err(admin_request_error(
2388                    "invoke_broker_success_request",
2389                    RetryInput::Contract(contract),
2390                ));
2391            }
2392            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2393        };
2394        if ResponseCode::from(response.code()) == ResponseCode::Success {
2395            return Ok(true);
2396        }
2397        Err(mq_client_err!(
2398            response.code(),
2399            response.remark().map_or("".to_string(), |s| s.to_string())
2400        ))
2401    }
2402
2403    #[cfg(feature = "admin-mutation")]
2404    pub(crate) async fn clean_expired_consume_queue(
2405        &self,
2406        addr: &CheetahString,
2407        timeout_millis: u64,
2408    ) -> ClientResult<bool> {
2409        self.invoke_broker_success_request(addr, RequestCode::CleanExpiredConsumequeue, timeout_millis)
2410            .await
2411    }
2412
2413    #[cfg(feature = "admin-mutation")]
2414    pub(crate) async fn delete_expired_commit_log(
2415        &self,
2416        addr: &CheetahString,
2417        timeout_millis: u64,
2418    ) -> ClientResult<bool> {
2419        self.invoke_broker_success_request(addr, RequestCode::DeleteExpiredCommitlog, timeout_millis)
2420            .await
2421    }
2422
2423    #[cfg(feature = "admin-mutation")]
2424    pub(crate) async fn clean_unused_topic(&self, addr: &CheetahString, timeout_millis: u64) -> ClientResult<bool> {
2425        self.invoke_broker_success_request(addr, RequestCode::CleanUnusedTopic, timeout_millis)
2426            .await
2427    }
2428
2429    #[cfg(feature = "admin-mutation")]
2430    pub(crate) async fn clean_unused_topic_by_addr(
2431        &self,
2432        addr: &CheetahString,
2433        timeout_millis: u64,
2434    ) -> ClientResult<bool> {
2435        self.clean_unused_topic(addr, timeout_millis).await
2436    }
2437
2438    #[cfg(feature = "admin-mutation")]
2439    pub(crate) async fn create_topic(
2440        &self,
2441        addr: &CheetahString,
2442        default_topic: CheetahString,
2443        topic_config: &TopicConfig,
2444        timeout_millis: u64,
2445    ) -> ClientResult<()> {
2446        let request_header = create_topic_request_header_like_java(default_topic, topic_config)?;
2447        self.update_or_create_topic(addr, request_header, timeout_millis).await
2448    }
2449
2450    #[cfg(feature = "admin-mutation")]
2451    pub(crate) async fn create_topic_list(
2452        &self,
2453        address: &CheetahString,
2454        topic_config_list: Vec<TopicConfig>,
2455        timeout_millis: u64,
2456    ) -> ClientResult<()> {
2457        let request = create_topic_list_request(&self.command_factory, topic_config_list)?;
2458        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, address.as_str());
2459        let outcome = self
2460            .remoting_client
2461            .invoke_request(Some(&broker_addr), request, timeout_millis)
2462            .await;
2463        let response = match outcome {
2464            Ok(OutboundRequestOutcome::Response(response)) => response,
2465            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2466                return Err(admin_request_error(
2467                    "create_topic_list",
2468                    RetryInput::Rejected(rejection),
2469                ));
2470            }
2471            Ok(OutboundRequestOutcome::Contract(contract)) => {
2472                return Err(admin_request_error("create_topic_list", RetryInput::Contract(contract)));
2473            }
2474            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2475        };
2476        match ResponseCode::from(response.code()) {
2477            ResponseCode::Success => Ok(()),
2478            _ => Err(mq_client_err!(
2479                response.code(),
2480                response.remark().map_or_else(String::new, |remark| remark.to_string())
2481            )),
2482        }
2483    }
2484
2485    pub(crate) async fn update_or_create_topic(
2486        &self,
2487        addr: &CheetahString,
2488        request_header: CreateTopicRequestHeader,
2489        timeout_millis: u64,
2490    ) -> ClientResult<()> {
2491        self.update_or_create_topic_once(
2492            addr,
2493            request_header,
2494            RequestDeadline::from_timeout_millis(timeout_millis),
2495        )
2496        .await
2497        .map_err(|input| admin_request_error("update_or_create_topic", input))
2498    }
2499
2500    pub(crate) async fn update_or_create_topic_once(
2501        &self,
2502        addr: &CheetahString,
2503        request_header: CreateTopicRequestHeader,
2504        deadline: RequestDeadline,
2505    ) -> Result<(), RetryInput> {
2506        let request = self.create_request_command(RequestCode::UpdateAndCreateTopic, request_header);
2507        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
2508        let outcome = self
2509            .remoting_client
2510            .invoke_request_with_deadline(Some(&broker_addr), request, deadline)
2511            .await;
2512        let response = match outcome {
2513            Ok(OutboundRequestOutcome::Response(response)) => response,
2514            Ok(OutboundRequestOutcome::Rejected(rejection)) => return Err(RetryInput::Rejected(rejection)),
2515            Ok(OutboundRequestOutcome::Contract(contract)) => return Err(RetryInput::Contract(contract)),
2516            Err(error) => return Err(RetryInput::Transport(error)),
2517        };
2518        if ResponseCode::from(response.code()) == ResponseCode::Success {
2519            Ok(())
2520        } else {
2521            let code = response.code();
2522            Err(RetryInput::Response {
2523                code,
2524                retry_after: None,
2525                terminal_error: mq_client_err!(code, response.remark().map_or("".to_string(), |s| s.to_string())),
2526            })
2527        }
2528    }
2529
2530    #[cfg(feature = "admin-mutation")]
2531    pub(crate) async fn create_subscription_group_list(
2532        &self,
2533        address: &CheetahString,
2534        configs: Vec<SubscriptionGroupConfig>,
2535        timeout_millis: u64,
2536    ) -> ClientResult<()> {
2537        let request = create_subscription_group_list_request(&self.command_factory, configs)?;
2538        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, address.as_str());
2539        let outcome = self
2540            .remoting_client
2541            .invoke_request(Some(&broker_addr), request, timeout_millis)
2542            .await;
2543        let response = match outcome {
2544            Ok(OutboundRequestOutcome::Response(response)) => response,
2545            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2546                return Err(admin_request_error(
2547                    "create_subscription_group_list",
2548                    RetryInput::Rejected(rejection),
2549                ));
2550            }
2551            Ok(OutboundRequestOutcome::Contract(contract)) => {
2552                return Err(admin_request_error(
2553                    "create_subscription_group_list",
2554                    RetryInput::Contract(contract),
2555                ));
2556            }
2557            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2558        };
2559
2560        match ResponseCode::from(response.code()) {
2561            ResponseCode::Success => Ok(()),
2562            _ => Err(mq_client_err!(
2563                response.code(),
2564                response.remark().map_or_else(String::new, |remark| remark.to_string())
2565            )),
2566        }
2567    }
2568
2569    #[cfg(feature = "admin-mutation")]
2570    pub(crate) async fn delete_topic_in_broker(
2571        &self,
2572        addr: &CheetahString,
2573        request_header: DeleteTopicRequestHeader,
2574        timeout_millis: u64,
2575    ) -> ClientResult<()> {
2576        let request = self.create_request_command(RequestCode::DeleteTopicInBroker, request_header);
2577        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
2578        let outcome = self
2579            .remoting_client
2580            .invoke_request(Some(&broker_addr), request, timeout_millis)
2581            .await;
2582        let response = match outcome {
2583            Ok(OutboundRequestOutcome::Response(response)) => response,
2584            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2585                return Err(admin_request_error(
2586                    "delete_topic_in_broker",
2587                    RetryInput::Rejected(rejection),
2588                ));
2589            }
2590            Ok(OutboundRequestOutcome::Contract(contract)) => {
2591                return Err(admin_request_error(
2592                    "delete_topic_in_broker",
2593                    RetryInput::Contract(contract),
2594                ));
2595            }
2596            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2597        };
2598        if ResponseCode::from(response.code()) == ResponseCode::Success {
2599            return Ok(());
2600        }
2601        Err(mq_client_err!(
2602            response.code(),
2603            response.remark().map_or("".to_string(), |s| s.to_string())
2604        ))
2605    }
2606
2607    /// Deletes multiple topics with one broker request.
2608    #[cfg(feature = "admin-mutation")]
2609    pub async fn delete_topic_in_broker_list(
2610        &self,
2611        addr: &CheetahString,
2612        topic_list: Vec<CheetahString>,
2613        timeout_millis: u64,
2614    ) -> ClientResult<()> {
2615        let request = delete_topic_list_request(&self.command_factory, topic_list)?;
2616        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
2617        let outcome = self
2618            .remoting_client
2619            .invoke_request(Some(&broker_addr), request, timeout_millis)
2620            .await;
2621        let response = match outcome {
2622            Ok(OutboundRequestOutcome::Response(response)) => response,
2623            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2624                return Err(admin_request_error(
2625                    "delete_topic_in_broker_list",
2626                    RetryInput::Rejected(rejection),
2627                ));
2628            }
2629            Ok(OutboundRequestOutcome::Contract(contract)) => {
2630                return Err(admin_request_error(
2631                    "delete_topic_in_broker_list",
2632                    RetryInput::Contract(contract),
2633                ));
2634            }
2635            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2636        };
2637
2638        match ResponseCode::from(response.code()) {
2639            ResponseCode::Success => Ok(()),
2640            _ => Err(mq_client_err!(
2641                response.code(),
2642                response.remark().map_or_else(String::new, |remark| remark.to_string())
2643            )),
2644        }
2645    }
2646
2647    #[cfg(feature = "admin-mutation")]
2648    pub(crate) async fn delete_topic_in_name_server(
2649        &self,
2650        addr: &CheetahString,
2651        cluster_name: Option<CheetahString>,
2652        topic: CheetahString,
2653        timeout_millis: u64,
2654    ) -> ClientResult<()> {
2655        self.delete_topic_in_nameserver(
2656            addr,
2657            DeleteTopicFromNamesrvRequestHeader::new(topic, cluster_name),
2658            timeout_millis,
2659        )
2660        .await
2661    }
2662
2663    #[cfg(feature = "admin-mutation")]
2664    pub(crate) async fn delete_topic_in_nameserver(
2665        &self,
2666        addr: &CheetahString,
2667        request_header: DeleteTopicFromNamesrvRequestHeader,
2668        timeout_millis: u64,
2669    ) -> ClientResult<()> {
2670        let request = self.create_request_command(RequestCode::DeleteTopicInNamesrv, request_header);
2671        let outcome = self
2672            .remoting_client
2673            .invoke_request(Some(addr), request, timeout_millis)
2674            .await;
2675        let response = match outcome {
2676            Ok(OutboundRequestOutcome::Response(response)) => response,
2677            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2678                return Err(admin_request_error(
2679                    "delete_topic_in_nameserver",
2680                    RetryInput::Rejected(rejection),
2681                ));
2682            }
2683            Ok(OutboundRequestOutcome::Contract(contract)) => {
2684                return Err(admin_request_error(
2685                    "delete_topic_in_nameserver",
2686                    RetryInput::Contract(contract),
2687                ));
2688            }
2689            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2690        };
2691        if ResponseCode::from(response.code()) == ResponseCode::Success {
2692            return Ok(());
2693        }
2694        Err(mq_client_err!(
2695            response.code(),
2696            response.remark().map_or("".to_string(), |s| s.to_string())
2697        ))
2698    }
2699
2700    #[cfg(feature = "admin-mutation")]
2701    pub(crate) async fn invoke_broker_to_reset_offset(
2702        &self,
2703        addr: &CheetahString,
2704        request_header: ResetOffsetRequestHeader,
2705        timeout_millis: u64,
2706    ) -> ClientResult<HashMap<MessageQueue, i64>> {
2707        let request = self.create_request_command(RequestCode::InvokeBrokerToResetOffset, request_header);
2708        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
2709        let outcome = self
2710            .remoting_client
2711            .invoke_request(Some(&broker_addr), request, timeout_millis)
2712            .await;
2713        let response = match outcome {
2714            Ok(OutboundRequestOutcome::Response(response)) => response,
2715            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2716                return Err(admin_request_error(
2717                    "invoke_broker_to_reset_offset",
2718                    RetryInput::Rejected(rejection),
2719                ));
2720            }
2721            Ok(OutboundRequestOutcome::Contract(contract)) => {
2722                return Err(admin_request_error(
2723                    "invoke_broker_to_reset_offset",
2724                    RetryInput::Contract(contract),
2725                ));
2726            }
2727            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2728        };
2729        if ResponseCode::from(response.code()) == ResponseCode::Success {
2730            return reset_offset_table_from_response(&response);
2731        }
2732        Err(mq_client_err!(
2733            response.code(),
2734            response.remark().map_or("".to_string(), |s| s.to_string())
2735        ))
2736    }
2737
2738    pub(crate) async fn query_correction_offset(
2739        &self,
2740        addr: &CheetahString,
2741        topic: CheetahString,
2742        group: CheetahString,
2743        filter_groups: Option<Vec<CheetahString>>,
2744        timeout_millis: u64,
2745    ) -> ClientResult<HashMap<i32, i64>> {
2746        let request = query_correction_offset_request(&self.command_factory, topic, group, filter_groups);
2747        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
2748        let outcome = self
2749            .remoting_client
2750            .invoke_request(Some(&broker_addr), request, timeout_millis)
2751            .await;
2752        let response = match outcome {
2753            Ok(OutboundRequestOutcome::Response(response)) => response,
2754            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2755                return Err(admin_request_error(
2756                    "query_correction_offset",
2757                    RetryInput::Rejected(rejection),
2758                ));
2759            }
2760            Ok(OutboundRequestOutcome::Contract(contract)) => {
2761                return Err(admin_request_error(
2762                    "query_correction_offset",
2763                    RetryInput::Contract(contract),
2764                ));
2765            }
2766            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2767        };
2768        if ResponseCode::from(response.code()) == ResponseCode::Success {
2769            if let Some(body) = response.get_body() {
2770                return QueryCorrectionOffsetBody::decode(body.as_ref())
2771                    .map(|body| body.correction_offsets)
2772                    .map_err(ClientError::from);
2773            }
2774        }
2775        Err(mq_client_err!(
2776            response.code(),
2777            response.remark().map_or("".to_string(), |s| s.to_string())
2778        ))
2779    }
2780
2781    pub(crate) async fn view_broker_stats_data(
2782        &self,
2783        addr: &CheetahString,
2784        request_header: rocketmq_protocol::protocol::header::view_broker_stats_data_request_header::ViewBrokerStatsDataRequestHeader,
2785        timeout_millis: u64,
2786    ) -> ClientResult<rocketmq_protocol::protocol::subscription::broker_stats_data::BrokerStatsData> {
2787        let request = self.create_request_command(RequestCode::ViewBrokerStatsData, request_header);
2788        let outcome = self
2789            .remoting_client
2790            .invoke_request(Some(addr), request, timeout_millis)
2791            .await;
2792        let response = match outcome {
2793            Ok(OutboundRequestOutcome::Response(response)) => response,
2794            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2795                return Err(admin_request_error(
2796                    "view_broker_stats_data",
2797                    RetryInput::Rejected(rejection),
2798                ));
2799            }
2800            Ok(OutboundRequestOutcome::Contract(contract)) => {
2801                return Err(admin_request_error(
2802                    "view_broker_stats_data",
2803                    RetryInput::Contract(contract),
2804                ));
2805            }
2806            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2807        };
2808        if ResponseCode::from(response.code()) == ResponseCode::Success {
2809            if let Some(body) = response.get_body() {
2810                return rocketmq_protocol::protocol::subscription::broker_stats_data::BrokerStatsData::decode(
2811                    body.as_ref(),
2812                )
2813                .map_err(ClientError::from);
2814            }
2815        }
2816        Err(mq_client_err!(
2817            response.code(),
2818            response.remark().map_or("".to_string(), |s| s.to_string())
2819        ))
2820    }
2821
2822    pub(crate) async fn fetch_consume_stats_in_broker(
2823        &self,
2824        addr: &CheetahString,
2825        request_header: rocketmq_protocol::protocol::header::get_consume_stats_in_broker_header::GetConsumeStatsInBrokerHeader,
2826        timeout_millis: u64,
2827    ) -> ClientResult<rocketmq_protocol::protocol::admin::consume_stats_list::ConsumeStatsList> {
2828        let request = self.create_request_command(RequestCode::GetBrokerConsumeStats, request_header);
2829        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
2830        let outcome = self
2831            .remoting_client
2832            .invoke_request(Some(&broker_addr), request, timeout_millis)
2833            .await;
2834        let response = match outcome {
2835            Ok(OutboundRequestOutcome::Response(response)) => response,
2836            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2837                return Err(admin_request_error(
2838                    "fetch_consume_stats_in_broker",
2839                    RetryInput::Rejected(rejection),
2840                ));
2841            }
2842            Ok(OutboundRequestOutcome::Contract(contract)) => {
2843                return Err(admin_request_error(
2844                    "fetch_consume_stats_in_broker",
2845                    RetryInput::Contract(contract),
2846                ));
2847            }
2848            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2849        };
2850
2851        if ResponseCode::from(response.code()) == ResponseCode::Success {
2852            if let Some(body) = response.get_body() {
2853                return rocketmq_protocol::protocol::admin::consume_stats_list::ConsumeStatsList::decode(body.as_ref())
2854                    .map_err(ClientError::from);
2855            }
2856        }
2857
2858        Err(client_broker_err!(
2859            response.code(),
2860            response.remark().map_or("".to_string(), |s| s.to_string()),
2861            addr
2862        ))
2863    }
2864
2865    #[cfg(feature = "admin-mutation")]
2866    pub(crate) async fn clone_group_offset(
2867        &self,
2868        addr: &CheetahString,
2869        src_group: CheetahString,
2870        dest_group: CheetahString,
2871        topic: CheetahString,
2872        is_offline: bool,
2873        timeout_millis: u64,
2874    ) -> ClientResult<()> {
2875        let request_header = CloneGroupOffsetRequestHeader {
2876            src_group,
2877            dest_group,
2878            topic: Some(topic),
2879            offline: is_offline,
2880            rpc_request_header: None,
2881        };
2882        let request = self.create_request_command(RequestCode::CloneGroupOffset, request_header);
2883        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
2884        let outcome = self
2885            .remoting_client
2886            .invoke_request(Some(&broker_addr), request, timeout_millis)
2887            .await;
2888        let response = match outcome {
2889            Ok(OutboundRequestOutcome::Response(response)) => response,
2890            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2891                return Err(admin_request_error(
2892                    "clone_group_offset",
2893                    RetryInput::Rejected(rejection),
2894                ));
2895            }
2896            Ok(OutboundRequestOutcome::Contract(contract)) => {
2897                return Err(admin_request_error(
2898                    "clone_group_offset",
2899                    RetryInput::Contract(contract),
2900                ));
2901            }
2902            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2903        };
2904
2905        if ResponseCode::from(response.code()) == ResponseCode::Success {
2906            return Ok(());
2907        }
2908
2909        Err(client_broker_err!(
2910            response.code(),
2911            response.remark().map_or("".to_string(), |s| s.to_string()),
2912            addr
2913        ))
2914    }
2915}
2916
2917impl MQClientAPIImpl {
2918    #[cfg(feature = "admin-read")]
2919    pub(crate) async fn get_proxy_drain_state(
2920        &self,
2921        addr: &CheetahString,
2922        timeout_millis: u64,
2923    ) -> ClientResult<ProxyDrainStateResponseBody> {
2924        let request = self.create_remoting_command(RequestCode::GetProxyDrainState);
2925        let outcome = self
2926            .remoting_client
2927            .invoke_request(Some(addr), request, timeout_millis)
2928            .await;
2929        let response = match outcome {
2930            Ok(OutboundRequestOutcome::Response(response)) => response,
2931            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2932                return Err(admin_request_error(
2933                    "get_proxy_drain_state",
2934                    RetryInput::Rejected(rejection),
2935                ));
2936            }
2937            Ok(OutboundRequestOutcome::Contract(contract)) => {
2938                return Err(admin_request_error(
2939                    "get_proxy_drain_state",
2940                    RetryInput::Contract(contract),
2941                ));
2942            }
2943            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
2944        };
2945        decode_proxy_drain_response(response, "get_proxy_drain_state")
2946    }
2947
2948    #[cfg(feature = "admin-mutation")]
2949    pub(crate) async fn begin_proxy_drain(
2950        &self,
2951        addr: &CheetahString,
2952        operation_id: CheetahString,
2953        timeout_millis: u64,
2954    ) -> ClientResult<ProxyDrainStateResponseBody> {
2955        self.proxy_drain_operation(addr, RequestCode::BeginProxyDrain, operation_id, timeout_millis)
2956            .await
2957    }
2958
2959    #[cfg(feature = "admin-mutation")]
2960    pub(crate) async fn cancel_proxy_drain(
2961        &self,
2962        addr: &CheetahString,
2963        operation_id: CheetahString,
2964        timeout_millis: u64,
2965    ) -> ClientResult<ProxyDrainStateResponseBody> {
2966        self.proxy_drain_operation(addr, RequestCode::CancelProxyDrain, operation_id, timeout_millis)
2967            .await
2968    }
2969
2970    #[cfg(feature = "admin-mutation")]
2971    async fn proxy_drain_operation(
2972        &self,
2973        addr: &CheetahString,
2974        request_code: RequestCode,
2975        operation_id: CheetahString,
2976        timeout_millis: u64,
2977    ) -> ClientResult<ProxyDrainStateResponseBody> {
2978        if operation_id.trim().is_empty() {
2979            return Err(ClientError::illegal_argument(
2980                "Proxy drain operation id must not be empty",
2981            ));
2982        }
2983        let body = ProxyDrainOperationRequestBody {
2984            schema_version: PROXY_DRAIN_SCHEMA_VERSION.to_owned(),
2985            operation_id: operation_id.to_string(),
2986        }
2987        .encode()?;
2988        let request = self.create_request(request_code, body);
2989        let outcome = self
2990            .remoting_client
2991            .invoke_request(Some(addr), request, timeout_millis)
2992            .await;
2993        let response = match outcome {
2994            Ok(OutboundRequestOutcome::Response(response)) => response,
2995            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
2996                return Err(admin_request_error(
2997                    "proxy_drain_operation",
2998                    RetryInput::Rejected(rejection),
2999                ));
3000            }
3001            Ok(OutboundRequestOutcome::Contract(contract)) => {
3002                return Err(admin_request_error(
3003                    "proxy_drain_operation",
3004                    RetryInput::Contract(contract),
3005                ));
3006            }
3007            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3008        };
3009        decode_proxy_drain_response(response, "proxy_drain_operation")
3010    }
3011
3012    #[cfg(feature = "admin-mutation")]
3013    pub(crate) async fn update_broker_config(
3014        &self,
3015        addr: &CheetahString,
3016        properties: HashMap<CheetahString, CheetahString>,
3017        timeout_millis: u64,
3018    ) -> ClientResult<()> {
3019        let validator_input = properties
3020            .iter()
3021            .map(|(key, value)| (key.to_string(), value.to_string()))
3022            .collect::<HashMap<String, String>>();
3023        crate::base::validators::Validators::check_broker_config(&validator_input)?;
3024
3025        let body = mix_all::properties_to_string(&properties);
3026        if body.is_empty() {
3027            return Ok(());
3028        }
3029
3030        let request = self
3031            .create_remoting_command(RequestCode::UpdateBrokerConfig)
3032            .set_body(body.to_string());
3033        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
3034        let outcome = self
3035            .remoting_client
3036            .invoke_request(Some(&broker_addr), request, timeout_millis)
3037            .await;
3038        let response = match outcome {
3039            Ok(OutboundRequestOutcome::Response(response)) => response,
3040            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3041                return Err(admin_request_error(
3042                    "update_broker_config",
3043                    RetryInput::Rejected(rejection),
3044                ));
3045            }
3046            Ok(OutboundRequestOutcome::Contract(contract)) => {
3047                return Err(admin_request_error(
3048                    "update_broker_config",
3049                    RetryInput::Contract(contract),
3050                ));
3051            }
3052            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3053        };
3054
3055        match ResponseCode::from(response.code()) {
3056            ResponseCode::Success => Ok(()),
3057            _ => Err(mq_client_err!(
3058                response.code(),
3059                response.remark().map_or_else(String::new, |remark| remark.to_string())
3060            )),
3061        }
3062    }
3063
3064    #[cfg(feature = "admin-mutation")]
3065    pub async fn add_broker(
3066        &self,
3067        addr: &CheetahString,
3068        broker_config_path: CheetahString,
3069        timeout_millis: u64,
3070    ) -> ClientResult<()> {
3071        let request_header = AddBrokerRequestHeader {
3072            config_path: Some(broker_config_path),
3073        };
3074        let request = self.create_request_command(RequestCode::AddBroker, request_header);
3075        let outcome = self
3076            .remoting_client
3077            .invoke_request(Some(addr), request, timeout_millis)
3078            .await;
3079        let response = match outcome {
3080            Ok(OutboundRequestOutcome::Response(response)) => response,
3081            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3082                return Err(admin_request_error("add_broker", RetryInput::Rejected(rejection)));
3083            }
3084            Ok(OutboundRequestOutcome::Contract(contract)) => {
3085                return Err(admin_request_error("add_broker", RetryInput::Contract(contract)));
3086            }
3087            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3088        };
3089
3090        match ResponseCode::from(response.code()) {
3091            ResponseCode::Success => Ok(()),
3092            _ => Err(mq_client_err!(
3093                response.code(),
3094                response.remark().map_or_else(String::new, |remark| remark.to_string())
3095            )),
3096        }
3097    }
3098
3099    #[cfg(feature = "admin-mutation")]
3100    pub async fn remove_broker(
3101        &self,
3102        addr: &CheetahString,
3103        cluster_name: CheetahString,
3104        broker_name: CheetahString,
3105        broker_id: u64,
3106        timeout_millis: u64,
3107    ) -> ClientResult<()> {
3108        let request_header = RemoveBrokerRequestHeader {
3109            broker_name,
3110            broker_cluster_name: cluster_name,
3111            broker_id,
3112        };
3113        let request = self.create_request_command(RequestCode::RemoveBroker, request_header);
3114        let outcome = self
3115            .remoting_client
3116            .invoke_request(Some(addr), request, timeout_millis)
3117            .await;
3118        let response = match outcome {
3119            Ok(OutboundRequestOutcome::Response(response)) => response,
3120            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3121                return Err(admin_request_error("remove_broker", RetryInput::Rejected(rejection)));
3122            }
3123            Ok(OutboundRequestOutcome::Contract(contract)) => {
3124                return Err(admin_request_error("remove_broker", RetryInput::Contract(contract)));
3125            }
3126            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3127        };
3128
3129        match ResponseCode::from(response.code()) {
3130            ResponseCode::Success => Ok(()),
3131            _ => Err(mq_client_err!(
3132                response.code(),
3133                response.remark().map_or_else(String::new, |remark| remark.to_string())
3134            )),
3135        }
3136    }
3137
3138    #[cfg(feature = "admin-mutation")]
3139    pub(crate) async fn notify_min_broker_id_changed(
3140        &self,
3141        broker_addr: &CheetahString,
3142        request_header: NotifyMinBrokerIdChangeRequestHeader,
3143        timeout_millis: u64,
3144    ) -> ClientResult<()> {
3145        let request = self.create_request_command(RequestCode::NotifyMinBrokerIdChange, request_header);
3146        self.remoting_client
3147            .invoke_request_oneway(broker_addr, request, timeout_millis)
3148            .await
3149            .map_err(ClientError::from)
3150    }
3151
3152    #[cfg(feature = "admin-mutation")]
3153    pub async fn export_rocksdb_config_to_json(
3154        &self,
3155        broker_addr: CheetahString,
3156        config_types: Vec<CheetahString>,
3157        timeout_millis: u64,
3158    ) -> ClientResult<()> {
3159        let mut config_type = config_types
3160            .into_iter()
3161            .map(|config_type| config_type.as_str().trim().to_string())
3162            .filter(|config_type| !config_type.is_empty())
3163            .collect::<Vec<_>>()
3164            .join(";");
3165        if !config_type.is_empty() {
3166            config_type.push(';');
3167        }
3168
3169        let request_header = ExportRocksdbConfigToJsonRequestHeader {
3170            config_type: CheetahString::from(config_type),
3171        };
3172        let request = self.create_request_command(RequestCode::ExportRocksdbConfigToJson, request_header);
3173        let outcome = self
3174            .remoting_client
3175            .invoke_request(Some(&broker_addr), request, timeout_millis)
3176            .await;
3177        let response = match outcome {
3178            Ok(OutboundRequestOutcome::Response(response)) => response,
3179            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3180                return Err(admin_request_error(
3181                    "export_rocksdb_config_to_json",
3182                    RetryInput::Rejected(rejection),
3183                ));
3184            }
3185            Ok(OutboundRequestOutcome::Contract(contract)) => {
3186                return Err(admin_request_error(
3187                    "export_rocksdb_config_to_json",
3188                    RetryInput::Contract(contract),
3189                ));
3190            }
3191            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3192        };
3193
3194        match ResponseCode::from(response.code()) {
3195            ResponseCode::Success => Ok(()),
3196            _ => Err(mq_client_err!(
3197                response.code(),
3198                response.remark().map_or_else(String::new, |remark| remark.to_string())
3199            )),
3200        }
3201    }
3202
3203    #[cfg(feature = "admin-mutation")]
3204    pub async fn export_rocks_db_config_to_json(
3205        &self,
3206        broker_addr: CheetahString,
3207        config_types: Vec<CheetahString>,
3208        timeout_millis: u64,
3209    ) -> ClientResult<()> {
3210        self.export_rocksdb_config_to_json(broker_addr, config_types, timeout_millis)
3211            .await
3212    }
3213
3214    pub(crate) async fn get_all_consumer_offset_json(
3215        &self,
3216        broker_addr: CheetahString,
3217        timeout_millis: u64,
3218    ) -> ClientResult<CheetahString> {
3219        let request = get_all_consumer_offset_request(&self.command_factory);
3220        let outcome = self
3221            .remoting_client
3222            .invoke_request(Some(&broker_addr), request, timeout_millis)
3223            .await;
3224        let response = match outcome {
3225            Ok(OutboundRequestOutcome::Response(response)) => response,
3226            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3227                return Err(admin_request_error(
3228                    "get_all_consumer_offset_json",
3229                    RetryInput::Rejected(rejection),
3230                ));
3231            }
3232            Ok(OutboundRequestOutcome::Contract(contract)) => {
3233                return Err(admin_request_error(
3234                    "get_all_consumer_offset_json",
3235                    RetryInput::Contract(contract),
3236                ));
3237            }
3238            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3239        };
3240
3241        consumer_offset_json_from_response(&response)
3242    }
3243
3244    pub(crate) async fn get_broker_config(
3245        &self,
3246        addr: &CheetahString,
3247        timeout_millis: u64,
3248    ) -> ClientResult<HashMap<CheetahString, CheetahString>> {
3249        self.get_broker_config_snapshot(addr, timeout_millis)
3250            .await
3251            .map(|snapshot| snapshot.properties)
3252    }
3253
3254    pub async fn get_name_server_config(
3255        &self,
3256        name_servers: Option<Vec<CheetahString>>,
3257        timeout_millis: Duration,
3258    ) -> ClientResult<Option<HashMap<CheetahString, HashMap<CheetahString, CheetahString>>>> {
3259        // Determine which name servers to invoke
3260        let invoke_name_servers = match name_servers {
3261            Some(servers) if !servers.is_empty() => servers,
3262            _ => self.remoting_client.get_name_server_address_list().to_vec(),
3263        };
3264
3265        if invoke_name_servers.is_empty() {
3266            return Ok(None);
3267        }
3268
3269        // Create request command
3270        let request = self.create_remoting_command(RequestCode::GetNamesrvConfig);
3271        let timeout_millis = duration_millis_to_u64("getNameServerConfig", timeout_millis)?;
3272        let mut config_map = HashMap::with_capacity(4);
3273        // Iterate through each name server
3274        for name_server in invoke_name_servers {
3275            // Make synchronous call with timeout
3276            let outcome = self
3277                .remoting_client
3278                .invoke_request(Some(&name_server), request.clone(), timeout_millis)
3279                .await;
3280            let response = match outcome {
3281                Ok(OutboundRequestOutcome::Response(response)) => response,
3282                Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3283                    return Err(admin_request_error(
3284                        "get_name_server_config",
3285                        RetryInput::Rejected(rejection),
3286                    ));
3287                }
3288                Ok(OutboundRequestOutcome::Contract(contract)) => {
3289                    return Err(admin_request_error(
3290                        "get_name_server_config",
3291                        RetryInput::Contract(contract),
3292                    ));
3293                }
3294                Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3295            };
3296            // Check response code
3297            match ResponseCode::from(response.code()) {
3298                ResponseCode::Success => {
3299                    // Parse response body as properties
3300                    match response.get_body() {
3301                        Some(body) => {
3302                            let body_str = String::from_utf8_lossy(body.as_ref()).to_string();
3303                            // if body_str contains =, return from Java version
3304                            let properties = if body_str.contains('=') {
3305                                mix_all::string_to_properties(&body_str).ok_or_else(|| {
3306                                    mq_client_err!("Failed to parse namesrv config properties".to_string())
3307                                })?
3308                            } else {
3309                                SerdeJsonUtils::from_json_str::<HashMap<CheetahString, CheetahString>>(&body_str)
3310                                    .map_err(|source| {
3311                                        ClientError::response_process_source("parse namesrv config JSON", source)
3312                                    })?
3313                            };
3314
3315                            config_map.insert(name_server.clone(), properties);
3316                        }
3317                        None => return Err(mq_client_err!("Body is empty".to_string())),
3318                    }
3319                }
3320                _code => {
3321                    return Err(mq_client_err!(
3322                        response.code(),
3323                        response.remark().map_or("".to_string(), |s| s.to_string())
3324                    ));
3325                }
3326            }
3327        }
3328        Ok(Some(config_map))
3329    }
3330
3331    pub async fn probe_name_server(&self, name_server: &CheetahString, timeout_millis: Duration) -> ClientResult<()> {
3332        let request = self.create_request_command(
3333            RequestCode::GetNamesrvConfig,
3334            GetNamesrvConfigRequestHeader::for_probe(),
3335        );
3336        let timeout_millis = duration_millis_to_u64("probeNameServer", timeout_millis)?;
3337        let outcome = self
3338            .remoting_client
3339            .invoke_request(Some(name_server), request, timeout_millis)
3340            .await;
3341        let response = match outcome {
3342            Ok(OutboundRequestOutcome::Response(response)) => response,
3343            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3344                return Err(admin_request_error(
3345                    "probe_name_server",
3346                    RetryInput::Rejected(rejection),
3347                ));
3348            }
3349            Ok(OutboundRequestOutcome::Contract(contract)) => {
3350                return Err(admin_request_error("probe_name_server", RetryInput::Contract(contract)));
3351            }
3352            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3353        };
3354
3355        match ResponseCode::from(response.code()) {
3356            ResponseCode::Success => Ok(()),
3357            _ => Err(mq_client_err!(
3358                response.code(),
3359                response.remark().map_or("".to_string(), |s| s.to_string())
3360            )),
3361        }
3362    }
3363
3364    pub async fn get_controller_metadata(
3365        &self,
3366        controller_address: CheetahString,
3367        timeout_millis: u64,
3368    ) -> ClientResult<GetMetaDataResponseHeader> {
3369        let request = self.create_remoting_command(RequestCode::ControllerGetMetadataInfo);
3370        let outcome = self
3371            .remoting_client
3372            .invoke_request(Some(&controller_address), request, timeout_millis)
3373            .await;
3374        let response = match outcome {
3375            Ok(OutboundRequestOutcome::Response(response)) => response,
3376            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3377                return Err(admin_request_error(
3378                    "get_controller_metadata",
3379                    RetryInput::Rejected(rejection),
3380                ));
3381            }
3382            Ok(OutboundRequestOutcome::Contract(contract)) => {
3383                return Err(admin_request_error(
3384                    "get_controller_metadata",
3385                    RetryInput::Contract(contract),
3386                ));
3387            }
3388            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3389        };
3390        match ResponseCode::from(response.code()) {
3391            ResponseCode::Success => response
3392                .decode_command_custom_header_fast::<GetMetaDataResponseHeader>()
3393                .map_err(|source| ClientError::response_process_source("decode GetMetaDataResponseHeader", source)),
3394            _ => Err(mq_client_err!(
3395                response.code(),
3396                response.remark().map_or("".to_string(), |s| s.to_string())
3397            )),
3398        }
3399    }
3400
3401    pub async fn get_controller_meta_data(
3402        &self,
3403        controller_address: CheetahString,
3404        timeout_millis: u64,
3405    ) -> ClientResult<GetMetaDataResponseHeader> {
3406        self.get_controller_metadata(controller_address, timeout_millis).await
3407    }
3408
3409    pub async fn get_in_sync_state_data(
3410        &self,
3411        controller_address: CheetahString,
3412        brokers: Vec<CheetahString>,
3413        timeout_millis: u64,
3414    ) -> ClientResult<BrokerReplicasInfo> {
3415        let controller_meta_data = self.get_controller_metadata(controller_address, timeout_millis).await?;
3416        let leader_address = controller_leader_address(controller_meta_data)?;
3417
3418        self.get_in_sync_state_data_at(leader_address, brokers, timeout_millis)
3419            .await
3420    }
3421
3422    /// Queries sync-state data at exactly the supplied Controller endpoint.
3423    ///
3424    /// Unlike [`Self::get_in_sync_state_data`], this method performs no metadata
3425    /// lookup and never follows a Controller-advertised leader address.
3426    pub async fn get_in_sync_state_data_at(
3427        &self,
3428        controller_address: CheetahString,
3429        brokers: Vec<CheetahString>,
3430        timeout_millis: u64,
3431    ) -> ClientResult<BrokerReplicasInfo> {
3432        let request = self.create_remoting_command(RequestCode::ControllerGetSyncStateData);
3433        let body =
3434            serde_json::to_vec(&brokers).map_err(|source| ClientError::internal("serialize broker names", source))?;
3435        let request = request.set_body(body);
3436        let outcome = self
3437            .remoting_client
3438            .invoke_request(Some(&controller_address), request, timeout_millis)
3439            .await;
3440        let response = match outcome {
3441            Ok(OutboundRequestOutcome::Response(response)) => response,
3442            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3443                return Err(admin_request_error(
3444                    "get_in_sync_state_data_at",
3445                    RetryInput::Rejected(rejection),
3446                ));
3447            }
3448            Ok(OutboundRequestOutcome::Contract(contract)) => {
3449                return Err(admin_request_error(
3450                    "get_in_sync_state_data_at",
3451                    RetryInput::Contract(contract),
3452                ));
3453            }
3454            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3455        };
3456        match ResponseCode::from(response.code()) {
3457            ResponseCode::Success => {
3458                if let Some(body) = response.body() {
3459                    serde_json::from_slice(body)
3460                        .map_err(|source| ClientError::response_process_source("decode BrokerReplicasInfo", source))
3461                } else {
3462                    Err(mq_client_err!(
3463                        "get_in_sync_state_data response body is empty".to_string()
3464                    ))
3465                }
3466            }
3467            _ => Err(mq_client_err!(
3468                response.code(),
3469                response.remark().map_or("".to_string(), |s| s.to_string())
3470            )),
3471        }
3472    }
3473
3474    pub async fn get_broker_epoch_cache(
3475        &self,
3476        broker_addr: CheetahString,
3477        timeout_millis: u64,
3478    ) -> ClientResult<EpochEntryCache> {
3479        let request = self.create_remoting_command(RequestCode::GetBrokerEpochCache);
3480        let outcome = self
3481            .remoting_client
3482            .invoke_request(Some(&broker_addr), request, timeout_millis)
3483            .await;
3484        let response = match outcome {
3485            Ok(OutboundRequestOutcome::Response(response)) => response,
3486            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3487                return Err(admin_request_error(
3488                    "get_broker_epoch_cache",
3489                    RetryInput::Rejected(rejection),
3490                ));
3491            }
3492            Ok(OutboundRequestOutcome::Contract(contract)) => {
3493                return Err(admin_request_error(
3494                    "get_broker_epoch_cache",
3495                    RetryInput::Contract(contract),
3496                ));
3497            }
3498            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3499        };
3500        match ResponseCode::from(response.code()) {
3501            ResponseCode::Success => {
3502                if let Some(body) = response.body() {
3503                    match EpochEntryCache::decode(body) {
3504                        Ok(value) => Ok(value),
3505                        Err(e) => Err(mq_client_err!(format!("decode EpochEntryCache failed: {}", e))),
3506                    }
3507                } else {
3508                    Err(mq_client_err!(
3509                        "get_broker_epoch_cache response body is empty".to_string()
3510                    ))
3511                }
3512            }
3513            _ => Err(mq_client_err!(
3514                response.code(),
3515                response.remark().map_or("".to_string(), |s| s.to_string())
3516            )),
3517        }
3518    }
3519
3520    pub async fn get_broker_ha_status(
3521        &self,
3522        broker_addr: CheetahString,
3523        timeout_millis: u64,
3524    ) -> ClientResult<HARuntimeInfo> {
3525        let request = self.create_remoting_command(RequestCode::GetBrokerHaStatus);
3526        let outcome = self
3527            .remoting_client
3528            .invoke_request(Some(&broker_addr), request, timeout_millis)
3529            .await;
3530        let response = match outcome {
3531            Ok(OutboundRequestOutcome::Response(response)) => response,
3532            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3533                return Err(admin_request_error(
3534                    "get_broker_ha_status",
3535                    RetryInput::Rejected(rejection),
3536                ));
3537            }
3538            Ok(OutboundRequestOutcome::Contract(contract)) => {
3539                return Err(admin_request_error(
3540                    "get_broker_ha_status",
3541                    RetryInput::Contract(contract),
3542                ));
3543            }
3544            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3545        };
3546        match ResponseCode::from(response.code()) {
3547            ResponseCode::Success => {
3548                if let Some(body) = response.body() {
3549                    serde_json::from_slice(body)
3550                        .map_err(|source| ClientError::response_process_source("decode HARuntimeInfo", source))
3551                } else {
3552                    Err(mq_client_err!("get_broker_ha_status response body is empty".to_string()))
3553                }
3554            }
3555            _ => Err(mq_client_err!(
3556                response.code(),
3557                response.remark().map_or("".to_string(), |s| s.to_string())
3558            )),
3559        }
3560    }
3561
3562    #[cfg(feature = "admin-mutation")]
3563    pub async fn update_and_get_group_forbidden(
3564        &self,
3565        broker_addr: &CheetahString,
3566        request_header: UpdateGroupForbiddenRequestHeader,
3567        timeout_millis: u64,
3568    ) -> ClientResult<GroupForbidden> {
3569        let request = self.create_request_command(RequestCode::UpdateAndGetGroupForbidden, request_header);
3570        let broker_addr_vip = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, broker_addr.as_str());
3571        let outcome = self
3572            .remoting_client
3573            .invoke_request(Some(&broker_addr_vip), request, timeout_millis)
3574            .await;
3575        let response = match outcome {
3576            Ok(OutboundRequestOutcome::Response(response)) => response,
3577            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3578                return Err(admin_request_error(
3579                    "update_and_get_group_forbidden",
3580                    RetryInput::Rejected(rejection),
3581                ));
3582            }
3583            Ok(OutboundRequestOutcome::Contract(contract)) => {
3584                return Err(admin_request_error(
3585                    "update_and_get_group_forbidden",
3586                    RetryInput::Contract(contract),
3587                ));
3588            }
3589            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3590        };
3591
3592        match ResponseCode::from(response.code()) {
3593            ResponseCode::Success => {
3594                let Some(body) = response.body() else {
3595                    return Err(mq_client_err!(
3596                        "update_and_get_group_forbidden response body is empty".to_string()
3597                    ));
3598                };
3599                SerdeJsonUtils::from_json_slice(body.as_ref())
3600                    .map_err(|source| ClientError::response_process_source("decode GroupForbidden", source))
3601            }
3602            _ => Err(client_broker_err!(
3603                response.code(),
3604                response.remark().map_or_else(String::new, |remark| remark.to_string()),
3605                broker_addr
3606            )),
3607        }
3608    }
3609
3610    pub async fn get_controller_config(
3611        &self,
3612        controller_address: CheetahString,
3613        timeout_millis: u64,
3614    ) -> ClientResult<HashMap<CheetahString, CheetahString>> {
3615        let request = self.create_remoting_command(RequestCode::GetControllerConfig);
3616        let outcome = self
3617            .remoting_client
3618            .invoke_request(Some(&controller_address), request, timeout_millis)
3619            .await;
3620        let response = match outcome {
3621            Ok(OutboundRequestOutcome::Response(response)) => response,
3622            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3623                return Err(admin_request_error(
3624                    "get_controller_config",
3625                    RetryInput::Rejected(rejection),
3626                ));
3627            }
3628            Ok(OutboundRequestOutcome::Contract(contract)) => {
3629                return Err(admin_request_error(
3630                    "get_controller_config",
3631                    RetryInput::Contract(contract),
3632                ));
3633            }
3634            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3635        };
3636
3637        match ResponseCode::from(response.code()) {
3638            ResponseCode::Success => {
3639                let body = response
3640                    .body()
3641                    .ok_or_else(|| mq_client_err!("Controller config response body is empty".to_string()))?;
3642                controller_config_from_response_body(body)
3643            }
3644            _ => Err(mq_client_err!(
3645                response.code(),
3646                response.remark().map_or("".to_string(), |s| s.to_string())
3647            )),
3648        }
3649    }
3650
3651    #[cfg(feature = "admin-mutation")]
3652    pub async fn update_controller_config(
3653        &self,
3654        properties: HashMap<CheetahString, CheetahString>,
3655        controllers: Vec<CheetahString>,
3656        timeout_millis: u64,
3657    ) -> ClientResult<()> {
3658        let body = mix_all::properties_to_string(&properties);
3659        if body.is_empty() || controllers.is_empty() {
3660            return Ok(());
3661        }
3662
3663        let request = self
3664            .create_remoting_command(RequestCode::UpdateControllerConfig)
3665            .set_body(body.to_string());
3666        let mut err_response = None;
3667        for controller_addr in controllers {
3668            let outcome = self
3669                .remoting_client
3670                .invoke_request(Some(&controller_addr), request.clone(), timeout_millis)
3671                .await;
3672            let response = match outcome {
3673                Ok(OutboundRequestOutcome::Response(response)) => response,
3674                Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3675                    return Err(admin_request_error(
3676                        "update_controller_config",
3677                        RetryInput::Rejected(rejection),
3678                    ));
3679                }
3680                Ok(OutboundRequestOutcome::Contract(contract)) => {
3681                    return Err(admin_request_error(
3682                        "update_controller_config",
3683                        RetryInput::Contract(contract),
3684                    ));
3685                }
3686                Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3687            };
3688            match ResponseCode::from(response.code()) {
3689                ResponseCode::Success => {}
3690                _ => err_response = Some(response),
3691            }
3692        }
3693
3694        if let Some(err_response) = err_response {
3695            return Err(mq_client_err!(
3696                err_response.code(),
3697                err_response
3698                    .remark()
3699                    .map_or_else(String::new, |remark| remark.to_string())
3700            ));
3701        }
3702        Ok(())
3703    }
3704
3705    #[cfg(feature = "admin-mutation")]
3706    pub async fn elect_master(
3707        &self,
3708        controller_addr: CheetahString,
3709        cluster_name: CheetahString,
3710        broker_name: CheetahString,
3711        broker_id: Option<u64>,
3712        timeout_millis: u64,
3713    ) -> ClientResult<(ElectMasterResponseHeader, BrokerMemberGroup)> {
3714        let controller_meta_data = self.get_controller_metadata(controller_addr, timeout_millis).await?;
3715        let leader_address = controller_leader_address(controller_meta_data)?;
3716        let designate_elect = broker_id.is_some();
3717        let broker_id = match broker_id {
3718            Some(broker_id) => i64::try_from(broker_id).map_err(ClientError::illegal_argument_source)?,
3719            None => -1,
3720        };
3721        let request_header = ElectMasterRequestHeader::new(
3722            cluster_name,
3723            broker_name,
3724            broker_id,
3725            designate_elect,
3726            rocketmq_runtime::common::time_utils::current_millis(),
3727        );
3728        let request = self.create_request_command(RequestCode::ControllerElectMaster, request_header);
3729        let outcome = self
3730            .remoting_client
3731            .invoke_request(Some(&leader_address), request, timeout_millis)
3732            .await;
3733        let response = match outcome {
3734            Ok(OutboundRequestOutcome::Response(response)) => response,
3735            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3736                return Err(admin_request_error("elect_master", RetryInput::Rejected(rejection)));
3737            }
3738            Ok(OutboundRequestOutcome::Contract(contract)) => {
3739                return Err(admin_request_error("elect_master", RetryInput::Contract(contract)));
3740            }
3741            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3742        };
3743
3744        match ResponseCode::from(response.code()) {
3745            ResponseCode::Success => {
3746                let response_header = response
3747                    .decode_command_custom_header_fast::<ElectMasterResponseHeader>()
3748                    .map_err(|source| {
3749                        ClientError::response_process_source("decode ElectMasterResponseHeader", source)
3750                    })?;
3751                let broker_member_group = response
3752                    .body()
3753                    .ok_or_else(|| mq_client_err!("elect_master response body is empty".to_string()))
3754                    .and_then(|body| {
3755                        serde_json::from_slice::<BrokerMemberGroup>(body)
3756                            .map_err(|source| ClientError::response_process_source("decode BrokerMemberGroup", source))
3757                    })?;
3758                Ok((response_header, broker_member_group))
3759            }
3760            _ => Err(mq_client_err!(
3761                response.code(),
3762                response.remark().map_or_else(String::new, |remark| remark.to_string())
3763            )),
3764        }
3765    }
3766
3767    #[cfg(feature = "admin-mutation")]
3768    pub async fn clean_controller_broker_data(
3769        &self,
3770        controller_addr: CheetahString,
3771        cluster_name: CheetahString,
3772        broker_name: CheetahString,
3773        broker_controller_ids_to_clean: Option<CheetahString>,
3774        clean_living_broker: bool,
3775        timeout_millis: u64,
3776    ) -> ClientResult<()> {
3777        let controller_meta_data = self.get_controller_metadata(controller_addr, timeout_millis).await?;
3778        let leader_address = controller_leader_address(controller_meta_data)?;
3779        let request_header = CleanBrokerDataRequestHeader {
3780            cluster_name: if cluster_name.is_empty() {
3781                None
3782            } else {
3783                Some(cluster_name)
3784            },
3785            broker_name,
3786            broker_controller_ids_to_clean,
3787            clean_living_broker,
3788            ..Default::default()
3789        };
3790        let request = self.create_request_command(RequestCode::CleanBrokerData, request_header);
3791        let outcome = self
3792            .remoting_client
3793            .invoke_request(Some(&leader_address), request, timeout_millis)
3794            .await;
3795        let response = match outcome {
3796            Ok(OutboundRequestOutcome::Response(response)) => response,
3797            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3798                return Err(admin_request_error(
3799                    "clean_controller_broker_data",
3800                    RetryInput::Rejected(rejection),
3801                ));
3802            }
3803            Ok(OutboundRequestOutcome::Contract(contract)) => {
3804                return Err(admin_request_error(
3805                    "clean_controller_broker_data",
3806                    RetryInput::Contract(contract),
3807                ));
3808            }
3809            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3810        };
3811
3812        match ResponseCode::from(response.code()) {
3813            ResponseCode::Success => Ok(()),
3814            _ => Err(mq_client_err!(
3815                response.code(),
3816                response.remark().map_or_else(String::new, |remark| remark.to_string())
3817            )),
3818        }
3819    }
3820
3821    #[cfg(feature = "admin-mutation")]
3822    pub async fn update_cold_data_flow_ctr_group_config(
3823        &self,
3824        broker_addr: CheetahString,
3825        properties: HashMap<CheetahString, CheetahString>,
3826        timeout_millis: u64,
3827    ) -> ClientResult<()> {
3828        let body = mix_all::properties_to_string(&properties);
3829        if body.is_empty() {
3830            return Ok(());
3831        }
3832
3833        let request = self.create_remoting_command(RequestCode::UpdateColdDataFlowCtrConfig);
3834        let request = request.set_body(body.to_string());
3835        let broker_addr =
3836            mix_all::broker_vip_channel(self.client_config.is_vip_channel_enabled(), broker_addr.as_str());
3837        let outcome = self
3838            .remoting_client
3839            .invoke_request(Some(broker_addr).as_ref(), request, timeout_millis)
3840            .await;
3841        let response = match outcome {
3842            Ok(OutboundRequestOutcome::Response(response)) => response,
3843            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3844                return Err(admin_request_error(
3845                    "update_cold_data_flow_ctr_group_config",
3846                    RetryInput::Rejected(rejection),
3847                ));
3848            }
3849            Ok(OutboundRequestOutcome::Contract(contract)) => {
3850                return Err(admin_request_error(
3851                    "update_cold_data_flow_ctr_group_config",
3852                    RetryInput::Contract(contract),
3853                ));
3854            }
3855            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3856        };
3857
3858        match ResponseCode::from(response.code()) {
3859            ResponseCode::Success => Ok(()),
3860            _ => Err(mq_client_err!(
3861                response.code(),
3862                response.remark().map(|s| s.to_string()).unwrap_or_default()
3863            )),
3864        }
3865    }
3866
3867    #[cfg(feature = "admin-mutation")]
3868    pub async fn remove_cold_data_flow_ctr_group_config(
3869        &self,
3870        broker_addr: CheetahString,
3871        consumer_group: CheetahString,
3872        timeout_millis: u64,
3873    ) -> ClientResult<()> {
3874        if consumer_group.is_empty() {
3875            return Ok(());
3876        }
3877
3878        let request = self
3879            .create_request_command(RequestCode::RemoveColdDataFlowCtrConfig, EmptyHeader {})
3880            .set_body(consumer_group.to_string());
3881        let broker_addr_vip =
3882            mix_all::broker_vip_channel(self.client_config.is_vip_channel_enabled(), broker_addr.as_str());
3883        let outcome = self
3884            .remoting_client
3885            .invoke_request(Some(&broker_addr_vip), request, timeout_millis)
3886            .await;
3887        let response = match outcome {
3888            Ok(OutboundRequestOutcome::Response(response)) => response,
3889            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3890                return Err(admin_request_error(
3891                    "remove_cold_data_flow_ctr_group_config",
3892                    RetryInput::Rejected(rejection),
3893                ));
3894            }
3895            Ok(OutboundRequestOutcome::Contract(contract)) => {
3896                return Err(admin_request_error(
3897                    "remove_cold_data_flow_ctr_group_config",
3898                    RetryInput::Contract(contract),
3899                ));
3900            }
3901            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3902        };
3903
3904        match ResponseCode::from(response.code()) {
3905            ResponseCode::Success => Ok(()),
3906            _ => Err(client_broker_err!(
3907                response.code(),
3908                response.remark().map_or_else(String::new, |remark| remark.to_string()),
3909                &broker_addr
3910            )),
3911        }
3912    }
3913
3914    pub async fn get_cold_data_flow_ctr_info(
3915        &self,
3916        broker_addr: CheetahString,
3917        timeout_millis: u64,
3918    ) -> ClientResult<CheetahString> {
3919        let request = self.create_request_command(RequestCode::GetColdDataFlowCtrInfo, EmptyHeader {});
3920        let outcome = self
3921            .remoting_client
3922            .invoke_request(Some(&broker_addr), request, timeout_millis)
3923            .await;
3924        let response = match outcome {
3925            Ok(OutboundRequestOutcome::Response(response)) => response,
3926            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3927                return Err(admin_request_error(
3928                    "get_cold_data_flow_ctr_info",
3929                    RetryInput::Rejected(rejection),
3930                ));
3931            }
3932            Ok(OutboundRequestOutcome::Contract(contract)) => {
3933                return Err(admin_request_error(
3934                    "get_cold_data_flow_ctr_info",
3935                    RetryInput::Contract(contract),
3936                ));
3937            }
3938            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3939        };
3940
3941        match ResponseCode::from(response.code()) {
3942            ResponseCode::Success => Ok(response
3943                .body()
3944                .filter(|body| !body.is_empty())
3945                .map(|body| CheetahString::from_string(String::from_utf8_lossy(body.as_ref()).into_owned()))
3946                .unwrap_or_default()),
3947            _ => Err(client_broker_err!(
3948                response.code(),
3949                response.remark().map_or_else(String::new, |remark| remark.to_string()),
3950                &broker_addr
3951            )),
3952        }
3953    }
3954
3955    #[cfg(feature = "admin-mutation")]
3956    pub async fn set_commit_log_read_ahead_mode(
3957        &self,
3958        broker_addr: CheetahString,
3959        mode: CheetahString,
3960        timeout_millis: u64,
3961    ) -> ClientResult<CheetahString> {
3962        let mut request = self.create_request_command(RequestCode::SetCommitlogReadMode, EmptyHeader {});
3963        request.ensure_ext_fields_initialized();
3964        request.add_ext_field(file_readahead_mode::READ_AHEAD_MODE, mode);
3965        let outcome = self
3966            .remoting_client
3967            .invoke_request(Some(&broker_addr), request, timeout_millis)
3968            .await;
3969        let response = match outcome {
3970            Ok(OutboundRequestOutcome::Response(response)) => response,
3971            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
3972                return Err(admin_request_error(
3973                    "set_commit_log_read_ahead_mode",
3974                    RetryInput::Rejected(rejection),
3975                ));
3976            }
3977            Ok(OutboundRequestOutcome::Contract(contract)) => {
3978                return Err(admin_request_error(
3979                    "set_commit_log_read_ahead_mode",
3980                    RetryInput::Contract(contract),
3981                ));
3982            }
3983            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
3984        };
3985
3986        match ResponseCode::from(response.code()) {
3987            ResponseCode::Success => Ok(response.remark().cloned().unwrap_or_default()),
3988            _ => Err(client_broker_err!(
3989                response.code(),
3990                response.remark().map_or_else(String::new, |remark| remark.to_string()),
3991                &broker_addr
3992            )),
3993        }
3994    }
3995
3996    #[cfg(feature = "admin-mutation")]
3997    pub async fn export_pop_record(&self, broker_addr: CheetahString, timeout_millis: u64) -> ClientResult<()> {
3998        let request = self.create_request_command(RequestCode::PopRollback, EmptyHeader {});
3999        let outcome = self
4000            .remoting_client
4001            .invoke_request(Some(&broker_addr), request, timeout_millis)
4002            .await;
4003        let response = match outcome {
4004            Ok(OutboundRequestOutcome::Response(response)) => response,
4005            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
4006                return Err(admin_request_error(
4007                    "export_pop_record",
4008                    RetryInput::Rejected(rejection),
4009                ));
4010            }
4011            Ok(OutboundRequestOutcome::Contract(contract)) => {
4012                return Err(admin_request_error("export_pop_record", RetryInput::Contract(contract)));
4013            }
4014            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
4015        };
4016
4017        if ResponseCode::from(response.code()) == ResponseCode::Success {
4018            return Ok(());
4019        }
4020
4021        Err(client_broker_err!(
4022            response.code(),
4023            response.remark().map_or_else(String::new, |remark| remark.to_string()),
4024            &broker_addr
4025        ))
4026    }
4027
4028    pub(crate) async fn get_all_subscription_group(
4029        &self,
4030        addr: &CheetahString,
4031        timeout_millis: u64,
4032    ) -> ClientResult<rocketmq_protocol::protocol::body::subscription_group_wrapper::SubscriptionGroupWrapper> {
4033        self.get_all_subscription_group_config(addr, timeout_millis).await
4034    }
4035
4036    pub(crate) async fn get_subscription_group_config(
4037        &self,
4038        addr: &CheetahString,
4039        group: CheetahString,
4040        timeout_millis: u64,
4041    ) -> ClientResult<rocketmq_protocol::protocol::subscription::subscription_group_config::SubscriptionGroupConfig>
4042    {
4043        let request = self.create_request_command(
4044            RequestCode::GetSubscriptionGroupConfig,
4045            rocketmq_protocol::protocol::header::get_subscription_group_config_request_header::GetSubscriptionGroupConfigRequestHeader {
4046                group,
4047                rpc_request_header: None,
4048            },
4049        );
4050        let outcome = self
4051            .remoting_client
4052            .invoke_request(Some(addr), request, timeout_millis)
4053            .await;
4054        let mut response = match outcome {
4055            Ok(OutboundRequestOutcome::Response(response)) => response,
4056            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
4057                return Err(admin_request_error(
4058                    "get_subscription_group_config",
4059                    RetryInput::Rejected(rejection),
4060                ));
4061            }
4062            Ok(OutboundRequestOutcome::Contract(contract)) => {
4063                return Err(admin_request_error(
4064                    "get_subscription_group_config",
4065                    RetryInput::Contract(contract),
4066                ));
4067            }
4068            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
4069        };
4070        if ResponseCode::from(response.code()) == ResponseCode::Success {
4071            if let Some(body) = response.take_body() {
4072                return rocketmq_protocol::protocol::subscription::subscription_group_config::SubscriptionGroupConfig::decode(
4073                    body.as_ref(),
4074                )
4075                .map_err(ClientError::from);
4076            }
4077        }
4078        Err(mq_client_err!(
4079            response.code(),
4080            response.remark().map_or("".to_string(), |s| s.to_string())
4081        ))
4082    }
4083
4084    #[cfg(feature = "admin-mutation")]
4085    pub(crate) async fn create_subscription_group(
4086        &self,
4087        addr: &CheetahString,
4088        config: &rocketmq_protocol::protocol::subscription::subscription_group_config::SubscriptionGroupConfig,
4089        timeout_millis: u64,
4090    ) -> ClientResult<()> {
4091        let request = self
4092            .create_request_command(RequestCode::UpdateAndCreateSubscriptionGroup, EmptyHeader {})
4093            .set_body(config.encode()?);
4094        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
4095        let outcome = self
4096            .remoting_client
4097            .invoke_request(Some(&broker_addr), request, timeout_millis)
4098            .await;
4099        let response = match outcome {
4100            Ok(OutboundRequestOutcome::Response(response)) => response,
4101            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
4102                return Err(admin_request_error(
4103                    "create_subscription_group",
4104                    RetryInput::Rejected(rejection),
4105                ));
4106            }
4107            Ok(OutboundRequestOutcome::Contract(contract)) => {
4108                return Err(admin_request_error(
4109                    "create_subscription_group",
4110                    RetryInput::Contract(contract),
4111                ));
4112            }
4113            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
4114        };
4115
4116        match ResponseCode::from(response.code()) {
4117            ResponseCode::Success => Ok(()),
4118            _ => Err(mq_client_err!(
4119                response.code(),
4120                response.remark().map_or_else(String::new, |remark| remark.to_string())
4121            )),
4122        }
4123    }
4124
4125    pub(crate) async fn get_consumer_running_info(
4126        &self,
4127        addr: &CheetahString,
4128        consumer_group: CheetahString,
4129        client_id: CheetahString,
4130        jstack: bool,
4131        timeout_millis: u64,
4132    ) -> ClientResult<rocketmq_protocol::protocol::body::consumer_running_info::ConsumerRunningInfo> {
4133        let request_header = GetConsumerRunningInfoRequestHeader {
4134            consumer_group,
4135            client_id,
4136            jstack_enable: jstack,
4137            rpc_request_header: None,
4138        };
4139        let request = self.create_request_command(RequestCode::GetConsumerRunningInfo, request_header);
4140        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
4141        let outcome = self
4142            .remoting_client
4143            .invoke_request(Some(&broker_addr), request, timeout_millis)
4144            .await;
4145        let mut response = match outcome {
4146            Ok(OutboundRequestOutcome::Response(response)) => response,
4147            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
4148                return Err(admin_request_error(
4149                    "get_consumer_running_info",
4150                    RetryInput::Rejected(rejection),
4151                ));
4152            }
4153            Ok(OutboundRequestOutcome::Contract(contract)) => {
4154                return Err(admin_request_error(
4155                    "get_consumer_running_info",
4156                    RetryInput::Contract(contract),
4157                ));
4158            }
4159            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
4160        };
4161
4162        match ResponseCode::from(response.code()) {
4163            ResponseCode::Success => {
4164                let Some(body) = response.take_body() else {
4165                    return Err(mq_client_err!(
4166                        "get_consumer_running_info response body is empty".to_string()
4167                    ));
4168                };
4169                rocketmq_protocol::protocol::body::consumer_running_info::ConsumerRunningInfo::decode(&body)
4170                    .map_err(|source| ClientError::response_process_source("decode get_consumer_running_info", source))
4171            }
4172            _ => Err(mq_client_err!(
4173                response.code(),
4174                response.remark().map_or_else(String::new, |remark| remark.to_string())
4175            )),
4176        }
4177    }
4178
4179    #[cfg(feature = "admin-mutation")]
4180    pub(crate) async fn consume_message_directly(
4181        &self,
4182        client_addr: &CheetahString,
4183        request_header: ConsumeMessageDirectlyResultRequestHeader,
4184        message: &MessageExt,
4185        timeout_millis: u64,
4186    ) -> ClientResult<rocketmq_protocol::protocol::body::consume_message_directly_result::ConsumeMessageDirectlyResult>
4187    {
4188        let body = MessageDecoder::encode(message, false)?;
4189        let request = self
4190            .create_request_command(RequestCode::ConsumeMessageDirectly, request_header)
4191            .set_body(body.to_vec());
4192        let outcome = self
4193            .remoting_client
4194            .invoke_request(Some(client_addr), request, timeout_millis)
4195            .await;
4196        let mut response = match outcome {
4197            Ok(OutboundRequestOutcome::Response(response)) => response,
4198            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
4199                return Err(admin_request_error(
4200                    "consume_message_directly",
4201                    RetryInput::Rejected(rejection),
4202                ));
4203            }
4204            Ok(OutboundRequestOutcome::Contract(contract)) => {
4205                return Err(admin_request_error(
4206                    "consume_message_directly",
4207                    RetryInput::Contract(contract),
4208                ));
4209            }
4210            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
4211        };
4212
4213        match ResponseCode::from(response.code()) {
4214            ResponseCode::Success => {
4215                if let Some(body) = response.take_body() {
4216                    rocketmq_protocol::protocol::body::consume_message_directly_result::ConsumeMessageDirectlyResult::decode(
4217                        body.as_ref(),
4218                    )
4219                    .map_err(ClientError::from)
4220                } else {
4221                    Err(mq_client_err!(
4222                        "consume_message_directly response body is empty".to_string()
4223                    ))
4224                }
4225            }
4226            _ => Err(mq_client_err!(
4227                response.code(),
4228                response.remark().map_or_else(String::new, |remark| remark.to_string())
4229            )),
4230        }
4231    }
4232
4233    pub async fn view_message(
4234        &self,
4235        addr: &CheetahString,
4236        request_header: ViewMessageRequestHeader,
4237        timeout_millis: u64,
4238    ) -> ClientResult<MessageExt> {
4239        let request = self.create_request_command(RequestCode::ViewMessageById, request_header);
4240        let outcome = self
4241            .remoting_client
4242            .invoke_request(Some(addr), request, timeout_millis)
4243            .await;
4244        let response = match outcome {
4245            Ok(OutboundRequestOutcome::Response(response)) => response,
4246            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
4247                return Err(admin_request_error("view_message", RetryInput::Rejected(rejection)));
4248            }
4249            Ok(OutboundRequestOutcome::Contract(contract)) => {
4250                return Err(admin_request_error("view_message", RetryInput::Contract(contract)));
4251            }
4252            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
4253        };
4254
4255        match ResponseCode::from(response.code()) {
4256            ResponseCode::Success => {
4257                if let Some(body) = response.get_body() {
4258                    let mut bytes = body.clone();
4259                    let body_len = bytes.len();
4260                    MessageDecoder::decode(&mut bytes, true, true, false, false, false).ok_or_else(|| {
4261                        mq_client_err!(format!(
4262                            "Failed to decode message from view_message response body: body_len={}, possible causes: \
4263                             CRC check failed or malformed message data",
4264                            body_len
4265                        ))
4266                    })
4267                } else {
4268                    Err(mq_client_err!("view_message response body is empty".to_string()))
4269                }
4270            }
4271            _ => Err(mq_client_err!(
4272                response.code(),
4273                response.remark().map_or("".to_string(), |s| s.to_string())
4274            )),
4275        }
4276    }
4277
4278    #[cfg(feature = "admin-mutation")]
4279    pub(crate) async fn resume_check_half_message(
4280        &self,
4281        addr: &CheetahString,
4282        topic: CheetahString,
4283        msg_id: CheetahString,
4284        timeout_millis: u64,
4285    ) -> ClientResult<bool> {
4286        let request_header = ResumeCheckHalfMessageRequestHeader {
4287            topic,
4288            msg_id: Some(msg_id),
4289        };
4290        let request = self.create_request_command(RequestCode::ResumeCheckHalfMessage, request_header);
4291        let broker_addr = mix_all::broker_vip_channel(self.client_config.vip_channel_enabled, addr.as_str());
4292        let outcome = self
4293            .remoting_client
4294            .invoke_request(Some(&broker_addr), request, timeout_millis)
4295            .await;
4296        let response = match outcome {
4297            Ok(OutboundRequestOutcome::Response(response)) => response,
4298            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
4299                return Err(admin_request_error(
4300                    "resume_check_half_message",
4301                    RetryInput::Rejected(rejection),
4302                ));
4303            }
4304            Ok(OutboundRequestOutcome::Contract(contract)) => {
4305                return Err(admin_request_error(
4306                    "resume_check_half_message",
4307                    RetryInput::Contract(contract),
4308                ));
4309            }
4310            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
4311        };
4312
4313        match ResponseCode::from(response.code()) {
4314            ResponseCode::Success => Ok(true),
4315            _ => {
4316                error!(
4317                    "Failed to resume half message check logic. Remark={}",
4318                    response.remark().map_or("", |remark| remark)
4319                );
4320                Ok(false)
4321            }
4322        }
4323    }
4324
4325    #[cfg(feature = "admin-mutation")]
4326    pub(crate) async fn switch_timer_engine(
4327        &self,
4328        broker_addr: &CheetahString,
4329        engine_type: CheetahString,
4330        timeout_millis: u64,
4331    ) -> ClientResult<()> {
4332        let mut request = self.create_request_command(RequestCode::SwitchTimerEngine, EmptyHeader {});
4333        request.ensure_ext_fields_initialized();
4334        request.add_ext_field(MessageConst::TIMER_ENGINE_TYPE, engine_type);
4335        let outcome = self
4336            .remoting_client
4337            .invoke_request(Some(broker_addr), request, timeout_millis)
4338            .await;
4339        let response = match outcome {
4340            Ok(OutboundRequestOutcome::Response(response)) => response,
4341            Ok(OutboundRequestOutcome::Rejected(rejection)) => {
4342                return Err(admin_request_error(
4343                    "switch_timer_engine",
4344                    RetryInput::Rejected(rejection),
4345                ));
4346            }
4347            Ok(OutboundRequestOutcome::Contract(contract)) => {
4348                return Err(admin_request_error(
4349                    "switch_timer_engine",
4350                    RetryInput::Contract(contract),
4351                ));
4352            }
4353            Err(error) => return Err(ClientError::from_shared(error.into_shared_error())),
4354        };
4355
4356        if ResponseCode::from(response.code()) == ResponseCode::Success {
4357            return Ok(());
4358        }
4359
4360        Err(client_broker_err!(
4361            response.code(),
4362            response.remark().map_or("".to_string(), |s| s.to_string()),
4363            broker_addr
4364        ))
4365    }
4366
4367    pub(super) fn build_queue_offset_sorted_map(
4368        topic: &str,
4369        msg_found_list: &[MessageExt],
4370    ) -> ClientResult<HashMap<String, Vec<i64>>> {
4371        let mut sort_map: HashMap<String, Vec<i64>> = HashMap::with_capacity(16);
4372        for message_ext in msg_found_list {
4373            let key: String;
4374            let dispatch = message_ext
4375                .property(&CheetahString::from_static_str(
4376                    MessageConst::PROPERTY_INNER_MULTI_DISPATCH,
4377                ))
4378                .unwrap_or_default();
4379            if mix_all::is_lmq(Some(topic)) && message_ext.reconsume_times() == 0 && !dispatch.is_empty() {
4380                // process LMQ
4381                let queues: Vec<&str> = dispatch.split_str(mix_all::MULTI_DISPATCH_QUEUE_SPLITTER).collect();
4382                let data = message_ext
4383                    .property(&CheetahString::from_static_str(
4384                        MessageConst::PROPERTY_INNER_MULTI_QUEUE_OFFSET,
4385                    ))
4386                    .unwrap_or_default();
4387                let queue_offsets: Vec<&str> = data.split_str(mix_all::MULTI_DISPATCH_QUEUE_SPLITTER).collect();
4388                // LMQ topic has only 1 queue, which queue id is 0
4389                key = ExtraInfoUtil::get_start_offset_info_map_key(topic, mix_all::LMQ_QUEUE_ID as i64);
4390                let Some(position) = queues.iter().position(|&q| q == topic) else {
4391                    warn!(
4392                        "LMQ dispatch queue does not contain topic={}, dispatch={}",
4393                        topic, dispatch
4394                    );
4395                    continue;
4396                };
4397                let Some(offset_value) = queue_offsets.get(position) else {
4398                    warn!(
4399                        "LMQ queue offset is missing for topic={}, dispatch={}, offsets={}",
4400                        topic, dispatch, data
4401                    );
4402                    continue;
4403                };
4404                let Ok(offset) = offset_value.parse::<i64>() else {
4405                    warn!(
4406                        "LMQ queue offset is invalid for topic={}, offset={}",
4407                        topic, offset_value
4408                    );
4409                    continue;
4410                };
4411                sort_map
4412                    .entry(key)
4413                    .or_insert_with(|| Vec::with_capacity(4))
4414                    .push(offset);
4415                continue;
4416            }
4417            // Value of POP_CK is used to determine whether it is a pop retry,
4418            // cause topic could be rewritten by broker.
4419            key = ExtraInfoUtil::get_start_offset_info_map_key_with_pop_ck(
4420                message_ext.topic(),
4421                message_ext
4422                    .property(&CheetahString::from_static_str(MessageConst::PROPERTY_POP_CK))
4423                    .clone()
4424                    .as_ref()
4425                    .map(|item| item.as_str()),
4426                message_ext.queue_id() as i64,
4427            )?;
4428            sort_map
4429                .entry(key)
4430                .or_insert_with(|| Vec::with_capacity(4))
4431                .push(message_ext.queue_offset());
4432        }
4433        Ok(sort_map)
4434    }
4435}
4436
4437#[cfg(any(feature = "admin-read", feature = "admin-mutation"))]
4438fn decode_proxy_drain_response(
4439    response: RemotingCommand,
4440    operation: &'static str,
4441) -> ClientResult<ProxyDrainStateResponseBody> {
4442    if ResponseCode::from(response.code()) != ResponseCode::Success {
4443        return Err(mq_client_err!(
4444            response.code(),
4445            response.remark().map_or_else(String::new, |remark| remark.to_string())
4446        ));
4447    }
4448    let body = response
4449        .body()
4450        .ok_or_else(|| ClientError::response_process_failed(operation, "Proxy drain response body is missing"))?;
4451    let state = ProxyDrainStateResponseBody::decode(body.as_ref())
4452        .map_err(|error| ClientError::response_process_source(operation, error))?;
4453    if state.schema_version != PROXY_DRAIN_SCHEMA_VERSION {
4454        return Err(ClientError::response_process_failed(
4455            operation,
4456            format!(
4457                "unsupported Proxy drain schema '{}'; expected '{PROXY_DRAIN_SCHEMA_VERSION}'",
4458                state.schema_version
4459            ),
4460        ));
4461    }
4462    Ok(state)
4463}
4464
4465pub(super) fn pop_msg_queue_offset_for_index(
4466    queue_id_key: &str,
4467    queue_offset: i64,
4468    sort_map: &HashMap<String, Vec<i64>>,
4469    msg_offset_info: &HashMap<String, Vec<i64>>,
4470) -> Option<i64> {
4471    let index = sort_map
4472        .get(queue_id_key)?
4473        .iter()
4474        .position(|&offset| offset == queue_offset)?;
4475    msg_offset_info
4476        .get(queue_id_key)
4477        .and_then(|offsets| offsets.get(index))
4478        .copied()
4479}
4480
4481pub(super) fn empty_broker_member_group(
4482    cluster_name: &CheetahString,
4483    broker_name: &CheetahString,
4484) -> BrokerMemberGroup {
4485    BrokerMemberGroup::new(cluster_name.clone(), broker_name.clone())
4486}
4487
4488pub(super) fn broker_member_group_from_route_data(
4489    cluster_name: &CheetahString,
4490    broker_name: &CheetahString,
4491    topic_route_data: &TopicRouteData,
4492) -> BrokerMemberGroup {
4493    let mut broker_member_group = empty_broker_member_group(cluster_name, broker_name);
4494    if let Some(broker_data) = topic_route_data
4495        .broker_datas
4496        .iter()
4497        .find(|broker_data| broker_data.cluster() == cluster_name.as_str() && broker_data.broker_name() == broker_name)
4498    {
4499        broker_member_group
4500            .broker_addrs
4501            .extend(broker_data.broker_addrs().clone());
4502    }
4503    broker_member_group
4504}
4505
4506pub(super) fn cluster_names_for_topic_route(
4507    cluster_info: &ClusterInfo,
4508    topic_route_data: &TopicRouteData,
4509) -> HashSet<CheetahString> {
4510    let mut cluster_names = HashSet::new();
4511    let Some(cluster_addr_table) = cluster_info.cluster_addr_table.as_ref() else {
4512        return cluster_names;
4513    };
4514
4515    for broker_data in &topic_route_data.broker_datas {
4516        cluster_names.extend(
4517            cluster_addr_table
4518                .iter()
4519                .filter(|(_, broker_names)| broker_names.contains(broker_data.broker_name()))
4520                .map(|(cluster_name, _)| cluster_name.clone()),
4521        );
4522    }
4523
4524    cluster_names
4525}
4526
4527pub(super) fn encode_topic_attributes_like_java(
4528    attributes: &HashMap<CheetahString, CheetahString>,
4529) -> Option<CheetahString> {
4530    if attributes.is_empty() {
4531        return None;
4532    }
4533
4534    let encoded = AttributeParser::parse_to_string(
4535        &attributes
4536            .iter()
4537            .map(|(key, value)| (key.to_string(), value.to_string()))
4538            .collect::<HashMap<String, String>>(),
4539    );
4540    if encoded.is_empty() {
4541        None
4542    } else {
4543        Some(CheetahString::from_string(encoded))
4544    }
4545}
4546
4547pub(super) fn topic_config_u32_to_java_i32(_field_name: &'static str, value: u32) -> ClientResult<i32> {
4548    i32::try_from(value).map_err(ClientError::illegal_argument_source)
4549}
4550
4551pub(super) fn create_topic_request_header_like_java(
4552    default_topic: CheetahString,
4553    topic_config: &TopicConfig,
4554) -> ClientResult<CreateTopicRequestHeader> {
4555    Validators::check_topic_config(topic_config)?;
4556    let topic = topic_config
4557        .topic_name
4558        .clone()
4559        .ok_or_else(|| mq_client_err!("topicConfig.topicName is required".to_string()))?;
4560
4561    Ok(CreateTopicRequestHeader {
4562        topic,
4563        default_topic,
4564        read_queue_nums: topic_config_u32_to_java_i32("readQueueNums", topic_config.read_queue_nums)?,
4565        write_queue_nums: topic_config_u32_to_java_i32("writeQueueNums", topic_config.write_queue_nums)?,
4566        perm: topic_config_u32_to_java_i32("perm", topic_config.perm)?,
4567        topic_filter_type: CheetahString::from_static_str(topic_config.topic_filter_type.as_str()),
4568        topic_sys_flag: Some(topic_config_u32_to_java_i32(
4569            "topicSysFlag",
4570            topic_config.topic_sys_flag,
4571        )?),
4572        order: topic_config.order,
4573        attributes: encode_topic_attributes_like_java(&topic_config.attributes),
4574        force: None,
4575        topic_request_header: None,
4576    })
4577}
4578
4579pub(super) fn query_correction_offset_request(
4580    command_factory: &RemotingCommandFactory,
4581    topic: CheetahString,
4582    group: CheetahString,
4583    filter_groups: Option<Vec<CheetahString>>,
4584) -> RemotingCommand {
4585    let filter_groups = filter_groups.map(|groups| {
4586        CheetahString::from_string(
4587            groups
4588                .into_iter()
4589                .map(|group| group.to_string())
4590                .collect::<Vec<_>>()
4591                .join(","),
4592        )
4593    });
4594    let request_header = QueryCorrectionOffsetHeader {
4595        filter_groups,
4596        compare_group: group,
4597        topic,
4598        topic_request_header: None,
4599    };
4600    command_factory.create_request_command(RequestCode::QueryCorrectionOffset, request_header)
4601}
4602
4603pub(super) fn split_lite_dispatch_value(value: &str) -> Vec<&str> {
4604    value
4605        .split(mix_all::MULTI_DISPATCH_QUEUE_SPLITTER)
4606        .filter(|segment| !segment.is_empty())
4607        .collect()
4608}
4609
4610pub(super) fn parse_lite_order_count_info_like_java(
4611    order_count_info: Option<&CheetahString>,
4612    msg_count: usize,
4613) -> Option<Vec<i32>> {
4614    let order_count_info = order_count_info?.as_str();
4615    if order_count_info.trim().is_empty() {
4616        return None;
4617    }
4618    let infos = order_count_info.split(';').collect::<Vec<_>>();
4619    if infos.len() != msg_count {
4620        return None;
4621    }
4622    Some(infos.into_iter().map(parse_lite_order_count_like_java).collect())
4623}
4624
4625pub(super) fn parse_lite_order_count_like_java(info: &str) -> i32 {
4626    if info.trim().is_empty() {
4627        return 0;
4628    }
4629    if !info.contains(MessageConst::KEY_SEPARATOR) {
4630        return info.parse::<i32>().unwrap_or_default();
4631    }
4632    let split = info.split(MessageConst::KEY_SEPARATOR).collect::<Vec<_>>();
4633    if split.len() != 3 {
4634        return 0;
4635    }
4636    split[2].parse::<i32>().unwrap_or_default()
4637}
4638
4639pub(super) fn filter_retry_topics_like_java(topic_list: &mut TopicList, contain_retry: bool) {
4640    if contain_retry {
4641        return;
4642    }
4643    topic_list
4644        .topic_list
4645        .retain(|topic| !topic.as_str().starts_with(mix_all::RETRY_GROUP_TOPIC_PREFIX));
4646}
4647
4648pub(super) fn controller_leader_address(header: GetMetaDataResponseHeader) -> ClientResult<CheetahString> {
4649    header
4650        .controller_leader_address
4651        .ok_or_else(|| mq_client_err!("Controller leader address is not available".to_string()))
4652}
4653
4654pub(super) fn controller_config_from_response_body(body: &[u8]) -> ClientResult<HashMap<CheetahString, CheetahString>> {
4655    let body_str = String::from_utf8_lossy(body);
4656    mix_all::string_to_properties(body_str.as_ref())
4657        .ok_or_else(|| mq_client_err!("Failed to parse controller config response body".to_string()))
4658}
4659
4660pub(super) fn should_fetch_system_topic_list_from_broker(topic_list: &TopicList) -> bool {
4661    !topic_list.topic_list.is_empty()
4662        && topic_list
4663            .broker_addr
4664            .as_ref()
4665            .is_some_and(|broker_addr| !broker_addr.as_str().trim().is_empty())
4666}
4667
4668pub(super) fn append_system_topic_list_from_broker_like_java(topic_list: &mut TopicList, broker_topic_list: TopicList) {
4669    if !broker_topic_list.topic_list.is_empty() {
4670        topic_list.topic_list.extend(broker_topic_list.topic_list);
4671    }
4672}
4673
4674pub(super) async fn merge_system_topic_list_from_broker(
4675    api: &MQClientAPIImpl,
4676    topic_list: &mut TopicList,
4677    timeout_millis: u64,
4678) -> ClientResult<()> {
4679    if !should_fetch_system_topic_list_from_broker(topic_list) {
4680        return Ok(());
4681    }
4682    let Some(broker_addr) = topic_list.broker_addr.clone() else {
4683        return Ok(());
4684    };
4685    let broker_topic_list = api
4686        .get_system_topic_list_from_broker(&broker_addr, timeout_millis)
4687        .await?;
4688    append_system_topic_list_from_broker_like_java(topic_list, broker_topic_list);
4689    Ok(())
4690}
4691
4692pub(super) fn decode_cluster_acl_version_info_response_body(
4693    body: Option<&bytes::Bytes>,
4694) -> ClientResult<ClusterAclVersionInfo> {
4695    let body = body.ok_or_else(|| mq_client_err!("get_broker_cluster_acl_version_info response body is empty"))?;
4696    SerdeJsonUtils::from_json_slice(body.as_ref())
4697        .map_err(|source| ClientError::response_process_source("decode ClusterAclVersionInfo", source))
4698}
4699
4700pub(super) fn admin_message_matches_query(
4701    topic: &CheetahString,
4702    key: &CheetahString,
4703    msg: &MessageExt,
4704    unique_key_flag: bool,
4705) -> bool {
4706    if msg.topic().as_str() != topic.as_str() {
4707        return false;
4708    }
4709
4710    if unique_key_flag {
4711        if let Some(uniq_id) = MessageClientIDSetter::get_uniq_id(msg) {
4712            if uniq_id.as_str() == key.as_str() {
4713                return true;
4714            }
4715        }
4716        return msg.msg_id.as_str() == key.as_str();
4717    }
4718
4719    let Some(keys) = MessageTrait::get_keys(msg) else {
4720        return false;
4721    };
4722
4723    keys.split_str(MessageConst::KEY_SEPARATOR)
4724        .any(|candidate| candidate == key.as_str())
4725}
4726
4727impl MQClientAPIImpl {
4728    pub(super) async fn invoke_admin_request(
4729        &self,
4730        address: &str,
4731        request: RemotingCommand,
4732        timeout_millis: u64,
4733    ) -> Result<OutboundRequestOutcome, rocketmq_transport::api::TransportError> {
4734        let address = CheetahString::from_slice(address);
4735        self.remoting_client
4736            .invoke_request(Some(&address), request, timeout_millis)
4737            .await
4738    }
4739}
4740
4741#[cfg(test)]
4742mod retry_executor_tests {
4743    use std::collections::VecDeque;
4744    use std::future::Future;
4745    use std::pin::Pin;
4746    use std::sync::atomic::AtomicUsize;
4747    use std::sync::atomic::Ordering;
4748    use std::sync::Arc;
4749    use std::sync::Mutex;
4750    use std::time::Duration;
4751
4752    use rocketmq_protocol::protocol::header::namesrv::kv_config_header::GetKVConfigResponseHeader;
4753    use rocketmq_protocol::RemotingCommand;
4754    use rocketmq_runtime::RuntimeContext;
4755    use rocketmq_runtime::ShutdownDeadline;
4756    use rocketmq_transport::api::AdmissionController;
4757    use rocketmq_transport::api::AdmissionLimits;
4758    use rocketmq_transport::test_support::SessionProcessor;
4759    use rocketmq_transport::test_support::SessionTransportServer;
4760    use rocketmq_transport::test_support::SessionTransportServerConfig;
4761
4762    use super::*;
4763    use crate::base::client_config::ClientConfig;
4764    use crate::factory::mq_client_instance::MQClientInstance;
4765
4766    #[derive(Default)]
4767    struct KvProcessor {
4768        responses: Mutex<VecDeque<RemotingCommand>>,
4769        requests: AtomicUsize,
4770    }
4771
4772    impl KvProcessor {
4773        fn push(&self, response: RemotingCommand) {
4774            self.responses.lock().expect("KV response queue").push_back(response);
4775        }
4776
4777        fn request_count(&self) -> usize {
4778            self.requests.load(Ordering::SeqCst)
4779        }
4780    }
4781
4782    impl SessionProcessor for KvProcessor {
4783        fn process(
4784            &self,
4785            request: RemotingCommand,
4786        ) -> Pin<Box<dyn Future<Output = Result<RemotingCommand, rocketmq_error::SharedError>> + Send + '_>> {
4787            Box::pin(async move {
4788                if request.code() != RequestCode::GetKvConfig.to_i32() {
4789                    return Err(
4790                        ClientError::illegal_argument(format!("unexpected request code {}", request.code())).into(),
4791                    );
4792                }
4793                self.requests.fetch_add(1, Ordering::SeqCst);
4794                let response = self
4795                    .responses
4796                    .lock()
4797                    .expect("KV response queue")
4798                    .pop_front()
4799                    .ok_or_else(|| ClientError::illegal_argument("unexpected KV request"))?;
4800                Ok(response.set_opaque(request.opaque()))
4801            })
4802        }
4803    }
4804
4805    async fn bind_kv_server(
4806        runtime: &RuntimeContext,
4807        scope: &'static str,
4808        processor: Arc<KvProcessor>,
4809    ) -> Arc<SessionTransportServer> {
4810        let admission = Arc::new(AdmissionController::new(AdmissionLimits::default()));
4811        let server = SessionTransportServer::bind(
4812            runtime.service_context(scope),
4813            SessionTransportServerConfig::loopback(),
4814            processor as Arc<dyn SessionProcessor>,
4815            admission,
4816        )
4817        .await
4818        .expect("bind KV server");
4819        server.start().expect("start KV server");
4820        server
4821    }
4822
4823    #[tokio::test]
4824    async fn kv_executor_switches_name_server_and_keeps_query_not_found_terminal() {
4825        let server_runtime = RuntimeContext::from_current("kv-retry-executor-test");
4826        let first = Arc::new(KvProcessor::default());
4827        let second = Arc::new(KvProcessor::default());
4828        let first_server = bind_kv_server(&server_runtime, "kv-first", Arc::clone(&first)).await;
4829        let second_server = bind_kv_server(&server_runtime, "kv-second", Arc::clone(&second)).await;
4830        let client_runtime = crate::runtime::test_client_runtime("kv-retry-executor-test");
4831        let client_instance = MQClientInstance::new_arc(
4832            ClientConfig::default(),
4833            0,
4834            "kv-retry-client",
4835            None,
4836            client_runtime.component("instance"),
4837            client_runtime.telemetry_handle().clone(),
4838            client_runtime.pool().request_future_holder(),
4839        );
4840        let api = client_instance.get_mq_client_api_impl().expect("client API");
4841        let first_addr = CheetahString::from_string(first_server.local_addr().to_string());
4842        let second_addr = CheetahString::from_string(second_server.local_addr().to_string());
4843        api.update_name_server_targets_sync(vec![first_addr.clone(), second_addr.clone()]);
4844        let selected_targets = api.remoting_client.get_name_server_address_list();
4845        assert_eq!(selected_targets.len(), 2, "both NameServers must be selected");
4846        let (initial, fallback) = if selected_targets[0] == first_addr {
4847            (&first, &second)
4848        } else {
4849            assert_eq!(selected_targets[0], second_addr);
4850            (&second, &first)
4851        };
4852        initial.push(RemotingCommand::create_response_command_with_code_remark(
4853            ResponseCode::SystemError,
4854            "initial-name-server-failed",
4855        ));
4856        let mut found = RemotingCommand::create_response_command_with_code_and_header(
4857            ResponseCode::Success,
4858            GetKVConfigResponseHeader::new(Some(CheetahString::from_static_str("expected-value"))),
4859        );
4860        found.make_custom_header_to_net();
4861        fallback.push(found);
4862        api.start().await.expect("start client transport");
4863
4864        let value = api
4865            .get_kvconfig_value("namespace".into(), "key".into(), 2_000)
4866            .await
4867            .expect("second NameServer should satisfy the query");
4868        assert_eq!(value.as_deref(), Some("expected-value"));
4869        assert_eq!((initial.request_count(), fallback.request_count()), (1, 1));
4870
4871        first.push(RemotingCommand::create_response_command_with_code(
4872            ResponseCode::QueryNotFound,
4873        ));
4874        api.update_name_server_targets_sync(vec![first_addr]);
4875        let first_before_terminal = first.request_count();
4876        let second_before_terminal = second.request_count();
4877        let missing = api
4878            .get_kvconfig_value("namespace".into(), "missing".into(), 2_000)
4879            .await
4880            .expect("QueryNotFound is a normal terminal outcome");
4881        assert_eq!(missing, None);
4882        assert_eq!(first.request_count(), first_before_terminal + 1);
4883        assert_eq!(
4884            second.request_count(),
4885            second_before_terminal,
4886            "QueryNotFound must not switch endpoint"
4887        );
4888
4889        client_instance.shutdown().await;
4890        client_runtime
4891            .shutdown()
4892            .await
4893            .assert_no_task_leak()
4894            .expect("client runtime tasks drained");
4895        for server in [first_server, second_server] {
4896            server
4897                .shutdown_until(ShutdownDeadline::after(Duration::from_secs(5)))
4898                .await
4899                .assert_no_task_leak()
4900                .expect("KV server tasks drained");
4901        }
4902        server_runtime
4903            .shutdown_tasks(Duration::from_secs(5))
4904            .await
4905            .assert_no_task_leak()
4906            .expect("KV runtime tasks drained");
4907    }
4908}
4909
4910#[cfg(test)]
4911mod json_error_boundary_tests {
4912    use std::error::Error as StdError;
4913
4914    #[test]
4915    fn admin_json_decode_preserves_source_and_public_boundary() {
4916        let error = super::decode_admin_json::<serde_json::Value>(b"invalid").unwrap_err();
4917
4918        assert_eq!(error.to_string(), "core.serialization.failed: Serialization failed");
4919        assert_eq!(error.descriptor().public_message(), "Serialization failed");
4920
4921        let canonical = StdError::source(&error).expect("canonical error source");
4922        assert!(canonical.downcast_ref::<rocketmq_error::Error>().is_some());
4923        assert!(error.source_ref::<serde_json::Error>().is_some());
4924
4925        assert!(error.is(&rocketmq_error::CORE_SERIALIZATION_FAILED));
4926    }
4927
4928    #[cfg(any(feature = "admin-read", feature = "admin-mutation"))]
4929    #[test]
4930    fn proxy_drain_decode_preserves_source() {
4931        let response = rocketmq_protocol::RemotingCommand::create_success_response_command().set_body("invalid");
4932
4933        let error =
4934            super::decode_proxy_drain_response(response, "proxy_drain_test").expect_err("invalid body must fail");
4935
4936        assert!(error.is(&rocketmq_error::PROTOCOL_RESPONSE_FAILED));
4937        assert!(error.source_ref::<rocketmq_error::Error>().is_some());
4938    }
4939}
4940
4941#[cfg(feature = "admin-mutation")]
4942fn acl_entry_delete_header(
4943    subject: CheetahString,
4944    policy_type: CheetahString,
4945    resource: CheetahString,
4946) -> ClientResult<DeleteAclRequestHeader> {
4947    if subject.trim().is_empty()
4948        || resource.trim().is_empty()
4949        || !(policy_type.eq_ignore_ascii_case("Custom") || policy_type.eq_ignore_ascii_case("Default"))
4950    {
4951        return Err(ClientError::illegal_argument(
4952            "ACL entry deletion requires subject, resource and Custom or Default policy type",
4953        ));
4954    }
4955    Ok(DeleteAclRequestHeader::with_policy_type(
4956        subject,
4957        Some(policy_type),
4958        Some(resource),
4959    ))
4960}
4961#[cfg(all(test, feature = "admin-mutation"))]
4962mod acl_policy_entry_tests {
4963    use super::*;
4964    #[test]
4965    fn acl_policy_entry_delete_preserves_type_and_never_expands_empty_resource() {
4966        let header = acl_entry_delete_header("User:alice".into(), "Default".into(), "Topic:orders".into()).unwrap();
4967        assert_eq!(header.policy_type.as_deref(), Some("Default"));
4968        assert_eq!(header.resource.as_deref(), Some("Topic:orders"));
4969        assert!(acl_entry_delete_header("User:alice".into(), "Default".into(), "".into()).is_err());
4970        assert!(acl_entry_delete_header("User:alice".into(), "invalid".into(), "Topic:orders".into()).is_err());
4971        let legacy = DeleteAclRequestHeader::new("User:alice".into(), None);
4972        assert!(legacy.policy_type.is_none());
4973        assert!(legacy.resource.is_none());
4974    }
4975}
rocketmq_client_rust/implementation/mq_client_api_impl/admin.rs

rocketmq_client_rust/implementation/mq_client_api_impl/
admin.rs
