rocketmq_client_rust/admin/
mq_admin_mutation_ext.rs

1// Copyright 2026 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// Unless required by applicable law or agreed to in writing, software
9// distributed under the License is distributed on an "AS IS" BASIS,
10// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
11// See the License for the specific language governing permissions and
12// limitations under the License.
13
14//! Narrow administration mutation surface.
15//!
16//! Method names are intentionally distinct from the full administration
17//! capability set, allowing mutation-only consumers to depend on an explicit
18//! supervised capability.
19
20use std::collections::HashMap;
21use std::collections::HashSet;
22
23use cheetah_string::CheetahString;
24use rocketmq_model::common::config::TopicConfig;
25use rocketmq_model::common::message::message_enum::MessageRequestMode;
26use rocketmq_model::common::message::message_ext::MessageExt;
27use rocketmq_model::common::message::message_queue::MessageQueue;
28use rocketmq_protocol::protocol::admin::rollback_stats::RollbackStats;
29use rocketmq_protocol::protocol::body::broker_body::cluster_info::ClusterInfo;
30use rocketmq_protocol::protocol::body::consume_message_directly_result::ConsumeMessageDirectlyResult;
31use rocketmq_protocol::protocol::body::proxy_drain::ProxyDrainStateResponseBody;
32use rocketmq_protocol::protocol::route::topic_route_data::TopicRouteData;
33use rocketmq_protocol::protocol::subscription::subscription_group_config::SubscriptionGroupConfig;
34
35use super::default_mq_admin_ext::DefaultMQAdminExt;
36
37/// Result of one generation-checked broker configuration patch.
38#[derive(Clone, Copy, Debug, Eq, PartialEq)]
39pub enum BrokerConfigPatchOutcome {
40    /// The patch was committed as the generation immediately after precheck.
41    Applied { previous_generation: u64, generation: u64 },
42    /// The broker changed after precheck; callers must stop and re-plan.
43    GenerationConflict {
44        expected_generation: u64,
45        actual_generation: u64,
46    },
47}
48
49/// Closed Topic fields accepted by the supervised version-CAS operation.
50#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
51pub struct TopicConfigPatch {
52    pub read_queue_nums: Option<u32>,
53    pub write_queue_nums: Option<u32>,
54    pub order: Option<bool>,
55}
56
57impl TopicConfigPatch {
58    #[must_use]
59    pub const fn is_empty(self) -> bool {
60        self.read_queue_nums.is_none() && self.write_queue_nums.is_none() && self.order.is_none()
61    }
62}
63
64/// Result of one version-checked Topic configuration patch.
65#[derive(Clone, Copy, Debug, Eq, PartialEq)]
66pub enum TopicConfigPatchOutcome {
67    /// The patch was committed as the version immediately after precheck.
68    Applied { previous_version: u64, version: u64 },
69    /// Topic metadata changed after precheck; callers must stop and re-plan.
70    VersionConflict { expected_version: u64, actual_version: u64 },
71}
72
73/// Allowlisted Topic state returned to an isolated mutation preflight.
74#[derive(Clone, Debug, PartialEq)]
75pub struct MutationTopicConfigVersioned {
76    pub version: u64,
77    pub config: TopicConfig,
78}
79
80/// Closed presence/version condition for full supervised metadata replacement.
81#[derive(Clone, Copy, Debug, Eq, PartialEq)]
82pub enum MutationExpectedState {
83    Absent,
84    Present { version: u64 },
85}
86
87/// Closed Topic message types supported by supervised replacement.
88#[derive(Clone, Copy, Debug, Eq, PartialEq)]
89pub enum MutationTopicMessageType {
90    Normal,
91    Fifo,
92    Delay,
93    Transaction,
94    Unspecified,
95}
96
97impl MutationTopicMessageType {
98    pub(crate) const fn wire_name(self) -> &'static str {
99        match self {
100            Self::Normal => "NORMAL",
101            Self::Fifo => "FIFO",
102            Self::Delay => "DELAY",
103            Self::Transaction => "TRANSACTION",
104            Self::Unspecified => "UNSPECIFIED",
105        }
106    }
107}
108
109/// Complete allowlisted Topic state used by a supervised replace operation.
110#[derive(Clone, Debug, Eq, PartialEq)]
111pub struct MutationTopicConfig {
112    pub read_queue_nums: u32,
113    pub write_queue_nums: u32,
114    pub perm: u32,
115    pub order: bool,
116    pub message_type: MutationTopicMessageType,
117}
118
119/// Presence-aware Topic preflight result.
120#[derive(Clone, Debug, Eq, PartialEq)]
121pub struct MutationTopicConfigState {
122    pub state: MutationExpectedState,
123    pub config: Option<MutationTopicConfig>,
124}
125
126/// Complete allowlisted Subscription Group state used by supervised replacement.
127#[derive(Clone, Debug, Eq, PartialEq)]
128pub struct MutationSubscriptionGroupConfig {
129    pub consume_enable: bool,
130    pub consume_from_min_enable: bool,
131    pub consume_broadcast_enable: bool,
132    pub consume_message_orderly: bool,
133    pub retry_queue_nums: i32,
134    pub retry_max_times: i32,
135    pub broker_id: u64,
136    pub which_broker_when_consume_slowly: u64,
137    pub notify_consumer_ids_changed_enable: bool,
138    pub group_sys_flag: i32,
139    pub consume_timeout_minute: i32,
140}
141
142/// Presence-aware Subscription Group preflight result.
143#[derive(Clone, Debug, Eq, PartialEq)]
144pub struct MutationSubscriptionGroupConfigState {
145    pub state: MutationExpectedState,
146    pub config: Option<MutationSubscriptionGroupConfig>,
147}
148
149/// Result of one presence/version conditional replacement.
150#[derive(Clone, Copy, Debug, Eq, PartialEq)]
151pub struct MutationStateCasOutcome {
152    pub applied: bool,
153    pub changed: bool,
154    pub state: MutationExpectedState,
155    pub persistence: MutationPersistenceState,
156}
157
158/// Durable state of an accepted supervised mutation.
159#[derive(Clone, Copy, Debug, Eq, PartialEq)]
160pub enum MutationPersistenceState {
161    NotRequired,
162    Persisted,
163    /// The write is confirmed not to have reached the target file.
164    Failed,
165    /// The write may already have reached the target file; the Broker ended
166    /// its observation without confirming either durability or failure.
167    Unconfirmed,
168}
169
170/// Six-key Broker state allowed into mutation preflight.
171#[derive(Clone, Copy, Debug, Eq, PartialEq)]
172pub struct BrokerMutationConfigState {
173    pub generation: u64,
174    pub auto_create_topic_enable: bool,
175    pub auto_create_subscription_group: bool,
176    pub broker_permission: u32,
177    pub default_topic_queue_nums: u32,
178    pub message_index_enable: bool,
179    pub trace_topic_enable: bool,
180}
181
182/// Exact conditional consumer-offset update result.
183#[derive(Clone, Copy, Debug, Eq, PartialEq)]
184pub struct ConditionalConsumerOffsetOutcome {
185    pub applied: bool,
186    pub actual_offset: i64,
187}
188
189/// Pure preflight row for one logical Broker queue; no network address escapes.
190#[derive(Clone, Debug, Eq, PartialEq)]
191pub struct MutationConsumerOffsetPreview {
192    pub broker_name: String,
193    pub queue_id: i32,
194    pub current_offset: i64,
195    pub planned_offset: i64,
196}
197
198/// Exact request-mode value returned by a selected Broker master.
199#[derive(Clone, Copy, Debug, Eq, PartialEq)]
200pub struct MutationMessageRequestMode {
201    pub mode: MessageRequestMode,
202    pub pop_share_queue_num: i32,
203}
204
205/// Closed request-mode precondition.
206#[derive(Clone, Copy, Debug, Eq, PartialEq)]
207pub enum MutationExpectedMessageRequestMode {
208    Absent,
209    Present(MutationMessageRequestMode),
210}
211
212/// Conditional request-mode result containing only the current typed value.
213#[derive(Clone, Copy, Debug, Eq, PartialEq)]
214pub struct MutationMessageRequestModeOutcome {
215    pub applied: bool,
216    pub changed: bool,
217    pub current: Option<MutationMessageRequestMode>,
218    pub persistence: MutationPersistenceState,
219}
220
221/// Closed failure categories for one detailed consumer-offset target.
222#[derive(Clone, Copy, Debug, Eq, PartialEq)]
223pub enum TopicOffsetMutationFailureCode {
224    InvalidData,
225    Unavailable,
226}
227
228/// One broker or queue result from a detailed reset/skip workflow.
229#[derive(Clone, Debug, Eq, PartialEq)]
230pub struct TopicOffsetMutationTargetOutcome {
231    pub broker_name: String,
232    pub queue_id: Option<i32>,
233    pub applied: bool,
234    pub offset: Option<u64>,
235    pub failure: Option<TopicOffsetMutationFailureCode>,
236    pub retryable: bool,
237}
238
239/// Failure-aware results for all exact broker/queue targets reached by one offset mutation.
240#[derive(Clone, Debug, Default, Eq, PartialEq)]
241pub struct TopicOffsetMutationOutcome {
242    pub targets: Vec<TopicOffsetMutationTargetOutcome>,
243}
244
245/// Closed Subscription Group fields accepted by the supervised version-CAS
246/// operation.
247#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
248pub struct SubscriptionGroupConfigPatch {
249    pub retry_max_times: Option<u32>,
250    pub retry_queue_nums: Option<u32>,
251    pub consume_timeout_minutes: Option<u32>,
252}
253
254impl SubscriptionGroupConfigPatch {
255    #[must_use]
256    pub const fn is_empty(self) -> bool {
257        self.retry_max_times.is_none() && self.retry_queue_nums.is_none() && self.consume_timeout_minutes.is_none()
258    }
259}
260
261/// Result of one version-checked Subscription Group configuration patch.
262#[derive(Clone, Copy, Debug, Eq, PartialEq)]
263pub enum SubscriptionGroupConfigPatchOutcome {
264    /// The patch was committed as the version immediately after precheck.
265    Applied { previous_version: u64, version: u64 },
266    /// Subscription Group metadata changed after precheck; callers must stop
267    /// and re-plan.
268    VersionConflict { expected_version: u64, actual_version: u64 },
269}
270
271/// Explicit RocketMQ mutation capability.
272///
273/// New integrations should request this trait deliberately and keep it out of
274/// read-only process dependency graphs.
275#[allow(async_fn_in_trait)]
276pub trait MQAdminMutationExt: Send {
277    /// Begins one authenticated, reversible drain operation for a Proxy.
278    async fn begin_proxy_drain(
279        &self,
280        proxy_addr: CheetahString,
281        operation_id: CheetahString,
282    ) -> crate::ClientResult<ProxyDrainStateResponseBody>;
283
284    /// Cancels a timed-out drain and restores Proxy admission/readiness.
285    async fn cancel_proxy_drain(
286        &self,
287        proxy_addr: CheetahString,
288        operation_id: CheetahString,
289    ) -> crate::ClientResult<ProxyDrainStateResponseBody>;
290
291    /// Reads the broker configuration generation used by supervised prechecks.
292    async fn broker_config_generation(&self, broker_addr: CheetahString) -> crate::ClientResult<u64>;
293
294    /// Applies a broker configuration patch only when `expected_generation`
295    /// still matches the broker's current generation.
296    async fn patch_broker_config_if_generation(
297        &self,
298        broker_addr: CheetahString,
299        expected_generation: u64,
300        properties: HashMap<CheetahString, CheetahString>,
301    ) -> crate::ClientResult<BrokerConfigPatchOutcome>;
302
303    /// Changes only the three fields in [`TopicConfigPatch`] when the Broker's
304    /// current Topic metadata version still matches `expected_version`.
305    async fn patch_topic_config_if_version(
306        &self,
307        broker_addr: CheetahString,
308        topic: CheetahString,
309        expected_version: u64,
310        patch: TopicConfigPatch,
311    ) -> crate::ClientResult<TopicConfigPatchOutcome>;
312
313    /// Reads only the Topic fields and version needed to prepare a supervised
314    /// queue-count compare-and-set mutation.
315    async fn mutation_topic_config_with_version(
316        &self,
317        broker_addr: CheetahString,
318        topic: CheetahString,
319    ) -> crate::ClientResult<MutationTopicConfigVersioned> {
320        let _ = (broker_addr, topic);
321        Err(crate::ClientError::illegal_argument(
322            "versioned Topic mutation preflight is not implemented by this admin client",
323        ))
324    }
325
326    /// Reads a presence-aware, allowlisted Topic state from one exact Broker.
327    async fn mutation_topic_config_state(
328        &self,
329        broker_addr: CheetahString,
330        topic: CheetahString,
331    ) -> crate::ClientResult<MutationTopicConfigState> {
332        let _ = (broker_addr, topic);
333        Err(crate::ClientError::illegal_argument(
334            "presence-aware Topic mutation preflight is not implemented by this admin client",
335        ))
336    }
337
338    /// Creates or fully replaces the allowlisted Topic state only when its
339    /// presence/version precondition still matches.
340    async fn replace_topic_config_if_state(
341        &self,
342        broker_addr: CheetahString,
343        topic: CheetahString,
344        expected_state: MutationExpectedState,
345        replacement: MutationTopicConfig,
346    ) -> crate::ClientResult<MutationStateCasOutcome> {
347        let _ = (broker_addr, topic, expected_state, replacement);
348        Err(crate::ClientError::illegal_argument(
349            "presence-aware Topic replacement is not implemented by this admin client",
350        ))
351    }
352
353    /// Changes only the three fields in [`SubscriptionGroupConfigPatch`] when
354    /// the Broker's current Subscription Group metadata version still matches
355    /// `expected_version`.
356    async fn patch_subscription_group_config_if_version(
357        &self,
358        broker_addr: CheetahString,
359        group: CheetahString,
360        expected_version: u64,
361        patch: SubscriptionGroupConfigPatch,
362    ) -> crate::ClientResult<SubscriptionGroupConfigPatchOutcome>;
363
364    /// Reads presence-aware, allowlisted Subscription Group state from one exact Broker.
365    async fn mutation_subscription_group_config_state(
366        &self,
367        broker_addr: CheetahString,
368        group: CheetahString,
369    ) -> crate::ClientResult<MutationSubscriptionGroupConfigState> {
370        let _ = (broker_addr, group);
371        Err(crate::ClientError::illegal_argument(
372            "presence-aware Subscription Group mutation preflight is not implemented by this admin client",
373        ))
374    }
375
376    /// Creates or fully replaces the allowlisted Subscription Group state under one state CAS.
377    async fn replace_subscription_group_config_if_state(
378        &self,
379        broker_addr: CheetahString,
380        group: CheetahString,
381        expected_state: MutationExpectedState,
382        replacement: MutationSubscriptionGroupConfig,
383    ) -> crate::ClientResult<MutationStateCasOutcome> {
384        let _ = (broker_addr, group, expected_state, replacement);
385        Err(crate::ClientError::illegal_argument(
386            "presence-aware Subscription Group replacement is not implemented by this admin client",
387        ))
388    }
389
390    /// Reads only the six Broker settings accepted by supervised control preflight.
391    async fn broker_mutation_config_state(
392        &self,
393        broker_addr: CheetahString,
394    ) -> crate::ClientResult<BrokerMutationConfigState> {
395        let _ = broker_addr;
396        Err(crate::ClientError::illegal_argument(
397            "allowlisted Broker mutation preflight is not implemented by this admin client",
398        ))
399    }
400
401    /// Applies one exact queue offset transition without route re-resolution or retry.
402    async fn reset_consumer_offset_if_current(
403        &self,
404        broker_addr: CheetahString,
405        consumer_group: CheetahString,
406        topic: CheetahString,
407        queue_id: i32,
408        expected_offset: i64,
409        new_offset: i64,
410    ) -> crate::ClientResult<ConditionalConsumerOffsetOutcome> {
411        let _ = (
412            broker_addr,
413            consumer_group,
414            topic,
415            queue_id,
416            expected_offset,
417            new_offset,
418        );
419        Err(crate::ClientError::illegal_argument(
420            "conditional consumer-offset mutation is not implemented by this admin client",
421        ))
422    }
423
424    /// Computes exact queue offsets for one already-selected Broker master without mutation.
425    async fn preview_consumer_offset_reset_on_broker(
426        &self,
427        broker_addr: CheetahString,
428        broker_name: CheetahString,
429        read_queue_nums: u32,
430        consumer_group: CheetahString,
431        topic: CheetahString,
432        timestamp: i64,
433    ) -> crate::ClientResult<Vec<MutationConsumerOffsetPreview>> {
434        let _ = (
435            broker_addr,
436            broker_name,
437            read_queue_nums,
438            consumer_group,
439            topic,
440            timestamp,
441        );
442        Err(crate::ClientError::illegal_argument(
443            "consumer-offset mutation preview is not implemented by this admin client",
444        ))
445    }
446
447    /// Re-reads one exact queue offset for postcondition verification.
448    async fn mutation_consumer_offset(
449        &self,
450        broker_addr: CheetahString,
451        consumer_group: CheetahString,
452        topic: CheetahString,
453        queue_id: i32,
454    ) -> crate::ClientResult<i64> {
455        let _ = (broker_addr, consumer_group, topic, queue_id);
456        Err(crate::ClientError::illegal_argument(
457            "consumer-offset mutation verification is not implemented by this admin client",
458        ))
459    }
460
461    /// Reads one exact Topic/group request-mode entry from a selected Broker master.
462    async fn mutation_message_request_mode(
463        &self,
464        broker_addr: CheetahString,
465        topic: CheetahString,
466        consumer_group: CheetahString,
467    ) -> crate::ClientResult<Option<MutationMessageRequestMode>> {
468        let _ = (broker_addr, topic, consumer_group);
469        Err(crate::ClientError::illegal_argument(
470            "request-mode mutation preflight is not implemented by this admin client",
471        ))
472    }
473
474    /// Conditionally replaces one exact Topic/group request-mode entry.
475    async fn replace_message_request_mode_if_current(
476        &self,
477        broker_addr: CheetahString,
478        topic: CheetahString,
479        consumer_group: CheetahString,
480        expected: MutationExpectedMessageRequestMode,
481        replacement: MutationMessageRequestMode,
482    ) -> crate::ClientResult<MutationMessageRequestModeOutcome> {
483        let _ = (broker_addr, topic, consumer_group, expected, replacement);
484        Err(crate::ClientError::illegal_argument(
485            "conditional request-mode mutation is not implemented by this admin client",
486        ))
487    }
488
489    /// Conditionally replaces one exact Topic/group request-mode entry using
490    /// the caller's bounded request timeout.
491    ///
492    /// The default deliberately fails closed so existing trait implementers
493    /// remain source compatible without silently ignoring the requested
494    /// timeout.
495    async fn replace_message_request_mode_if_current_with_timeout(
496        &self,
497        broker_addr: CheetahString,
498        topic: CheetahString,
499        consumer_group: CheetahString,
500        expected: MutationExpectedMessageRequestMode,
501        replacement: MutationMessageRequestMode,
502        timeout_millis: u64,
503    ) -> crate::ClientResult<MutationMessageRequestModeOutcome> {
504        let _ = (
505            broker_addr,
506            topic,
507            consumer_group,
508            expected,
509            replacement,
510            timeout_millis,
511        );
512        Err(crate::ClientError::illegal_argument(
513            "timeout-aware conditional request-mode mutation is not implemented by this admin client",
514        ))
515    }
516
517    /// Applies one bounded Broker logger override with an automatic TTL.
518    ///
519    /// Implementations must reject arbitrary filter expressions. `logger`
520    /// identifies one `rocketmq_broker::` target, `level` is limited to
521    /// `INFO` or `DEBUG`, and the TTL is limited to the server-supported
522    /// diagnostic window.
523    async fn set_broker_log_filter_ttl(
524        &self,
525        broker_addr: CheetahString,
526        logger: CheetahString,
527        level: CheetahString,
528        ttl_seconds: u32,
529        operation_id: CheetahString,
530    ) -> crate::ClientResult<()> {
531        let _ = (broker_addr, logger, level, ttl_seconds, operation_id);
532        Err(crate::ClientError::illegal_argument(
533            "typed broker log-filter mutation is not implemented by this admin client",
534        ))
535    }
536
537    /// Restores the Broker logger baseline for one bounded operation.
538    async fn restore_broker_log_filter(
539        &self,
540        broker_addr: CheetahString,
541        operation_id: CheetahString,
542    ) -> crate::ClientResult<()> {
543        let _ = (broker_addr, operation_id);
544        Err(crate::ClientError::illegal_argument(
545            "typed broker log-filter restoration is not implemented by this admin client",
546        ))
547    }
548
549    async fn upsert_topic_config(&self, broker_addr: CheetahString, config: TopicConfig) -> crate::ClientResult<()>;
550
551    async fn remove_topic(&self, topic_name: CheetahString, cluster_name: CheetahString) -> crate::ClientResult<()>;
552
553    async fn reset_consumer_offset(
554        &self,
555        cluster_name: Option<CheetahString>,
556        topic: CheetahString,
557        consumer_group: CheetahString,
558        timestamp: u64,
559        force: bool,
560    ) -> crate::ClientResult<HashMap<MessageQueue, u64>>;
561
562    /// Resets each exact broker/queue target independently and retains partial outcomes.
563    async fn reset_consumer_offset_detailed(
564        &self,
565        cluster_name: CheetahString,
566        topic: CheetahString,
567        consumer_group: CheetahString,
568        timestamp: u64,
569        force: bool,
570    ) -> crate::ClientResult<TopicOffsetMutationOutcome> {
571        let _ = (cluster_name, topic, consumer_group, timestamp, force);
572        Err(crate::ClientError::illegal_argument(
573            "detailed reset-offset mutation is not implemented by this admin client",
574        ))
575    }
576
577    /// Advances one exact consumer group and Topic to the latest available
578    /// offsets. This is deliberately separate from timestamp reset because the
579    /// RocketMQ protocol represents latest with the signed `-1` sentinel.
580    async fn skip_accumulated_message(
581        &self,
582        cluster_name: Option<CheetahString>,
583        topic: CheetahString,
584        consumer_group: CheetahString,
585        force: bool,
586    ) -> crate::ClientResult<usize> {
587        let _ = (cluster_name, topic, consumer_group, force);
588        Err(crate::ClientError::illegal_argument(
589            "skip-accumulated mutation is not implemented by this admin client",
590        ))
591    }
592
593    /// Advances each exact broker/queue target to latest while retaining partial outcomes.
594    async fn skip_accumulated_message_detailed(
595        &self,
596        cluster_name: CheetahString,
597        topic: CheetahString,
598        consumer_group: CheetahString,
599        force: bool,
600    ) -> crate::ClientResult<TopicOffsetMutationOutcome> {
601        let _ = (cluster_name, topic, consumer_group, force);
602        Err(crate::ClientError::illegal_argument(
603            "detailed skip-accumulated mutation is not implemented by this admin client",
604        ))
605    }
606
607    async fn upsert_subscription_group(
608        &self,
609        broker_addr: CheetahString,
610        config: SubscriptionGroupConfig,
611    ) -> crate::ClientResult<()>;
612
613    async fn remove_subscription_group(
614        &self,
615        broker_addr: CheetahString,
616        group_name: CheetahString,
617        remove_offset: Option<bool>,
618    ) -> crate::ClientResult<()>;
619
620    /// Removes multiple subscription groups from one broker in a single request.
621    async fn remove_subscription_groups(
622        &self,
623        broker_addr: CheetahString,
624        group_names: Vec<CheetahString>,
625        clean_offset: bool,
626    ) -> crate::ClientResult<()>;
627
628    async fn configure_message_request_mode(
629        &self,
630        broker_addr: CheetahString,
631        topic: CheetahString,
632        consumer_group: CheetahString,
633        mode: MessageRequestMode,
634        pop_work_group_size: i32,
635        timeout_millis: u64,
636    ) -> crate::ClientResult<()>;
637
638    async fn consume_directly(
639        &self,
640        consumer_group: CheetahString,
641        client_id: CheetahString,
642        topic: CheetahString,
643        message_id: CheetahString,
644    ) -> crate::ClientResult<ConsumeMessageDirectlyResult>;
645
646    async fn clone_consumer_group_offset(
647        &self,
648        source_group: CheetahString,
649        destination_group: CheetahString,
650        topic: CheetahString,
651        offline: bool,
652    ) -> crate::ClientResult<()>;
653
654    /// Returns the cluster topology needed to resolve mutation targets.
655    async fn mutation_cluster_info(&self) -> crate::ClientResult<ClusterInfo>;
656
657    /// Returns route data used to scope a mutation to the topic's brokers.
658    async fn mutation_topic_route(&self, topic: CheetahString) -> crate::ClientResult<Option<TopicRouteData>>;
659
660    /// Reads a broker's current topic configuration before a bounded test send.
661    async fn mutation_topic_config(
662        &self,
663        broker_addr: CheetahString,
664        topic: CheetahString,
665    ) -> crate::ClientResult<TopicConfig>;
666
667    /// Removes a topic from an explicitly resolved broker set.
668    async fn remove_topic_from_brokers(
669        &self,
670        broker_addrs: HashSet<CheetahString>,
671        topic: CheetahString,
672    ) -> crate::ClientResult<()>;
673
674    /// Removes multiple topics from one broker in a single request.
675    async fn remove_topics_from_broker(
676        &self,
677        broker_addr: CheetahString,
678        topics: Vec<CheetahString>,
679    ) -> crate::ClientResult<()>;
680
681    /// Removes a topic from an explicitly resolved NameServer set.
682    async fn remove_topic_from_name_servers(
683        &self,
684        namesrv_addrs: HashSet<CheetahString>,
685        cluster_name: Option<CheetahString>,
686        topic: CheetahString,
687    ) -> crate::ClientResult<()>;
688
689    /// Returns the NameServer targets owned by this started admin session.
690    async fn mutation_name_server_addresses(&self) -> crate::ClientResult<Vec<CheetahString>>;
691
692    /// Updates the order-topic configuration associated with a topic mutation.
693    async fn upsert_order_topic_config(
694        &self,
695        topic: CheetahString,
696        value: CheetahString,
697        cluster_wide: bool,
698    ) -> crate::ClientResult<()>;
699
700    /// Reads the exact NameServer-wide order-topic value used by a supervised mutation guard.
701    async fn mutation_order_topic_config(&self, topic: CheetahString) -> crate::ClientResult<Option<CheetahString>> {
702        let _ = topic;
703        Err(crate::ClientError::illegal_argument(
704            "order Topic mutation preflight is not implemented by this admin client",
705        ))
706    }
707
708    /// Deletes the global order-topic entry after an unordered update or a
709    /// complete Topic deletion.
710    async fn delete_order_topic_config(&self, topic: CheetahString) -> crate::ClientResult<()> {
711        let _ = topic;
712        Err(crate::ClientError::illegal_argument(
713            "order Topic configuration deletion is not implemented by this admin client",
714        ))
715    }
716
717    /// Performs the broker-by-broker offset reset fallback used for offline consumers.
718    async fn reset_consumer_offset_legacy(
719        &self,
720        cluster_name: Option<CheetahString>,
721        consumer_group: CheetahString,
722        topic: CheetahString,
723        timestamp: u64,
724        force: bool,
725    ) -> crate::ClientResult<Vec<RollbackStats>>;
726
727    /// Views the message metadata required to validate a DLQ resend.
728    async fn view_message_for_mutation(
729        &self,
730        topic: CheetahString,
731        message_id: CheetahString,
732    ) -> crate::ClientResult<MessageExt>;
733}
734
735impl MQAdminMutationExt for DefaultMQAdminExt {
736    async fn begin_proxy_drain(
737        &self,
738        proxy_addr: CheetahString,
739        operation_id: CheetahString,
740    ) -> crate::ClientResult<ProxyDrainStateResponseBody> {
741        MQAdminMutationExt::begin_proxy_drain(self.inner(), proxy_addr, operation_id).await
742    }
743
744    async fn cancel_proxy_drain(
745        &self,
746        proxy_addr: CheetahString,
747        operation_id: CheetahString,
748    ) -> crate::ClientResult<ProxyDrainStateResponseBody> {
749        MQAdminMutationExt::cancel_proxy_drain(self.inner(), proxy_addr, operation_id).await
750    }
751
752    async fn broker_config_generation(&self, broker_addr: CheetahString) -> crate::ClientResult<u64> {
753        MQAdminMutationExt::broker_config_generation(self.inner(), broker_addr).await
754    }
755
756    async fn patch_broker_config_if_generation(
757        &self,
758        broker_addr: CheetahString,
759        expected_generation: u64,
760        properties: HashMap<CheetahString, CheetahString>,
761    ) -> crate::ClientResult<BrokerConfigPatchOutcome> {
762        MQAdminMutationExt::patch_broker_config_if_generation(
763            self.inner(),
764            broker_addr,
765            expected_generation,
766            properties,
767        )
768        .await
769    }
770
771    async fn preview_consumer_offset_reset_on_broker(
772        &self,
773        broker_addr: CheetahString,
774        broker_name: CheetahString,
775        read_queue_nums: u32,
776        consumer_group: CheetahString,
777        topic: CheetahString,
778        timestamp: i64,
779    ) -> crate::ClientResult<Vec<MutationConsumerOffsetPreview>> {
780        MQAdminMutationExt::preview_consumer_offset_reset_on_broker(
781            self.inner(),
782            broker_addr,
783            broker_name,
784            read_queue_nums,
785            consumer_group,
786            topic,
787            timestamp,
788        )
789        .await
790    }
791
792    async fn mutation_consumer_offset(
793        &self,
794        broker_addr: CheetahString,
795        consumer_group: CheetahString,
796        topic: CheetahString,
797        queue_id: i32,
798    ) -> crate::ClientResult<i64> {
799        MQAdminMutationExt::mutation_consumer_offset(self.inner(), broker_addr, consumer_group, topic, queue_id).await
800    }
801
802    async fn patch_topic_config_if_version(
803        &self,
804        broker_addr: CheetahString,
805        topic: CheetahString,
806        expected_version: u64,
807        patch: TopicConfigPatch,
808    ) -> crate::ClientResult<TopicConfigPatchOutcome> {
809        MQAdminMutationExt::patch_topic_config_if_version(self.inner(), broker_addr, topic, expected_version, patch)
810            .await
811    }
812
813    async fn mutation_topic_config_with_version(
814        &self,
815        broker_addr: CheetahString,
816        topic: CheetahString,
817    ) -> crate::ClientResult<MutationTopicConfigVersioned> {
818        MQAdminMutationExt::mutation_topic_config_with_version(self.inner(), broker_addr, topic).await
819    }
820
821    async fn mutation_topic_config_state(
822        &self,
823        broker_addr: CheetahString,
824        topic: CheetahString,
825    ) -> crate::ClientResult<MutationTopicConfigState> {
826        MQAdminMutationExt::mutation_topic_config_state(self.inner(), broker_addr, topic).await
827    }
828
829    async fn replace_topic_config_if_state(
830        &self,
831        broker_addr: CheetahString,
832        topic: CheetahString,
833        expected_state: MutationExpectedState,
834        replacement: MutationTopicConfig,
835    ) -> crate::ClientResult<MutationStateCasOutcome> {
836        MQAdminMutationExt::replace_topic_config_if_state(self.inner(), broker_addr, topic, expected_state, replacement)
837            .await
838    }
839
840    async fn patch_subscription_group_config_if_version(
841        &self,
842        broker_addr: CheetahString,
843        group: CheetahString,
844        expected_version: u64,
845        patch: SubscriptionGroupConfigPatch,
846    ) -> crate::ClientResult<SubscriptionGroupConfigPatchOutcome> {
847        MQAdminMutationExt::patch_subscription_group_config_if_version(
848            self.inner(),
849            broker_addr,
850            group,
851            expected_version,
852            patch,
853        )
854        .await
855    }
856
857    async fn mutation_subscription_group_config_state(
858        &self,
859        broker_addr: CheetahString,
860        group: CheetahString,
861    ) -> crate::ClientResult<MutationSubscriptionGroupConfigState> {
862        MQAdminMutationExt::mutation_subscription_group_config_state(self.inner(), broker_addr, group).await
863    }
864
865    async fn replace_subscription_group_config_if_state(
866        &self,
867        broker_addr: CheetahString,
868        group: CheetahString,
869        expected_state: MutationExpectedState,
870        replacement: MutationSubscriptionGroupConfig,
871    ) -> crate::ClientResult<MutationStateCasOutcome> {
872        MQAdminMutationExt::replace_subscription_group_config_if_state(
873            self.inner(),
874            broker_addr,
875            group,
876            expected_state,
877            replacement,
878        )
879        .await
880    }
881
882    async fn broker_mutation_config_state(
883        &self,
884        broker_addr: CheetahString,
885    ) -> crate::ClientResult<BrokerMutationConfigState> {
886        MQAdminMutationExt::broker_mutation_config_state(self.inner(), broker_addr).await
887    }
888
889    async fn reset_consumer_offset_if_current(
890        &self,
891        broker_addr: CheetahString,
892        consumer_group: CheetahString,
893        topic: CheetahString,
894        queue_id: i32,
895        expected_offset: i64,
896        new_offset: i64,
897    ) -> crate::ClientResult<ConditionalConsumerOffsetOutcome> {
898        MQAdminMutationExt::reset_consumer_offset_if_current(
899            self.inner(),
900            broker_addr,
901            consumer_group,
902            topic,
903            queue_id,
904            expected_offset,
905            new_offset,
906        )
907        .await
908    }
909
910    async fn mutation_message_request_mode(
911        &self,
912        broker_addr: CheetahString,
913        topic: CheetahString,
914        consumer_group: CheetahString,
915    ) -> crate::ClientResult<Option<MutationMessageRequestMode>> {
916        MQAdminMutationExt::mutation_message_request_mode(self.inner(), broker_addr, topic, consumer_group).await
917    }
918
919    async fn replace_message_request_mode_if_current(
920        &self,
921        broker_addr: CheetahString,
922        topic: CheetahString,
923        consumer_group: CheetahString,
924        expected: MutationExpectedMessageRequestMode,
925        replacement: MutationMessageRequestMode,
926    ) -> crate::ClientResult<MutationMessageRequestModeOutcome> {
927        MQAdminMutationExt::replace_message_request_mode_if_current(
928            self.inner(),
929            broker_addr,
930            topic,
931            consumer_group,
932            expected,
933            replacement,
934        )
935        .await
936    }
937
938    async fn replace_message_request_mode_if_current_with_timeout(
939        &self,
940        broker_addr: CheetahString,
941        topic: CheetahString,
942        consumer_group: CheetahString,
943        expected: MutationExpectedMessageRequestMode,
944        replacement: MutationMessageRequestMode,
945        timeout_millis: u64,
946    ) -> crate::ClientResult<MutationMessageRequestModeOutcome> {
947        MQAdminMutationExt::replace_message_request_mode_if_current_with_timeout(
948            self.inner(),
949            broker_addr,
950            topic,
951            consumer_group,
952            expected,
953            replacement,
954            timeout_millis,
955        )
956        .await
957    }
958
959    async fn set_broker_log_filter_ttl(
960        &self,
961        broker_addr: CheetahString,
962        logger: CheetahString,
963        level: CheetahString,
964        ttl_seconds: u32,
965        operation_id: CheetahString,
966    ) -> crate::ClientResult<()> {
967        MQAdminMutationExt::set_broker_log_filter_ttl(
968            self.inner(),
969            broker_addr,
970            logger,
971            level,
972            ttl_seconds,
973            operation_id,
974        )
975        .await
976    }
977
978    async fn restore_broker_log_filter(
979        &self,
980        broker_addr: CheetahString,
981        operation_id: CheetahString,
982    ) -> crate::ClientResult<()> {
983        MQAdminMutationExt::restore_broker_log_filter(self.inner(), broker_addr, operation_id).await
984    }
985
986    async fn upsert_topic_config(&self, broker_addr: CheetahString, config: TopicConfig) -> crate::ClientResult<()> {
987        MQAdminMutationExt::upsert_topic_config(self.inner(), broker_addr, config).await
988    }
989
990    async fn remove_topic(&self, topic_name: CheetahString, cluster_name: CheetahString) -> crate::ClientResult<()> {
991        MQAdminMutationExt::remove_topic(self.inner(), topic_name, cluster_name).await
992    }
993
994    async fn reset_consumer_offset(
995        &self,
996        cluster_name: Option<CheetahString>,
997        topic: CheetahString,
998        consumer_group: CheetahString,
999        timestamp: u64,
1000        force: bool,
1001    ) -> crate::ClientResult<HashMap<MessageQueue, u64>> {
1002        MQAdminMutationExt::reset_consumer_offset(self.inner(), cluster_name, topic, consumer_group, timestamp, force)
1003            .await
1004    }
1005
1006    async fn reset_consumer_offset_detailed(
1007        &self,
1008        cluster_name: CheetahString,
1009        topic: CheetahString,
1010        consumer_group: CheetahString,
1011        timestamp: u64,
1012        force: bool,
1013    ) -> crate::ClientResult<TopicOffsetMutationOutcome> {
1014        MQAdminMutationExt::reset_consumer_offset_detailed(
1015            self.inner(),
1016            cluster_name,
1017            topic,
1018            consumer_group,
1019            timestamp,
1020            force,
1021        )
1022        .await
1023    }
1024
1025    async fn skip_accumulated_message(
1026        &self,
1027        cluster_name: Option<CheetahString>,
1028        topic: CheetahString,
1029        consumer_group: CheetahString,
1030        force: bool,
1031    ) -> crate::ClientResult<usize> {
1032        MQAdminMutationExt::skip_accumulated_message(self.inner(), cluster_name, topic, consumer_group, force).await
1033    }
1034
1035    async fn skip_accumulated_message_detailed(
1036        &self,
1037        cluster_name: CheetahString,
1038        topic: CheetahString,
1039        consumer_group: CheetahString,
1040        force: bool,
1041    ) -> crate::ClientResult<TopicOffsetMutationOutcome> {
1042        MQAdminMutationExt::skip_accumulated_message_detailed(self.inner(), cluster_name, topic, consumer_group, force)
1043            .await
1044    }
1045
1046    async fn upsert_subscription_group(
1047        &self,
1048        broker_addr: CheetahString,
1049        config: SubscriptionGroupConfig,
1050    ) -> crate::ClientResult<()> {
1051        MQAdminMutationExt::upsert_subscription_group(self.inner(), broker_addr, config).await
1052    }
1053
1054    async fn remove_subscription_group(
1055        &self,
1056        broker_addr: CheetahString,
1057        group_name: CheetahString,
1058        remove_offset: Option<bool>,
1059    ) -> crate::ClientResult<()> {
1060        MQAdminMutationExt::remove_subscription_group(self.inner(), broker_addr, group_name, remove_offset).await
1061    }
1062
1063    async fn remove_subscription_groups(
1064        &self,
1065        broker_addr: CheetahString,
1066        group_names: Vec<CheetahString>,
1067        clean_offset: bool,
1068    ) -> crate::ClientResult<()> {
1069        MQAdminMutationExt::remove_subscription_groups(self.inner(), broker_addr, group_names, clean_offset).await
1070    }
1071
1072    async fn configure_message_request_mode(
1073        &self,
1074        broker_addr: CheetahString,
1075        topic: CheetahString,
1076        consumer_group: CheetahString,
1077        mode: MessageRequestMode,
1078        pop_work_group_size: i32,
1079        timeout_millis: u64,
1080    ) -> crate::ClientResult<()> {
1081        MQAdminMutationExt::configure_message_request_mode(
1082            self.inner(),
1083            broker_addr,
1084            topic,
1085            consumer_group,
1086            mode,
1087            pop_work_group_size,
1088            timeout_millis,
1089        )
1090        .await
1091    }
1092
1093    async fn consume_directly(
1094        &self,
1095        consumer_group: CheetahString,
1096        client_id: CheetahString,
1097        topic: CheetahString,
1098        message_id: CheetahString,
1099    ) -> crate::ClientResult<ConsumeMessageDirectlyResult> {
1100        MQAdminMutationExt::consume_directly(self.inner(), consumer_group, client_id, topic, message_id).await
1101    }
1102
1103    async fn clone_consumer_group_offset(
1104        &self,
1105        source_group: CheetahString,
1106        destination_group: CheetahString,
1107        topic: CheetahString,
1108        offline: bool,
1109    ) -> crate::ClientResult<()> {
1110        MQAdminMutationExt::clone_consumer_group_offset(self.inner(), source_group, destination_group, topic, offline)
1111            .await
1112    }
1113
1114    async fn mutation_cluster_info(&self) -> crate::ClientResult<ClusterInfo> {
1115        MQAdminMutationExt::mutation_cluster_info(self.inner()).await
1116    }
1117
1118    async fn mutation_topic_route(&self, topic: CheetahString) -> crate::ClientResult<Option<TopicRouteData>> {
1119        MQAdminMutationExt::mutation_topic_route(self.inner(), topic).await
1120    }
1121
1122    async fn mutation_topic_config(
1123        &self,
1124        broker_addr: CheetahString,
1125        topic: CheetahString,
1126    ) -> crate::ClientResult<TopicConfig> {
1127        MQAdminMutationExt::mutation_topic_config(self.inner(), broker_addr, topic).await
1128    }
1129
1130    async fn remove_topic_from_brokers(
1131        &self,
1132        broker_addrs: HashSet<CheetahString>,
1133        topic: CheetahString,
1134    ) -> crate::ClientResult<()> {
1135        MQAdminMutationExt::remove_topic_from_brokers(self.inner(), broker_addrs, topic).await
1136    }
1137
1138    async fn remove_topics_from_broker(
1139        &self,
1140        broker_addr: CheetahString,
1141        topics: Vec<CheetahString>,
1142    ) -> crate::ClientResult<()> {
1143        MQAdminMutationExt::remove_topics_from_broker(self.inner(), broker_addr, topics).await
1144    }
1145
1146    async fn remove_topic_from_name_servers(
1147        &self,
1148        namesrv_addrs: HashSet<CheetahString>,
1149        cluster_name: Option<CheetahString>,
1150        topic: CheetahString,
1151    ) -> crate::ClientResult<()> {
1152        MQAdminMutationExt::remove_topic_from_name_servers(self.inner(), namesrv_addrs, cluster_name, topic).await
1153    }
1154
1155    async fn mutation_name_server_addresses(&self) -> crate::ClientResult<Vec<CheetahString>> {
1156        MQAdminMutationExt::mutation_name_server_addresses(self.inner()).await
1157    }
1158
1159    async fn upsert_order_topic_config(
1160        &self,
1161        topic: CheetahString,
1162        value: CheetahString,
1163        cluster_wide: bool,
1164    ) -> crate::ClientResult<()> {
1165        MQAdminMutationExt::upsert_order_topic_config(self.inner(), topic, value, cluster_wide).await
1166    }
1167
1168    async fn mutation_order_topic_config(&self, topic: CheetahString) -> crate::ClientResult<Option<CheetahString>> {
1169        MQAdminMutationExt::mutation_order_topic_config(self.inner(), topic).await
1170    }
1171
1172    async fn delete_order_topic_config(&self, topic: CheetahString) -> crate::ClientResult<()> {
1173        MQAdminMutationExt::delete_order_topic_config(self.inner(), topic).await
1174    }
1175
1176    async fn reset_consumer_offset_legacy(
1177        &self,
1178        cluster_name: Option<CheetahString>,
1179        consumer_group: CheetahString,
1180        topic: CheetahString,
1181        timestamp: u64,
1182        force: bool,
1183    ) -> crate::ClientResult<Vec<RollbackStats>> {
1184        MQAdminMutationExt::reset_consumer_offset_legacy(
1185            self.inner(),
1186            cluster_name,
1187            consumer_group,
1188            topic,
1189            timestamp,
1190            force,
1191        )
1192        .await
1193    }
1194
1195    async fn view_message_for_mutation(
1196        &self,
1197        topic: CheetahString,
1198        message_id: CheetahString,
1199    ) -> crate::ClientResult<MessageExt> {
1200        MQAdminMutationExt::view_message_for_mutation(self.inner(), topic, message_id).await
1201    }
1202}
rocketmq_client_rust/admin/mq_admin_mutation_ext.rs

rocketmq_client_rust/admin/
mq_admin_mutation_ext.rs
