kafrust_protocol/api/
metadata.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 3;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct MetadataRequestV1 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub topics: Option<Vec<String>>,
12}
13
14impl MetadataRequestV1 {
15    pub fn encode(&self) -> Result<Vec<u8>> {
16        let mut encoder = Encoder::new();
17        RequestHeader {
18            api_key: API_KEY,
19            api_version: 1,
20            correlation_id: self.correlation_id,
21            client_id: self.client_id.clone(),
22        }
23        .encode_v1(&mut encoder)?;
24        encoder.write_array(self.topics.as_deref(), |encoder, topic| {
25            encoder.write_string(topic)
26        })?;
27        Ok(encoder.into_bytes())
28    }
29}
30
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct BrokerMetadata {
33    pub node_id: i32,
34    pub host: String,
35    pub port: i32,
36    pub rack: Option<String>,
37}
38
39impl BrokerMetadata {
40    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
41        Ok(Self {
42            node_id: decoder.read_i32()?,
43            host: decoder.read_string()?,
44            port: decoder.read_i32()?,
45            rack: decoder.read_nullable_string()?,
46        })
47    }
48}
49
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub struct PartitionMetadata {
52    pub error_code: i16,
53    pub partition_index: i32,
54    pub leader_id: i32,
55    pub replica_nodes: Vec<i32>,
56    pub isr_nodes: Vec<i32>,
57}
58
59impl PartitionMetadata {
60    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
61        Ok(Self {
62            error_code: decoder.read_i16()?,
63            partition_index: decoder.read_i32()?,
64            leader_id: decoder.read_i32()?,
65            replica_nodes: decoder
66                .read_array("replica nodes", |decoder| decoder.read_i32())?
67                .unwrap_or_default(),
68            isr_nodes: decoder
69                .read_array("isr nodes", |decoder| decoder.read_i32())?
70                .unwrap_or_default(),
71        })
72    }
73}
74
75#[derive(Debug, Clone, PartialEq, Eq)]
76pub struct TopicMetadata {
77    pub error_code: i16,
78    pub name: String,
79    pub is_internal: bool,
80    pub partitions: Vec<PartitionMetadata>,
81}
82
83impl TopicMetadata {
84    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
85        Ok(Self {
86            error_code: decoder.read_i16()?,
87            name: decoder.read_string()?,
88            is_internal: decoder.read_bool()?,
89            partitions: decoder
90                .read_array("partitions", PartitionMetadata::decode)?
91                .unwrap_or_default(),
92        })
93    }
94}
95
96#[derive(Debug, Clone, PartialEq, Eq)]
97pub struct MetadataResponseV1 {
98    pub brokers: Vec<BrokerMetadata>,
99    pub controller_id: i32,
100    pub topics: Vec<TopicMetadata>,
101}
102
103impl MetadataResponseV1 {
104    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
105        Ok(Self {
106            brokers: decoder
107                .read_array("brokers", BrokerMetadata::decode)?
108                .unwrap_or_default(),
109            controller_id: decoder.read_i32()?,
110            topics: decoder
111                .read_array("topics", TopicMetadata::decode)?
112                .unwrap_or_default(),
113        })
114    }
115}
116
117/// Topic selector used by Metadata v12.
118#[derive(Debug, Clone, PartialEq, Eq)]
119pub struct MetadataRequestTopicV12 {
120    pub topic_id: [u8; 16],
121    pub name: Option<String>,
122}
123
124impl MetadataRequestTopicV12 {
125    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
126        encoder.write_uuid(&self.topic_id);
127        encoder.write_compact_nullable_string(self.name.as_deref())?;
128        encoder.write_empty_tagged_fields();
129        Ok(())
130    }
131}
132
133/// Metadata v12 request with topic UUID support.
134#[derive(Debug, Clone, PartialEq, Eq)]
135pub struct MetadataRequestV12 {
136    pub correlation_id: i32,
137    pub client_id: Option<String>,
138    pub topics: Option<Vec<MetadataRequestTopicV12>>,
139    pub allow_auto_topic_creation: bool,
140    pub include_topic_authorized_operations: bool,
141}
142
143impl MetadataRequestV12 {
144    pub fn encode(&self) -> Result<Vec<u8>> {
145        let mut encoder = Encoder::new();
146        RequestHeader {
147            api_key: API_KEY,
148            api_version: 12,
149            correlation_id: self.correlation_id,
150            client_id: self.client_id.clone(),
151        }
152        .encode_v2(&mut encoder)?;
153        encoder.write_compact_array(self.topics.as_deref(), |encoder, topic| {
154            topic.encode(encoder)
155        })?;
156        encoder.write_bool(self.allow_auto_topic_creation);
157        encoder.write_bool(self.include_topic_authorized_operations);
158        encoder.write_empty_tagged_fields();
159        Ok(encoder.into_bytes())
160    }
161}
162
163/// Broker endpoint returned by Metadata v12.
164#[derive(Debug, Clone, PartialEq, Eq)]
165pub struct MetadataBrokerV12 {
166    pub node_id: i32,
167    pub host: String,
168    pub port: i32,
169    pub rack: Option<String>,
170}
171
172impl MetadataBrokerV12 {
173    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
174        let broker = Self {
175            node_id: decoder.read_i32()?,
176            host: decoder.read_compact_string()?,
177            port: decoder.read_i32()?,
178            rack: decoder.read_compact_nullable_string()?,
179        };
180        decoder.read_tagged_fields()?;
181        Ok(broker)
182    }
183}
184
185/// Partition metadata returned by Metadata v12.
186#[derive(Debug, Clone, PartialEq, Eq)]
187pub struct MetadataPartitionV12 {
188    pub error_code: i16,
189    pub partition_index: i32,
190    pub leader_id: i32,
191    pub leader_epoch: i32,
192    pub replica_nodes: Vec<i32>,
193    pub isr_nodes: Vec<i32>,
194    pub offline_replicas: Vec<i32>,
195}
196
197impl MetadataPartitionV12 {
198    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
199        let partition = Self {
200            error_code: decoder.read_i16()?,
201            partition_index: decoder.read_i32()?,
202            leader_id: decoder.read_i32()?,
203            leader_epoch: decoder.read_i32()?,
204            replica_nodes: decoder
205                .read_compact_array("metadata v12 replica nodes", |decoder| decoder.read_i32())?
206                .unwrap_or_default(),
207            isr_nodes: decoder
208                .read_compact_array("metadata v12 isr nodes", |decoder| decoder.read_i32())?
209                .unwrap_or_default(),
210            offline_replicas: decoder
211                .read_compact_array("metadata v12 offline replicas", |decoder| {
212                    decoder.read_i32()
213                })?
214                .unwrap_or_default(),
215        };
216        decoder.read_tagged_fields()?;
217        Ok(partition)
218    }
219}
220
221/// Topic metadata returned by Metadata v12, including the stable topic UUID.
222#[derive(Debug, Clone, PartialEq, Eq)]
223pub struct MetadataTopicV12 {
224    pub error_code: i16,
225    pub name: Option<String>,
226    pub topic_id: [u8; 16],
227    pub is_internal: bool,
228    pub partitions: Vec<MetadataPartitionV12>,
229    pub topic_authorized_operations: i32,
230}
231
232impl MetadataTopicV12 {
233    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
234        let topic = Self {
235            error_code: decoder.read_i16()?,
236            name: decoder.read_compact_nullable_string()?,
237            topic_id: decoder.read_uuid()?,
238            is_internal: decoder.read_bool()?,
239            partitions: decoder
240                .read_compact_array("metadata v12 partitions", MetadataPartitionV12::decode)?
241                .unwrap_or_default(),
242            topic_authorized_operations: decoder.read_i32()?,
243        };
244        decoder.read_tagged_fields()?;
245        Ok(topic)
246    }
247}
248
249/// Metadata v12 response with topic UUIDs and flexible encoding.
250#[derive(Debug, Clone, PartialEq, Eq)]
251pub struct MetadataResponseV12 {
252    pub throttle_time_ms: i32,
253    pub brokers: Vec<MetadataBrokerV12>,
254    pub cluster_id: Option<String>,
255    pub controller_id: i32,
256    pub topics: Vec<MetadataTopicV12>,
257}
258
259impl MetadataResponseV12 {
260    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
261        let response = Self {
262            throttle_time_ms: decoder.read_i32()?,
263            brokers: decoder
264                .read_compact_array("metadata v12 brokers", MetadataBrokerV12::decode)?
265                .unwrap_or_default(),
266            cluster_id: decoder.read_compact_nullable_string()?,
267            controller_id: decoder.read_i32()?,
268            topics: decoder
269                .read_compact_array("metadata v12 topics", MetadataTopicV12::decode)?
270                .unwrap_or_default(),
271        };
272        decoder.read_tagged_fields()?;
273        Ok(response)
274    }
275}
276
277#[cfg(test)]
278#[allow(clippy::unwrap_used)]
279mod tests {
280    use super::{
281        MetadataRequestTopicV12, MetadataRequestV1, MetadataRequestV12, MetadataResponseV1, API_KEY,
282    };
283    use crate::codec::{Decoder, Encoder};
284
285    #[test]
286    fn encodes_metadata_request_v1_for_topics() {
287        let request = MetadataRequestV1 {
288            correlation_id: 9,
289            client_id: Some("kafrust".to_owned()),
290            topics: Some(vec!["orders".to_owned()]),
291        };
292
293        assert_eq!(
294            request.encode().unwrap(),
295            [
296                0, 3, // api key
297                0, 1, // api version
298                0, 0, 0, 9, // correlation id
299                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
300                0, 0, 0, 1, // topics count
301                0, 6, b'o', b'r', b'd', b'e', b'r', b's', // topic name
302            ]
303        );
304        assert_eq!(API_KEY, 3);
305    }
306
307    #[test]
308    fn encodes_metadata_request_v1_for_all_topics() {
309        let request = MetadataRequestV1 {
310            correlation_id: 9,
311            client_id: None,
312            topics: None,
313        };
314
315        assert_eq!(
316            request.encode().unwrap(),
317            [
318                0, 3, // api key
319                0, 1, // api version
320                0, 0, 0, 9, // correlation id
321                0xff, 0xff, // null client id
322                0xff, 0xff, 0xff, 0xff, // null topics array
323            ]
324        );
325    }
326
327    #[test]
328    fn encodes_metadata_request_v12_with_topic_name_selector() {
329        let request = MetadataRequestV12 {
330            correlation_id: 11,
331            client_id: Some("kafrust".to_owned()),
332            topics: Some(vec![MetadataRequestTopicV12 {
333                topic_id: [0; 16],
334                name: Some("orders".to_owned()),
335            }]),
336            allow_auto_topic_creation: false,
337            include_topic_authorized_operations: false,
338        };
339
340        let encoded = request.encode().unwrap();
341        assert_eq!(&encoded[0..4], &[0, 3, 0, 12]);
342        assert_eq!(&encoded[4..8], &[0, 0, 0, 11]);
343        assert!(encoded.windows(6).any(|bytes| bytes == b"orders"));
344        assert!(encoded.ends_with(&[0, 0, 0]));
345    }
346
347    #[test]
348    fn decodes_metadata_response_v12_with_topic_uuid() {
349        let mut bytes = Encoder::new();
350        bytes.write_i32(0);
351        bytes
352            .write_compact_array(
353                Some(&[super::MetadataBrokerV12 {
354                    node_id: 1,
355                    host: "localhost".to_owned(),
356                    port: 9092,
357                    rack: None,
358                }]),
359                |encoder, broker| {
360                    encoder.write_i32(broker.node_id);
361                    encoder.write_compact_string(&broker.host)?;
362                    encoder.write_i32(broker.port);
363                    encoder.write_compact_nullable_string(broker.rack.as_deref())?;
364                    encoder.write_empty_tagged_fields();
365                    Ok(())
366                },
367            )
368            .unwrap();
369        bytes
370            .write_compact_nullable_string(Some("cluster"))
371            .unwrap();
372        bytes.write_i32(1);
373        bytes
374            .write_compact_array(
375                Some(&[super::MetadataTopicV12 {
376                    error_code: 0,
377                    name: Some("orders".to_owned()),
378                    topic_id: [7; 16],
379                    is_internal: false,
380                    partitions: vec![super::MetadataPartitionV12 {
381                        error_code: 0,
382                        partition_index: 0,
383                        leader_id: 1,
384                        leader_epoch: 3,
385                        replica_nodes: vec![1],
386                        isr_nodes: vec![1],
387                        offline_replicas: Vec::new(),
388                    }],
389                    topic_authorized_operations: -2147483648,
390                }]),
391                |encoder, topic| {
392                    encoder.write_i16(topic.error_code);
393                    encoder.write_compact_nullable_string(topic.name.as_deref())?;
394                    encoder.write_uuid(&topic.topic_id);
395                    encoder.write_bool(topic.is_internal);
396                    encoder.write_compact_array(
397                        Some(&topic.partitions),
398                        |encoder, partition| {
399                            encoder.write_i16(partition.error_code);
400                            encoder.write_i32(partition.partition_index);
401                            encoder.write_i32(partition.leader_id);
402                            encoder.write_i32(partition.leader_epoch);
403                            encoder.write_compact_array(
404                                Some(&partition.replica_nodes),
405                                |encoder, value| {
406                                    encoder.write_i32(*value);
407                                    Ok(())
408                                },
409                            )?;
410                            encoder.write_compact_array(
411                                Some(&partition.isr_nodes),
412                                |encoder, value| {
413                                    encoder.write_i32(*value);
414                                    Ok(())
415                                },
416                            )?;
417                            encoder.write_compact_array(
418                                Some(&partition.offline_replicas),
419                                |encoder, value| {
420                                    encoder.write_i32(*value);
421                                    Ok(())
422                                },
423                            )?;
424                            encoder.write_empty_tagged_fields();
425                            Ok(())
426                        },
427                    )?;
428                    encoder.write_i32(topic.topic_authorized_operations);
429                    encoder.write_empty_tagged_fields();
430                    Ok(())
431                },
432            )
433            .unwrap();
434        bytes.write_empty_tagged_fields();
435
436        let bytes = bytes.into_bytes();
437        let mut decoder = Decoder::new(&bytes);
438        let response = super::MetadataResponseV12::decode_body(&mut decoder).unwrap();
439
440        assert_eq!(response.cluster_id.as_deref(), Some("cluster"));
441        assert_eq!(response.topics[0].name.as_deref(), Some("orders"));
442        assert_eq!(response.topics[0].topic_id, [7; 16]);
443        assert_eq!(response.topics[0].partitions[0].leader_epoch, 3);
444        assert!(decoder.is_empty());
445    }
446
447    #[test]
448    fn decodes_metadata_response_v1() {
449        let bytes = [
450            0, 0, 0, 1, // brokers count
451            0, 0, 0, 1, // node id
452            0, 9, b'l', b'o', b'c', b'a', b'l', b'h', b'o', b's', b't', // host
453            0, 0, 35, 132, // port 9092
454            0xff, 0xff, // null rack
455            0, 0, 0, 1, // controller id
456            0, 0, 0, 1, // topics count
457            0, 0, // topic error code
458            0, 6, b'o', b'r', b'd', b'e', b'r', b's', // topic name
459            0,    // is internal false
460            0, 0, 0, 1, // partition count
461            0, 0, // partition error code
462            0, 0, 0, 0, // partition index
463            0, 0, 0, 1, // leader id
464            0, 0, 0, 1, // replica count
465            0, 0, 0, 1, // replica node
466            0, 0, 0, 1, // isr count
467            0, 0, 0, 1, // isr node
468        ];
469
470        let mut decoder = Decoder::new(&bytes);
471        let response = MetadataResponseV1::decode_body(&mut decoder).unwrap();
472
473        assert_eq!(response.controller_id, 1);
474        assert_eq!(response.brokers.len(), 1);
475        assert_eq!(response.brokers[0].host, "localhost");
476        assert_eq!(response.brokers[0].port, 9092);
477        assert_eq!(response.topics.len(), 1);
478        assert_eq!(response.topics[0].name, "orders");
479        assert_eq!(response.topics[0].partitions[0].leader_id, 1);
480        assert_eq!(response.topics[0].partitions[0].replica_nodes, vec![1]);
481        assert_eq!(response.topics[0].partitions[0].isr_nodes, vec![1]);
482        assert!(decoder.is_empty());
483    }
484}
kafrust_protocol/api/metadata.rs

kafrust_protocol/api/
metadata.rs
