Skip to main content

rocketmq_client_rust/admin/
default_mq_admin_ext_impl.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
15#![allow(dead_code)]
16#[cfg(feature = "admin-full")]
17use std::collections::HashMap;
18#[cfg(feature = "admin-full")]
19use std::collections::HashSet;
20use std::sync::Arc;
21#[cfg(feature = "admin-full")]
22use std::sync::OnceLock;
23use std::time::Duration;
24
25#[cfg(feature = "admin-full")]
26use crate::admin::capability::{AuthAdmin, BrokerAdmin, ConsumerAdmin, OffsetAdmin, RouteAdmin, TopicAdmin};
27use crate::base::client_config::ClientConfig;
28#[cfg(feature = "admin-full")]
29use crate::base::validators::Validators;
30#[cfg(feature = "admin-full")]
31use crate::common::admin_tool_result::AdminToolResult;
32#[cfg(feature = "admin-full")]
33use crate::common::admin_tools_result_code_enum::AdminToolsResultCodeEnum;
34#[cfg(feature = "admin-full")]
35use crate::consumer::consumer_impl::pull_request_ext::PullResultExt;
36#[cfg(feature = "admin-full")]
37use crate::consumer::pull_callback::PullCallback;
38#[cfg(feature = "admin-full")]
39use crate::consumer::pull_status::PullStatus;
40use crate::factory::mq_client_instance::MQClientInstance;
41#[cfg(feature = "admin-full")]
42use crate::implementation::communication_mode::CommunicationMode;
43use crate::implementation::mq_client_api_impl::MQClientAPIImpl;
44use crate::implementation::mq_client_manager::ClientPool;
45use crate::implementation::mq_client_manager::ClientPoolToken;
46use crate::runtime::ClientRuntime;
47#[cfg(feature = "admin-full")]
48use crate::ClientError;
49use cheetah_string::CheetahString;
50#[cfg(feature = "admin-full")]
51use rand::seq::IndexedRandom;
52#[cfg(feature = "admin-full")]
53use rocketmq_model::common::attribute::attribute_parser::AttributeParser;
54#[cfg(feature = "admin-full")]
55use rocketmq_model::common::attribute::topic_attributes::TopicAttributes;
56#[cfg(feature = "admin-full")]
57use rocketmq_model::common::attribute::topic_message_type::TopicMessageType;
58#[cfg(feature = "admin-full")]
59use rocketmq_model::common::attribute::Attribute;
60#[cfg(feature = "admin-full")]
61use rocketmq_model::common::base::plain_access_config::PlainAccessConfig;
62use rocketmq_model::common::base::service_state::ServiceState;
63#[cfg(feature = "admin-full")]
64use rocketmq_model::common::config::TopicConfig;
65#[cfg(feature = "admin-full")]
66use rocketmq_model::common::constant::PermName;
67#[cfg(feature = "admin-full")]
68use rocketmq_model::common::message::message_enum::MessageRequestMode;
69#[cfg(feature = "admin-full")]
70use rocketmq_model::common::message::message_ext::MessageExt;
71#[cfg(feature = "admin-full")]
72use rocketmq_model::common::message::message_queue::MessageQueue;
73#[cfg(feature = "admin-full")]
74use rocketmq_model::common::message::MessageConst;
75#[cfg(feature = "admin-full")]
76use rocketmq_model::common::message::MessageTrait;
77#[cfg(feature = "admin-full")]
78use rocketmq_model::common::mix_all;
79#[cfg(feature = "admin-full")]
80use rocketmq_model::common::mix_all::DLQ_GROUP_TOPIC_PREFIX;
81#[cfg(feature = "admin-full")]
82use rocketmq_model::common::mix_all::RETRY_GROUP_TOPIC_PREFIX;
83#[cfg(feature = "admin-full")]
84use rocketmq_model::common::sys_flag::pull_sys_flag::PullSysFlag;
85#[allow(deprecated)]
86#[cfg(feature = "admin-full")]
87use rocketmq_model::common::tools::broker_operator_result::BrokerOperatorResult;
88#[allow(deprecated)]
89#[cfg(feature = "admin-full")]
90use rocketmq_model::common::tools::message_track::MessageTrack;
91#[allow(deprecated)]
92#[cfg(feature = "admin-full")]
93use rocketmq_model::common::tools::track_type::TrackType;
94#[cfg(feature = "admin-full")]
95use rocketmq_model::common::topic::TopicValidator;
96#[cfg(feature = "admin-full")]
97use rocketmq_protocol::code::response_code::ResponseCode;
98#[cfg(feature = "admin-full")]
99use rocketmq_protocol::common::message::message_decoder as MessageDecoder;
100#[cfg(feature = "admin-full")]
101use rocketmq_protocol::protocol::admin::consume_stats::ConsumeStats;
102#[cfg(feature = "admin-full")]
103use rocketmq_protocol::protocol::admin::consume_stats_list::ConsumeStatsList;
104#[cfg(feature = "admin-full")]
105use rocketmq_protocol::protocol::admin::offset_wrapper::OffsetWrapper;
106#[cfg(feature = "admin-full")]
107use rocketmq_protocol::protocol::admin::rollback_stats::RollbackStats;
108#[cfg(feature = "admin-full")]
109use rocketmq_protocol::protocol::admin::topic_offset::TopicOffset;
110#[cfg(feature = "admin-full")]
111use rocketmq_protocol::protocol::admin::topic_stats_table::TopicStatsTable;
112#[cfg(feature = "admin-full")]
113use rocketmq_protocol::protocol::body::acl_info::AclInfo;
114#[cfg(feature = "admin-full")]
115use rocketmq_protocol::protocol::body::acl_info::PolicyEntryInfo;
116#[cfg(feature = "admin-full")]
117use rocketmq_protocol::protocol::body::acl_info::PolicyInfo;
118#[cfg(feature = "admin-full")]
119use rocketmq_protocol::protocol::body::broker_body::broker_member_group::BrokerMemberGroup;
120#[cfg(feature = "admin-full")]
121use rocketmq_protocol::protocol::body::broker_body::cluster_info::ClusterInfo;
122#[cfg(feature = "admin-full")]
123use rocketmq_protocol::protocol::body::broker_replicas_info::BrokerReplicasInfo;
124#[cfg(feature = "admin-full")]
125use rocketmq_protocol::protocol::body::check_rocksdb_cqwrite_progress_response_body::CheckRocksdbCqWriteResult;
126#[cfg(feature = "admin-full")]
127use rocketmq_protocol::protocol::body::consume_message_directly_result::ConsumeMessageDirectlyResult;
128#[cfg(feature = "admin-full")]
129use rocketmq_protocol::protocol::body::consumer_connection::ConsumerConnection;
130#[cfg(feature = "admin-full")]
131use rocketmq_protocol::protocol::body::consumer_running_info::ConsumerRunningInfo;
132#[cfg(feature = "admin-full")]
133use rocketmq_protocol::protocol::body::epoch_entry_cache::EpochEntryCache;
134#[cfg(feature = "admin-full")]
135use rocketmq_protocol::protocol::body::get_broker_lite_info_response_body::GetBrokerLiteInfoResponseBody;
136#[cfg(feature = "admin-full")]
137use rocketmq_protocol::protocol::body::get_lite_client_info_response_body::GetLiteClientInfoResponseBody;
138#[cfg(feature = "admin-full")]
139use rocketmq_protocol::protocol::body::get_lite_group_info_response_body::GetLiteGroupInfoResponseBody;
140#[cfg(feature = "admin-full")]
141use rocketmq_protocol::protocol::body::get_lite_topic_info_response_body::GetLiteTopicInfoResponseBody;
142#[cfg(feature = "admin-full")]
143use rocketmq_protocol::protocol::body::get_parent_topic_info_response_body::GetParentTopicInfoResponseBody;
144#[cfg(feature = "admin-full")]
145use rocketmq_protocol::protocol::body::group_list::GroupList;
146#[cfg(feature = "admin-full")]
147use rocketmq_protocol::protocol::body::ha_runtime_info::HARuntimeInfo;
148#[cfg(feature = "admin-full")]
149use rocketmq_protocol::protocol::body::kv_table::KVTable;
150#[cfg(feature = "admin-full")]
151use rocketmq_protocol::protocol::body::producer_connection::ProducerConnection;
152#[cfg(feature = "admin-full")]
153use rocketmq_protocol::protocol::body::producer_table_info::ProducerTableInfo;
154#[cfg(feature = "admin-full")]
155use rocketmq_protocol::protocol::body::query_consume_queue_response_body::QueryConsumeQueueResponseBody;
156#[cfg(feature = "admin-full")]
157use rocketmq_protocol::protocol::body::queue_time_span::QueueTimeSpan;
158#[cfg(feature = "admin-full")]
159use rocketmq_protocol::protocol::body::subscription_group_wrapper::SubscriptionGroupWrapper;
160#[cfg(feature = "admin-full")]
161use rocketmq_protocol::protocol::body::topic::topic_list::TopicList;
162#[cfg(feature = "admin-full")]
163use rocketmq_protocol::protocol::body::topic_info_wrapper::TopicConfigSerializeWrapper;
164#[cfg(feature = "admin-full")]
165use rocketmq_protocol::protocol::body::user_info::UserInfo;
166#[cfg(feature = "admin-full")]
167use rocketmq_protocol::protocol::header::consume_message_directly_result_request_header::ConsumeMessageDirectlyResultRequestHeader;
168#[cfg(feature = "admin-full")]
169use rocketmq_protocol::protocol::header::create_topic_request_header::CreateTopicRequestHeader;
170#[cfg(feature = "admin-full")]
171use rocketmq_protocol::protocol::header::delete_topic_request_header::DeleteTopicRequestHeader;
172#[cfg(feature = "admin-full")]
173use rocketmq_protocol::protocol::header::elect_master_response_header::ElectMasterResponseHeader;
174#[cfg(feature = "admin-full")]
175use rocketmq_protocol::protocol::header::get_consume_stats_in_broker_header::GetConsumeStatsInBrokerHeader;
176#[cfg(feature = "admin-full")]
177use rocketmq_protocol::protocol::header::get_consume_stats_request_header::GetConsumeStatsRequestHeader;
178#[cfg(feature = "admin-full")]
179use rocketmq_protocol::protocol::header::get_meta_data_response_header::GetMetaDataResponseHeader;
180#[cfg(feature = "admin-full")]
181use rocketmq_protocol::protocol::header::get_topic_stats_info_request_header::GetTopicStatsInfoRequestHeader;
182#[cfg(feature = "admin-full")]
183use rocketmq_protocol::protocol::header::namesrv::brokerid_change_request_header::NotifyMinBrokerIdChangeRequestHeader;
184#[cfg(feature = "admin-full")]
185use rocketmq_protocol::protocol::header::namesrv::topic_operation_header::DeleteTopicFromNamesrvRequestHeader;
186#[cfg(feature = "admin-full")]
187use rocketmq_protocol::protocol::header::namesrv::topic_operation_header::TopicRequestHeader;
188#[cfg(feature = "admin-full")]
189use rocketmq_protocol::protocol::header::pull_message_request_header::PullMessageRequestHeader;
190#[cfg(feature = "admin-full")]
191use rocketmq_protocol::protocol::header::query_consume_time_span_request_header::QueryConsumeTimeSpanRequestHeader;
192#[cfg(feature = "admin-full")]
193use rocketmq_protocol::protocol::header::query_subscription_by_consumer_request_header::QuerySubscriptionByConsumerRequestHeader;
194#[cfg(feature = "admin-full")]
195use rocketmq_protocol::protocol::header::query_topic_consume_by_who_request_header::QueryTopicConsumeByWhoRequestHeader;
196#[cfg(feature = "admin-full")]
197use rocketmq_protocol::protocol::header::query_topics_by_consumer_request_header::QueryTopicsByConsumerRequestHeader;
198#[cfg(feature = "admin-full")]
199use rocketmq_protocol::protocol::header::reset_offset_request_header::ResetOffsetRequestHeader;
200#[cfg(feature = "admin-full")]
201use rocketmq_protocol::protocol::header::update_consumer_offset_header::UpdateConsumerOffsetRequestHeader;
202#[cfg(feature = "admin-full")]
203use rocketmq_protocol::protocol::header::update_group_forbidden_request_header::UpdateGroupForbiddenRequestHeader;
204#[cfg(feature = "admin-full")]
205use rocketmq_protocol::protocol::header::view_broker_stats_data_request_header::ViewBrokerStatsDataRequestHeader;
206#[cfg(feature = "admin-full")]
207use rocketmq_protocol::protocol::header::view_message_request_header::ViewMessageRequestHeader;
208#[cfg(feature = "admin-full")]
209use rocketmq_protocol::protocol::heartbeat::consume_type::ConsumeType;
210#[cfg(feature = "admin-full")]
211use rocketmq_protocol::protocol::heartbeat::message_model::MessageModel;
212#[cfg(feature = "admin-full")]
213use rocketmq_protocol::protocol::heartbeat::subscription_data::SubscriptionData;
214#[cfg(feature = "admin-full")]
215use rocketmq_protocol::protocol::route::route_data_view::QueueData;
216#[cfg(feature = "admin-full")]
217use rocketmq_protocol::protocol::route::topic_route_data::TopicRouteData;
218#[cfg(feature = "admin-full")]
219use rocketmq_protocol::protocol::route_facade::BrokerDataExt;
220#[cfg(feature = "admin-full")]
221use rocketmq_protocol::protocol::static_topic::topic_queue_mapping_detail::TopicQueueMappingDetail;
222#[cfg(feature = "admin-full")]
223use rocketmq_protocol::protocol::subscription::broker_stats_data::BrokerStatsData;
224#[cfg(feature = "admin-full")]
225use rocketmq_protocol::protocol::subscription::group_forbidden::GroupForbidden;
226#[cfg(feature = "admin-full")]
227use rocketmq_protocol::protocol::subscription::subscription_group_config::SubscriptionGroupConfig;
228use rocketmq_transport::api::RPCHook;
229#[cfg(feature = "admin-full")]
230use rocketmq_transport::api::RpcRequestHeader;
231#[cfg(feature = "admin-full")]
232use tracing::info;
233#[cfg(feature = "admin-full")]
234use tracing::warn;
235
236#[cfg(feature = "admin-full")]
237static SYSTEM_GROUP_SET: OnceLock<HashSet<CheetahString>> = OnceLock::new();
238
239const SOCKS_PROXY_JSON: &str = "socksProxyJson";
240const NAMESPACE_ORDER_TOPIC_CONFIG: &str = "ORDER_TOPIC_CONFIG";
241#[cfg(feature = "admin-full")]
242const ROCKSDB_CONFIG_TYPE_CONSUMER_OFFSETS: &str = "consumerOffsets";
243
244pub struct DefaultMQAdminExtImpl {
245    client_pool: ClientPool,
246    client_pool_token: Option<ClientPoolToken>,
247    service_state: ServiceState,
248    client_instance: Option<Arc<MQClientInstance>>,
249    rpc_hook: Option<Arc<dyn RPCHook>>,
250    timeout_millis: Duration,
251    kv_namespace_to_delete_list: Vec<CheetahString>,
252    client_config: ClientConfig,
253    admin_ext_group: CheetahString,
254}
255
256#[cfg(feature = "admin-full")]
257mod admin_api;
258#[cfg(feature = "admin-full")]
259mod broker;
260#[cfg(feature = "admin-read")]
261mod controller_read;
262#[cfg(feature = "admin-full")]
263mod group;
264mod lifecycle;
265#[cfg(feature = "admin-mutation")]
266mod mutation_api;
267#[cfg(feature = "admin-full")]
268mod security;
269#[cfg(feature = "admin-full")]
270mod topic;
271
272#[cfg(all(test, feature = "admin-full"))]
273#[path = "../../tests/admin/default_mq_admin_ext_impl/unit.rs"]
274mod tests;