kafrust_protocol/api/
list_config_resources.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5/// Kafka ListConfigResources API key.
6pub const API_KEY: i16 = 74;
7
8/// Kafka configuration resource type for topics.
9pub const TOPIC_RESOURCE_TYPE: i8 = 2;
10/// Kafka configuration resource type for brokers.
11pub const BROKER_RESOURCE_TYPE: i8 = 4;
12/// Kafka configuration resource type for broker loggers.
13pub const BROKER_LOGGER_RESOURCE_TYPE: i8 = 8;
14/// Kafka configuration resource type for client metrics.
15pub const CLIENT_METRICS_RESOURCE_TYPE: i8 = 16;
16/// Kafka configuration resource type for consumer groups.
17pub const GROUP_RESOURCE_TYPE: i8 = 32;
18
19/// Kafka ListConfigResources v1 request.
20///
21/// Version 0 uses the same API key for the older ListClientMetricsResources
22/// operation and has no resource-type request field.
23#[derive(Debug, Clone, PartialEq, Eq)]
24pub struct ListConfigResourcesRequestV0 {
25    pub correlation_id: i32,
26    pub client_id: Option<String>,
27}
28
29impl ListConfigResourcesRequestV0 {
30    /// Encodes the flexible v0 request header and empty body.
31    pub fn encode(&self) -> Result<Vec<u8>> {
32        let mut encoder = Encoder::new();
33        RequestHeader {
34            api_key: API_KEY,
35            api_version: 0,
36            correlation_id: self.correlation_id,
37            client_id: self.client_id.clone(),
38        }
39        .encode_v2(&mut encoder)?;
40        encoder.write_empty_tagged_fields();
41        Ok(encoder.into_bytes())
42    }
43}
44
45/// Kafka ListClientMetricsResources v0 response carried by API key 74.
46#[derive(Debug, Clone, PartialEq, Eq)]
47pub struct ListConfigResourcesResponseV0 {
48    pub throttle_time_ms: i32,
49    pub error_code: i16,
50    pub resources: Vec<ListedClientMetricsResourceV0>,
51}
52
53impl ListConfigResourcesResponseV0 {
54    /// Decodes the flexible v0 response body.
55    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
56        let throttle_time_ms = decoder.read_i32()?;
57        let error_code = decoder.read_i16()?;
58        let resources = decoder
59            .read_compact_array("list client metrics resources", |decoder| {
60                let name = decoder.read_compact_string()?;
61                decoder.read_tagged_fields()?;
62                Ok(ListedClientMetricsResourceV0 { name })
63            })?
64            .unwrap_or_default();
65        decoder.read_tagged_fields()?;
66        Ok(Self {
67            throttle_time_ms,
68            error_code,
69            resources,
70        })
71    }
72}
73
74/// One client-metrics resource returned by the API 74 v0 compatibility path.
75#[derive(Debug, Clone, PartialEq, Eq)]
76pub struct ListedClientMetricsResourceV0 {
77    pub name: String,
78}
79
80/// Kafka ListConfigResources v1 request.
81#[derive(Debug, Clone, PartialEq, Eq)]
82pub struct ListConfigResourcesRequestV1 {
83    pub correlation_id: i32,
84    pub client_id: Option<String>,
85    /// Empty requests ask Kafka for every supported configuration resource type.
86    pub resource_types: Vec<i8>,
87}
88
89impl ListConfigResourcesRequestV1 {
90    /// Encodes the flexible request header and body.
91    pub fn encode(&self) -> Result<Vec<u8>> {
92        let mut encoder = Encoder::new();
93        RequestHeader {
94            api_key: API_KEY,
95            api_version: 1,
96            correlation_id: self.correlation_id,
97            client_id: self.client_id.clone(),
98        }
99        .encode_v2(&mut encoder)?;
100        encoder.write_compact_array(Some(&self.resource_types), |encoder, resource_type| {
101            encoder.write_i8(*resource_type);
102            Ok(())
103        })?;
104        encoder.write_empty_tagged_fields();
105        Ok(encoder.into_bytes())
106    }
107}
108
109/// Kafka ListConfigResources v1 response.
110#[derive(Debug, Clone, PartialEq, Eq)]
111pub struct ListConfigResourcesResponseV1 {
112    pub throttle_time_ms: i32,
113    pub error_code: i16,
114    pub resources: Vec<ListedConfigResourceV1>,
115}
116
117impl ListConfigResourcesResponseV1 {
118    /// Decodes a flexible response body.
119    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
120        let throttle_time_ms = decoder.read_i32()?;
121        let error_code = decoder.read_i16()?;
122        let resources = decoder
123            .read_compact_array("list config resources", |decoder| {
124                let resource_name = decoder.read_compact_string()?;
125                let resource_type = decoder.read_i8()?;
126                decoder.read_tagged_fields()?;
127                Ok(ListedConfigResourceV1 {
128                    resource_name,
129                    resource_type,
130                })
131            })?
132            .unwrap_or_default();
133        decoder.read_tagged_fields()?;
134        Ok(Self {
135            throttle_time_ms,
136            error_code,
137            resources,
138        })
139    }
140}
141
142/// One configuration resource returned by Kafka.
143#[derive(Debug, Clone, PartialEq, Eq)]
144pub struct ListedConfigResourceV1 {
145    pub resource_name: String,
146    pub resource_type: i8,
147}
148
149#[cfg(test)]
150#[allow(clippy::unwrap_used)]
151mod tests {
152    use super::{
153        ListConfigResourcesRequestV0, ListConfigResourcesRequestV1, ListConfigResourcesResponseV0,
154        ListConfigResourcesResponseV1, API_KEY, BROKER_RESOURCE_TYPE, GROUP_RESOURCE_TYPE,
155        TOPIC_RESOURCE_TYPE,
156    };
157    use crate::codec::{Decoder, Encoder};
158
159    #[test]
160    fn encodes_list_config_resources_v1_request() {
161        let request = ListConfigResourcesRequestV1 {
162            correlation_id: 12,
163            client_id: Some("kafrust".to_owned()),
164            resource_types: vec![TOPIC_RESOURCE_TYPE, GROUP_RESOURCE_TYPE],
165        };
166
167        assert_eq!(
168            request.encode().unwrap(),
169            [
170                0, 74, // API key
171                0, 1, // API version
172                0, 0, 0, 12, // correlation ID
173                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client ID
174                0,    // request header tagged fields
175                3,    // compact resource type count
176                2, 32, // topic and group
177                0,  // request tagged fields
178            ]
179        );
180        assert_eq!(API_KEY, 74);
181    }
182
183    #[test]
184    fn encodes_list_client_metrics_resources_v0_request() {
185        let request = ListConfigResourcesRequestV0 {
186            correlation_id: 13,
187            client_id: None,
188        };
189
190        assert_eq!(
191            request.encode().unwrap(),
192            [
193                0, 74, // API key
194                0, 0, // API version
195                0, 0, 0, 13, // correlation ID
196                0xff, 0xff, // nullable client ID
197                0,    // request header tagged fields
198                0,    // empty v0 request body tagged fields
199            ]
200        );
201    }
202
203    #[test]
204    fn decodes_list_config_resources_v1_response() {
205        let mut body = Encoder::new();
206        body.write_i32(9);
207        body.write_i16(0);
208        body.write_compact_array(
209            Some(&["orders".to_owned(), "payments".to_owned()]),
210            |encoder, name| {
211                encoder.write_compact_string(name)?;
212                encoder.write_i8(if name == "orders" {
213                    TOPIC_RESOURCE_TYPE
214                } else {
215                    BROKER_RESOURCE_TYPE
216                });
217                encoder.write_empty_tagged_fields();
218                Ok(())
219            },
220        )
221        .unwrap();
222        body.write_empty_tagged_fields();
223
224        let bytes = body.into_bytes();
225        let mut decoder = Decoder::new(&bytes);
226        let response = ListConfigResourcesResponseV1::decode_body(&mut decoder).unwrap();
227
228        assert_eq!(response.throttle_time_ms, 9);
229        assert_eq!(response.error_code, 0);
230        assert_eq!(response.resources.len(), 2);
231        assert_eq!(response.resources[0].resource_name, "orders");
232        assert_eq!(response.resources[0].resource_type, TOPIC_RESOURCE_TYPE);
233        assert_eq!(response.resources[1].resource_type, BROKER_RESOURCE_TYPE);
234        assert!(decoder.is_empty());
235    }
236
237    #[test]
238    fn decodes_list_client_metrics_resources_v0_response() {
239        let mut body = Encoder::new();
240        body.write_i32(5);
241        body.write_i16(0);
242        body.write_compact_array(Some(&["latency", "throughput"]), |encoder, name| {
243            encoder.write_compact_string(name)?;
244            encoder.write_empty_tagged_fields();
245            Ok(())
246        })
247        .unwrap();
248        body.write_empty_tagged_fields();
249
250        let bytes = body.into_bytes();
251        let mut decoder = Decoder::new(&bytes);
252        let response = ListConfigResourcesResponseV0::decode_body(&mut decoder).unwrap();
253
254        assert_eq!(response.throttle_time_ms, 5);
255        assert_eq!(response.error_code, 0);
256        assert_eq!(response.resources.len(), 2);
257        assert_eq!(response.resources[0].name, "latency");
258        assert_eq!(response.resources[1].name, "throughput");
259        assert!(decoder.is_empty());
260    }
261}
kafrust_protocol/api/list_config_resources.rs

kafrust_protocol/api/
list_config_resources.rs
