kafrust_protocol/api/
describe_configs.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 32;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct DescribeConfigsRequestV1 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub resources: Vec<DescribeConfigsResourceV1>,
12    pub include_synonyms: bool,
13}
14
15impl DescribeConfigsRequestV1 {
16    pub fn encode(&self) -> Result<Vec<u8>> {
17        let mut encoder = Encoder::new();
18        RequestHeader {
19            api_key: API_KEY,
20            api_version: 1,
21            correlation_id: self.correlation_id,
22            client_id: self.client_id.clone(),
23        }
24        .encode_v1(&mut encoder)?;
25        encoder.write_array(Some(&self.resources), |encoder, resource| {
26            resource.encode(encoder)
27        })?;
28        encoder.write_bool(self.include_synonyms);
29        Ok(encoder.into_bytes())
30    }
31}
32
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub struct DescribeConfigsResourceV1 {
35    pub resource_type: i8,
36    pub resource_name: String,
37    pub configuration_keys: Option<Vec<String>>,
38}
39
40impl DescribeConfigsResourceV1 {
41    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
42        encoder.write_i8(self.resource_type);
43        encoder.write_string(&self.resource_name)?;
44        encoder.write_array(
45            self.configuration_keys.as_deref(),
46            |encoder, configuration_key| encoder.write_string(configuration_key),
47        )
48    }
49}
50
51/// Kafka DescribeConfigs v4 request.
52#[derive(Debug, Clone, PartialEq, Eq)]
53pub struct DescribeConfigsRequestV4 {
54    pub correlation_id: i32,
55    pub client_id: Option<String>,
56    pub resources: Vec<DescribeConfigsResourceV4>,
57    pub include_synonyms: bool,
58    pub include_documentation: bool,
59}
60
61impl DescribeConfigsRequestV4 {
62    /// Encodes the flexible request header and body.
63    pub fn encode(&self) -> Result<Vec<u8>> {
64        let mut encoder = Encoder::new();
65        RequestHeader {
66            api_key: API_KEY,
67            api_version: 4,
68            correlation_id: self.correlation_id,
69            client_id: self.client_id.clone(),
70        }
71        .encode_v2(&mut encoder)?;
72        encoder.write_compact_array(Some(&self.resources), |encoder, resource| {
73            encoder.write_i8(resource.resource_type);
74            encoder.write_compact_string(&resource.resource_name)?;
75            encoder.write_compact_array(
76                resource.configuration_keys.as_deref(),
77                |encoder, configuration_key| encoder.write_compact_string(configuration_key),
78            )?;
79            encoder.write_empty_tagged_fields();
80            Ok(())
81        })?;
82        encoder.write_bool(self.include_synonyms);
83        encoder.write_bool(self.include_documentation);
84        encoder.write_empty_tagged_fields();
85        Ok(encoder.into_bytes())
86    }
87}
88
89/// One resource in a DescribeConfigs v4 request.
90#[derive(Debug, Clone, PartialEq, Eq)]
91pub struct DescribeConfigsResourceV4 {
92    pub resource_type: i8,
93    pub resource_name: String,
94    pub configuration_keys: Option<Vec<String>>,
95}
96
97#[derive(Debug, Clone, PartialEq, Eq)]
98pub struct DescribeConfigsResponseV1 {
99    pub throttle_time_ms: i32,
100    pub results: Vec<DescribeConfigsResultV1>,
101}
102
103impl DescribeConfigsResponseV1 {
104    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
105        Ok(Self {
106            throttle_time_ms: decoder.read_i32()?,
107            results: decoder
108                .read_array("describe configs results", DescribeConfigsResultV1::decode)?
109                .unwrap_or_default(),
110        })
111    }
112}
113
114/// Kafka DescribeConfigs v4 response.
115#[derive(Debug, Clone, PartialEq, Eq)]
116pub struct DescribeConfigsResponseV4 {
117    pub throttle_time_ms: i32,
118    pub results: Vec<DescribeConfigsResultV4>,
119}
120
121impl DescribeConfigsResponseV4 {
122    /// Decodes a flexible response body.
123    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
124        let throttle_time_ms = decoder.read_i32()?;
125        let results = decoder
126            .read_compact_array("describe configs v4 results", |decoder| {
127                let error_code = decoder.read_i16()?;
128                let error_message = decoder.read_compact_nullable_string()?;
129                let resource_type = decoder.read_i8()?;
130                let resource_name = decoder.read_compact_string()?;
131                let configs = decoder
132                    .read_compact_array("describe configs v4 entries", |decoder| {
133                        let name = decoder.read_compact_string()?;
134                        let value = decoder.read_compact_nullable_string()?;
135                        let read_only = decoder.read_bool()?;
136                        let config_source = decoder.read_i8()?;
137                        let is_sensitive = decoder.read_bool()?;
138                        let synonyms = decoder
139                            .read_compact_array(
140                                "describe configs v4 synonyms",
141                                DescribeConfigsSynonymV4::decode,
142                            )?
143                            .unwrap_or_default();
144                        let config_type = decoder.read_i8()?;
145                        let documentation = decoder.read_compact_nullable_string()?;
146                        decoder.read_tagged_fields()?;
147                        Ok(DescribeConfigsEntryV4 {
148                            name,
149                            value,
150                            read_only,
151                            config_source,
152                            is_sensitive,
153                            synonyms,
154                            config_type,
155                            documentation,
156                        })
157                    })?
158                    .unwrap_or_default();
159                decoder.read_tagged_fields()?;
160                Ok(DescribeConfigsResultV4 {
161                    error_code,
162                    error_message,
163                    resource_type,
164                    resource_name,
165                    configs,
166                })
167            })?
168            .unwrap_or_default();
169        decoder.read_tagged_fields()?;
170        Ok(Self {
171            throttle_time_ms,
172            results,
173        })
174    }
175}
176
177/// One resource result in a DescribeConfigs v4 response.
178#[derive(Debug, Clone, PartialEq, Eq)]
179pub struct DescribeConfigsResultV4 {
180    pub error_code: i16,
181    pub error_message: Option<String>,
182    pub resource_type: i8,
183    pub resource_name: String,
184    pub configs: Vec<DescribeConfigsEntryV4>,
185}
186
187/// One configuration entry in a DescribeConfigs v4 response.
188#[derive(Debug, Clone, PartialEq, Eq)]
189pub struct DescribeConfigsEntryV4 {
190    pub name: String,
191    pub value: Option<String>,
192    pub read_only: bool,
193    pub config_source: i8,
194    pub is_sensitive: bool,
195    pub synonyms: Vec<DescribeConfigsSynonymV4>,
196    pub config_type: i8,
197    pub documentation: Option<String>,
198}
199
200/// One synonym in a DescribeConfigs v4 response.
201#[derive(Debug, Clone, PartialEq, Eq)]
202pub struct DescribeConfigsSynonymV4 {
203    pub name: String,
204    pub value: Option<String>,
205    pub source: i8,
206}
207
208impl DescribeConfigsSynonymV4 {
209    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
210        let name = decoder.read_compact_string()?;
211        let value = decoder.read_compact_nullable_string()?;
212        let source = decoder.read_i8()?;
213        decoder.read_tagged_fields()?;
214        Ok(Self {
215            name,
216            value,
217            source,
218        })
219    }
220}
221
222#[derive(Debug, Clone, PartialEq, Eq)]
223pub struct DescribeConfigsResultV1 {
224    pub error_code: i16,
225    pub error_message: Option<String>,
226    pub resource_type: i8,
227    pub resource_name: String,
228    pub configs: Vec<DescribeConfigsEntryV1>,
229}
230
231impl DescribeConfigsResultV1 {
232    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
233        Ok(Self {
234            error_code: decoder.read_i16()?,
235            error_message: decoder.read_nullable_string()?,
236            resource_type: decoder.read_i8()?,
237            resource_name: decoder.read_string()?,
238            configs: decoder
239                .read_array("describe configs entries", DescribeConfigsEntryV1::decode)?
240                .unwrap_or_default(),
241        })
242    }
243}
244
245#[derive(Debug, Clone, PartialEq, Eq)]
246pub struct DescribeConfigsEntryV1 {
247    pub name: String,
248    pub value: Option<String>,
249    pub read_only: bool,
250    pub config_source: i8,
251    pub is_sensitive: bool,
252    pub synonyms: Vec<DescribeConfigsSynonymV1>,
253}
254
255impl DescribeConfigsEntryV1 {
256    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
257        Ok(Self {
258            name: decoder.read_string()?,
259            value: decoder.read_nullable_string()?,
260            read_only: decoder.read_bool()?,
261            config_source: decoder.read_i8()?,
262            is_sensitive: decoder.read_bool()?,
263            synonyms: decoder
264                .read_array(
265                    "describe configs synonyms",
266                    DescribeConfigsSynonymV1::decode,
267                )?
268                .unwrap_or_default(),
269        })
270    }
271}
272
273#[derive(Debug, Clone, PartialEq, Eq)]
274pub struct DescribeConfigsSynonymV1 {
275    pub name: String,
276    pub value: Option<String>,
277    pub source: i8,
278}
279
280impl DescribeConfigsSynonymV1 {
281    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
282        Ok(Self {
283            name: decoder.read_string()?,
284            value: decoder.read_nullable_string()?,
285            source: decoder.read_i8()?,
286        })
287    }
288}
289
290#[cfg(test)]
291#[allow(clippy::unwrap_used)]
292mod tests {
293    use super::{
294        DescribeConfigsRequestV1, DescribeConfigsRequestV4, DescribeConfigsResourceV1,
295        DescribeConfigsResourceV4, DescribeConfigsResponseV1, DescribeConfigsResponseV4, API_KEY,
296    };
297    use crate::codec::{Decoder, Encoder};
298
299    #[test]
300    fn encodes_describe_configs_v1_request() {
301        let request = DescribeConfigsRequestV1 {
302            correlation_id: 12,
303            client_id: Some("kafrust".to_owned()),
304            resources: vec![DescribeConfigsResourceV1 {
305                resource_type: 2,
306                resource_name: "orders".to_owned(),
307                configuration_keys: Some(vec!["cleanup.policy".to_owned()]),
308            }],
309            include_synonyms: true,
310        };
311
312        assert_eq!(
313            request.encode().unwrap(),
314            [
315                0, 32, // API key
316                0, 1, // API version
317                0, 0, 0, 12, // correlation ID
318                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client ID
319                0, 0, 0, 1, // resource count
320                2, // topic resource
321                0, 6, b'o', b'r', b'd', b'e', b'r', b's', // resource name
322                0, 0, 0, 1, // configuration key count
323                0, 14, b'c', b'l', b'e', b'a', b'n', b'u', b'p', b'.', b'p', b'o', b'l', b'i',
324                b'c', b'y', // configuration key
325                1,    // include synonyms
326            ]
327        );
328        assert_eq!(API_KEY, 32);
329    }
330
331    #[test]
332    fn decodes_describe_configs_v1_response() {
333        let bytes = [
334            0, 0, 0, 9, // throttle time
335            0, 0, 0, 1, // result count
336            0, 0, // success
337            0xff, 0xff, // null error message
338            2,    // topic resource
339            0, 6, b'o', b'r', b'd', b'e', b'r', b's', // resource name
340            0, 0, 0, 1, // config count
341            0, 14, b'c', b'l', b'e', b'a', b'n', b'u', b'p', b'.', b'p', b'o', b'l', b'i', b'c',
342            b'y', // config name
343            0, 7, b'c', b'o', b'm', b'p', b'a', b'c', b't', // value
344            0,    // read only
345            1,    // dynamic topic config source
346            0,    // not sensitive
347            0, 0, 0, 1, // synonym count
348            0, 14, b'c', b'l', b'e', b'a', b'n', b'u', b'p', b'.', b'p', b'o', b'l', b'i', b'c',
349            b'y', // synonym name
350            0, 6, b'd', b'e', b'l', b'e', b't', b'e', // synonym value
351            5,    // default config source
352        ];
353        let mut decoder = Decoder::new(&bytes);
354
355        let response = DescribeConfigsResponseV1::decode_body(&mut decoder).unwrap();
356
357        assert_eq!(response.throttle_time_ms, 9);
358        assert_eq!(response.results.len(), 1);
359        let result = &response.results[0];
360        assert_eq!(result.resource_type, 2);
361        assert_eq!(result.resource_name, "orders");
362        assert_eq!(result.configs.len(), 1);
363        assert_eq!(result.configs[0].name, "cleanup.policy");
364        assert_eq!(result.configs[0].value.as_deref(), Some("compact"));
365        assert_eq!(result.configs[0].config_source, 1);
366        assert_eq!(result.configs[0].synonyms[0].source, 5);
367        assert!(decoder.is_empty());
368    }
369
370    #[test]
371    fn encodes_describe_configs_v4_request_with_documentation() {
372        let request = DescribeConfigsRequestV4 {
373            correlation_id: 12,
374            client_id: Some("kafrust".to_owned()),
375            resources: vec![DescribeConfigsResourceV4 {
376                resource_type: 2,
377                resource_name: "orders".to_owned(),
378                configuration_keys: Some(vec!["cleanup.policy".to_owned()]),
379            }],
380            include_synonyms: true,
381            include_documentation: true,
382        };
383
384        let bytes = request.encode().unwrap();
385        assert_eq!(&bytes[0..4], &[0, API_KEY as u8, 0, 4]);
386        assert_eq!(bytes[17], 0); // request header tagged fields
387        assert_eq!(bytes[18], 2); // one compact resource
388        assert!(bytes.ends_with(&[1, 1, 0]));
389    }
390
391    #[test]
392    fn decodes_describe_configs_v4_response_with_documentation() {
393        let mut body = Encoder::new();
394        body.write_i32(9);
395        body.write_compact_array(Some(&[()]), |encoder, ()| {
396            encoder.write_i16(0);
397            encoder.write_compact_nullable_string(None)?;
398            encoder.write_i8(2);
399            encoder.write_compact_string("orders")?;
400            encoder.write_compact_array(Some(&[()]), |encoder, ()| {
401                encoder.write_compact_string("cleanup.policy")?;
402                encoder.write_compact_nullable_string(Some("compact"))?;
403                encoder.write_bool(false);
404                encoder.write_i8(1);
405                encoder.write_bool(false);
406                encoder.write_compact_array(Some(&[()]), |encoder, ()| {
407                    encoder.write_compact_string("cleanup.policy")?;
408                    encoder.write_compact_nullable_string(Some("delete"))?;
409                    encoder.write_i8(5);
410                    encoder.write_empty_tagged_fields();
411                    Ok(())
412                })?;
413                encoder.write_i8(7);
414                encoder.write_compact_nullable_string(Some("The cleanup policy."))?;
415                encoder.write_empty_tagged_fields();
416                Ok(())
417            })?;
418            encoder.write_empty_tagged_fields();
419            Ok(())
420        })
421        .unwrap();
422        body.write_empty_tagged_fields();
423
424        let bytes = body.into_bytes();
425        let mut decoder = Decoder::new(&bytes);
426        let response = DescribeConfigsResponseV4::decode_body(&mut decoder).unwrap();
427
428        let entry = &response.results[0].configs[0];
429        assert_eq!(entry.config_type, 7);
430        assert_eq!(entry.documentation.as_deref(), Some("The cleanup policy."));
431        assert_eq!(entry.synonyms[0].source, 5);
432        assert!(decoder.is_empty());
433    }
434}
kafrust_protocol/api/describe_configs.rs

kafrust_protocol/api/
describe_configs.rs
