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

kafrust_protocol/api/
metadata.rs
