krabka_client_admin/
config_resources.rs

1//! Config resource listing (`ListConfigResources`, KIP-1142), as Kafka's
2//! `Admin.listConfigResources` and `Admin.listClientMetricsResources`.
3//!
4//! `ListConfigResources` v0 is the former `ListClientMetricsResources`
5//! (KIP-714): it takes no resource types and lists client-metrics
6//! subscriptions only. v1 adds the resource types.
7
8use std::collections::BTreeSet;
9
10use krabka_protocol::owned::{
11    list_config_resources_request::ListConfigResourcesRequest,
12    list_config_resources_response::ListConfigResourcesResponse,
13};
14
15use crate::{AdminClient, AdminError, kafka_error_name, retry::ControllerRetry};
16
17/// The type of a config resource, as Kafka's `ConfigResource.Type`.
18#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
19pub enum ConfigResourceType {
20    /// A type that Kafka does not define (`UNKNOWN`, id 0 or any other id).
21    Unknown,
22    /// A topic (id 2).
23    Topic,
24    /// A broker (id 4).
25    Broker,
26    /// The loggers of a broker (id 8).
27    BrokerLogger,
28    /// A client-metrics subscription (id 16, KIP-714).
29    ClientMetrics,
30    /// A group (id 32, KIP-848).
31    Group,
32}
33
34impl ConfigResourceType {
35    /// The wire id of the type.
36    #[must_use]
37    pub const fn id(self) -> i8 {
38        match self {
39            Self::Unknown => 0,
40            Self::Topic => 2,
41            Self::Broker => 4,
42            Self::BrokerLogger => 8,
43            Self::ClientMetrics => 16,
44            Self::Group => 32,
45        }
46    }
47
48    /// The type of a wire id, [`Self::Unknown`] for an id Kafka does not
49    /// define, as `ConfigResource.Type.forId`.
50    #[must_use]
51    pub const fn from_id(id: i8) -> Self {
52        match id {
53            2 => Self::Topic,
54            4 => Self::Broker,
55            8 => Self::BrokerLogger,
56            16 => Self::ClientMetrics,
57            32 => Self::Group,
58            _ => Self::Unknown,
59        }
60    }
61}
62
63/// One config resource, as Kafka's `ConfigResource`.
64#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
65pub struct ConfigResource {
66    /// The type of the resource.
67    pub resource_type: ConfigResourceType,
68    /// The name of the resource.
69    pub name: String,
70}
71
72/// Whether a request for `types` can go as `ListConfigResources` v0, which
73/// Kafka's `ListConfigResourcesRequest.Builder` allows for exactly
74/// `{CLIENT_METRICS}`.
75fn fits_v0(types: &BTreeSet<ConfigResourceType>) -> bool {
76    types.len() == 1 && types.contains(&ConfigResourceType::ClientMetrics)
77}
78
79fn list_request(types: &BTreeSet<ConfigResourceType>) -> ListConfigResourcesRequest {
80    ListConfigResourcesRequest {
81        resource_types: types.iter().map(|kind| kind.id()).collect(),
82        ..Default::default()
83    }
84}
85
86/// The resources of an answer, as Kafka's `ListConfigResourcesResponse`
87/// reads it. A top-level error fails the call.
88fn listings(response: ListConfigResourcesResponse) -> Result<Vec<ConfigResource>, AdminError> {
89    if response.error_code != 0 {
90        return Err(AdminError::Broker {
91            api: "ListConfigResources",
92            code: response.error_code,
93            name: kafka_error_name(response.error_code),
94            message: None,
95        });
96    }
97    Ok(response
98        .config_resources
99        .into_iter()
100        .map(|resource| ConfigResource {
101            resource_type: ConfigResourceType::from_id(resource.resource_type),
102            name: resource.resource_name,
103        })
104        .collect())
105}
106
107impl AdminClient {
108    /// Lists the config resources of `types`, as Kafka's
109    /// `listConfigResources` operation does. An empty set asks for every type
110    /// the broker supports.
111    ///
112    /// The client sends one `ListConfigResources` request on its connection.
113    /// Exactly `{ClientMetrics}` may go as v0, as Kafka's builder allows;
114    /// every other set needs v1 (KIP-1142).
115    ///
116    /// # Errors
117    /// Returns [`AdminError::Broker`] for a top-level error, such as
118    /// `CLUSTER_AUTHORIZATION_FAILED` (31) or `UNSUPPORTED_VERSION` (35) for
119    /// a type the broker does not list, and with `REQUEST_TIMED_OUT` (7) at
120    /// `default.api.timeout.ms` (60 s). Returns [`AdminError::Transport`]
121    /// when the connection fails, or when the broker supports only v0 and
122    /// `types` is not `{ClientMetrics}`.
123    pub async fn list_config_resources(
124        &self,
125        types: &BTreeSet<ConfigResourceType>,
126    ) -> Result<Vec<ConfigResource>, AdminError> {
127        let request = list_request(types);
128        let retry = ControllerRetry::new("ListConfigResources", self.retry);
129        let response = if fits_v0(types) {
130            retry.bounded(self.conn.send(request)).await?
131        } else {
132            retry.bounded(self.conn.send_at_least(request, 1)).await?
133        };
134        listings(response)
135    }
136
137    /// Lists the names of the client-metrics subscriptions (KIP-714), as
138    /// Kafka's `listClientMetricsResources` operation does: a
139    /// `ListConfigResources` request for `{ClientMetrics}`, keeping only the
140    /// resources of that type.
141    ///
142    /// # Errors
143    /// See [`Self::list_config_resources`].
144    pub async fn list_client_metrics_resources(&self) -> Result<Vec<String>, AdminError> {
145        let types = maplit::btreeset! {
146            ConfigResourceType::ClientMetrics,
147        };
148        Ok(self
149            .list_config_resources(&types)
150            .await?
151            .into_iter()
152            .filter(|listing| listing.resource_type == ConfigResourceType::ClientMetrics)
153            .map(|listing| listing.name)
154            .collect())
155    }
156}
157
158#[cfg(test)]
159mod tests {
160    use std::sync::{Arc, Mutex};
161
162    use krabka_client_core::{ClientError, MockReply};
163    use krabka_protocol::owned::{
164        list_config_resources_request,
165        list_config_resources_response::ConfigResource as ListedResource,
166    };
167
168    use super::*;
169    use crate::partition_leaders::test_support::{
170        ApiVersionRange, decode_request, encode_response, fast_admin, scripted_broker,
171    };
172
173    #[test]
174    fn resource_type_ids_are_kafkas() {
175        for (kind, id) in [
176            (ConfigResourceType::Unknown, 0),
177            (ConfigResourceType::Topic, 2),
178            (ConfigResourceType::Broker, 4),
179            (ConfigResourceType::BrokerLogger, 8),
180            (ConfigResourceType::ClientMetrics, 16),
181            (ConfigResourceType::Group, 32),
182        ] {
183            assert2::assert!((kind.id(), ConfigResourceType::from_id(id)) == (id, kind));
184        }
185        for id in [1, 3, 64, -1] {
186            assert2::assert!(ConfigResourceType::from_id(id) == ConfigResourceType::Unknown);
187        }
188    }
189
190    fn resource(resource_type: i8, name: &str) -> ListedResource {
191        ListedResource {
192            resource_name: name.to_owned(),
193            resource_type,
194            ..Default::default()
195        }
196    }
197
198    /// What one `ListConfigResources` call did: its result, and the version
199    /// and request of each request the broker got.
200    type Outcome<T> = (Result<T, String>, Vec<(i16, ListConfigResourcesRequest)>);
201
202    /// Runs `call` against a broker that advertises `ListConfigResources`
203    /// `min..=max` and gives the `n`th request `answers[n]`: a `Close`, or an
204    /// error code with the resources of `resources`.
205    async fn run<T, F>(
206        max_version: i16,
207        answers: Vec<Option<i16>>,
208        resources: Vec<ListedResource>,
209        call: F,
210    ) -> Outcome<T>
211    where
212        F: AsyncFnOnce(&AdminClient) -> Result<T, AdminError>,
213    {
214        let requests = Arc::new(Mutex::new(Vec::new()));
215        let handler_requests = Arc::clone(&requests);
216        let broker = scripted_broker(
217            vec![
218                ApiVersionRange::builder()
219                    .api_key(list_config_resources_request::API_KEY)
220                    .max_version(max_version)
221                    .build(),
222            ],
223            move |api_key, version, body, _| {
224                if api_key != list_config_resources_request::API_KEY {
225                    return MockReply::Silent;
226                }
227                let mut requests = handler_requests.lock().expect("requests lock");
228                let answer = answers[requests.len().min(answers.len() - 1)];
229                let request: ListConfigResourcesRequest = decode_request(body, version, true);
230                requests.push((version, request));
231                drop(requests);
232                let Some(error_code) = answer else {
233                    return MockReply::Close;
234                };
235                MockReply::Respond(encode_response(
236                    &ListConfigResourcesResponse {
237                        error_code,
238                        config_resources: resources.clone(),
239                        ..Default::default()
240                    },
241                    version,
242                    true,
243                ))
244            },
245        )
246        .await;
247        let admin = fast_admin(broker.addr, krabka_units::secs(5)).await;
248        let result = call(&admin).await.map_err(|error| match error {
249            AdminError::Broker { code, .. } => format!("broker {code}"),
250            AdminError::Transport(ClientError::IncompatibleVersion { .. }) => {
251                "incompatible version".to_owned()
252            }
253            other => panic!("unexpected error {other:?}"),
254        });
255        broker.stop();
256        let requests = requests.lock().expect("requests lock").clone();
257        (result, requests)
258    }
259
260    fn request(types: &[i8]) -> ListConfigResourcesRequest {
261        ListConfigResourcesRequest {
262            resource_types: types.to_vec(),
263            ..Default::default()
264        }
265    }
266
267    /// Kafka's `listConfigResources` sends the type ids, maps each resource
268    /// with `ConfigResource.Type.forId`, retries a lost connection, fails the
269    /// call on a top-level error, and needs v1 for any set other than
270    /// `{CLIENT_METRICS}`.
271    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
272    async fn list_config_resources_maps_types_and_gates_v0() {
273        let listed = vec![
274            ConfigResource {
275                resource_type: ConfigResourceType::Topic,
276                name: "orders".to_owned(),
277            },
278            ConfigResource {
279                resource_type: ConfigResourceType::Group,
280                name: "billing".to_owned(),
281            },
282            ConfigResource {
283                resource_type: ConfigResourceType::Unknown,
284                name: "future".to_owned(),
285            },
286        ];
287        let resources = vec![
288            resource(2, "orders"),
289            resource(32, "billing"),
290            resource(64, "future"),
291        ];
292        let topic_and_group = maplit::btreeset! {
293            ConfigResourceType::Topic,
294            ConfigResourceType::Group,
295        };
296        for (name, max_version, types, answers, expected) in [
297            (
298                "v1 sends the types",
299                1,
300                topic_and_group.clone(),
301                vec![Some(0)],
302                (Ok(listed.clone()), vec![(1, request(&[2, 32]))]),
303            ),
304            (
305                "an empty set asks for every type",
306                1,
307                BTreeSet::new(),
308                vec![Some(0)],
309                (Ok(listed.clone()), vec![(1, request(&[]))]),
310            ),
311            (
312                "a lost connection is retried",
313                1,
314                topic_and_group.clone(),
315                vec![None, Some(0)],
316                (
317                    Ok(listed.clone()),
318                    vec![(1, request(&[2, 32])), (1, request(&[2, 32]))],
319                ),
320            ),
321            (
322                "a top-level error fails the call",
323                1,
324                topic_and_group.clone(),
325                vec![Some(31)],
326                (Err("broker 31".to_owned()), vec![(1, request(&[2, 32]))]),
327            ),
328            (
329                "a v0 broker cannot list other types",
330                0,
331                topic_and_group,
332                vec![Some(0)],
333                (Err("incompatible version".to_owned()), vec![]),
334            ),
335        ] {
336            let outcome = run(max_version, answers, resources.clone(), async |admin| {
337                admin.list_config_resources(&types).await
338            })
339            .await;
340            assert2::assert!(outcome == expected, "case {name}");
341        }
342    }
343
344    /// Kafka's `listClientMetricsResources` asks for `{CLIENT_METRICS}`,
345    /// which a v0 broker serves too, and keeps only that type.
346    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
347    async fn list_client_metrics_resources_keeps_client_metrics_only() {
348        let resources = vec![resource(16, "all-metrics"), resource(2, "orders")];
349        for (name, max_version) in [("v0 broker", 0), ("v1 broker", 1)] {
350            let outcome = run(
351                max_version,
352                vec![Some(0)],
353                resources.clone(),
354                async |admin| admin.list_client_metrics_resources().await,
355            )
356            .await;
357            // A v0 request carries no types, so it decodes as an empty list.
358            let sent = if max_version == 0 {
359                request(&[])
360            } else {
361                request(&[16])
362            };
363            let expected = if max_version == 0 {
364                // A v0 answer has no type field, so each resource reads as the
365                // schema default, CLIENT_METRICS.
366                vec!["all-metrics".to_owned(), "orders".to_owned()]
367            } else {
368                vec!["all-metrics".to_owned()]
369            };
370            assert2::assert!(
371                outcome == (Ok(expected), vec![(max_version, sent)]),
372                "case {name}"
373            );
374        }
375    }
376}
krabka_client_admin/config_resources.rs

krabka_client_admin/
config_resources.rs
