kafrust_protocol/api/
create_topics.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 19;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct CreateTopicsRequestV2 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub topics: Vec<CreateTopicsTopicV2>,
12    pub timeout_ms: i32,
13    pub validate_only: bool,
14}
15
16impl CreateTopicsRequestV2 {
17    pub fn encode(&self) -> Result<Vec<u8>> {
18        let mut encoder = Encoder::new();
19        RequestHeader {
20            api_key: API_KEY,
21            api_version: 2,
22            correlation_id: self.correlation_id,
23            client_id: self.client_id.clone(),
24        }
25        .encode_v1(&mut encoder)?;
26        encoder.write_array(Some(&self.topics), |encoder, topic| topic.encode(encoder))?;
27        encoder.write_i32(self.timeout_ms);
28        encoder.write_bool(self.validate_only);
29        Ok(encoder.into_bytes())
30    }
31}
32
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub struct CreateTopicsTopicV2 {
35    pub name: String,
36    pub num_partitions: i32,
37    pub replication_factor: i16,
38    pub assignments: Vec<CreateTopicsAssignmentV2>,
39    pub configs: Vec<CreateTopicsConfigV2>,
40}
41
42impl CreateTopicsTopicV2 {
43    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
44        encoder.write_string(&self.name)?;
45        encoder.write_i32(self.num_partitions);
46        encoder.write_i16(self.replication_factor);
47        encoder.write_array(Some(&self.assignments), |encoder, assignment| {
48            assignment.encode(encoder)
49        })?;
50        encoder.write_array(Some(&self.configs), |encoder, config| {
51            config.encode(encoder)
52        })
53    }
54}
55
56#[derive(Debug, Clone, PartialEq, Eq)]
57pub struct CreateTopicsAssignmentV2 {
58    pub partition_index: i32,
59    pub broker_ids: Vec<i32>,
60}
61
62impl CreateTopicsAssignmentV2 {
63    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
64        encoder.write_i32(self.partition_index);
65        encoder.write_array(Some(&self.broker_ids), |encoder, broker_id| {
66            encoder.write_i32(*broker_id);
67            Ok(())
68        })
69    }
70}
71
72#[derive(Debug, Clone, PartialEq, Eq)]
73pub struct CreateTopicsConfigV2 {
74    pub name: String,
75    pub value: Option<String>,
76}
77
78impl CreateTopicsConfigV2 {
79    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
80        encoder.write_string(&self.name)?;
81        encoder.write_nullable_string(self.value.as_deref())
82    }
83}
84
85#[derive(Debug, Clone, PartialEq, Eq)]
86pub struct CreateTopicsResponseV2 {
87    pub throttle_time_ms: i32,
88    pub topics: Vec<CreateTopicsTopicResultV2>,
89}
90
91impl CreateTopicsResponseV2 {
92    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
93        Ok(Self {
94            throttle_time_ms: decoder.read_i32()?,
95            topics: decoder
96                .read_array("create topics results", CreateTopicsTopicResultV2::decode)?
97                .unwrap_or_default(),
98        })
99    }
100}
101
102#[derive(Debug, Clone, PartialEq, Eq)]
103pub struct CreateTopicsTopicResultV2 {
104    pub name: String,
105    pub error_code: i16,
106    pub error_message: Option<String>,
107}
108
109impl CreateTopicsTopicResultV2 {
110    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
111        Ok(Self {
112            name: decoder.read_string()?,
113            error_code: decoder.read_i16()?,
114            error_message: decoder.read_nullable_string()?,
115        })
116    }
117}
118
119#[cfg(test)]
120#[allow(clippy::unwrap_used)]
121mod tests {
122    use super::{
123        CreateTopicsAssignmentV2, CreateTopicsConfigV2, CreateTopicsRequestV2,
124        CreateTopicsResponseV2, CreateTopicsTopicV2, API_KEY,
125    };
126    use crate::codec::Decoder;
127
128    #[test]
129    fn encodes_create_topics_v2_request() {
130        let request = CreateTopicsRequestV2 {
131            correlation_id: 9,
132            client_id: Some("kafrust".to_owned()),
133            topics: vec![CreateTopicsTopicV2 {
134                name: "orders".to_owned(),
135                num_partitions: -1,
136                replication_factor: -1,
137                assignments: vec![CreateTopicsAssignmentV2 {
138                    partition_index: 0,
139                    broker_ids: vec![1, 2],
140                }],
141                configs: vec![CreateTopicsConfigV2 {
142                    name: "cleanup.policy".to_owned(),
143                    value: Some("compact".to_owned()),
144                }],
145            }],
146            timeout_ms: 30_000,
147            validate_only: true,
148        };
149
150        assert_eq!(
151            request.encode().unwrap(),
152            [
153                0, 19, // API key
154                0, 2, // API version
155                0, 0, 0, 9, // correlation ID
156                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client ID
157                0, 0, 0, 1, // topic count
158                0, 6, b'o', b'r', b'd', b'e', b'r', b's', // topic name
159                0xff, 0xff, 0xff, 0xff, // default partition count
160                0xff, 0xff, // default replication factor
161                0, 0, 0, 1, // assignment count
162                0, 0, 0, 0, // partition index
163                0, 0, 0, 2, // broker count
164                0, 0, 0, 1, // broker 1
165                0, 0, 0, 2, // broker 2
166                0, 0, 0, 1, // config count
167                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',
168                b'c', b'y', // config name
169                0, 7, b'c', b'o', b'm', b'p', b'a', b'c', b't', // config value
170                0, 0, 117, 48, // timeout
171                1,  // validate only
172            ]
173        );
174        assert_eq!(API_KEY, 19);
175    }
176
177    #[test]
178    fn decodes_create_topics_v2_response() {
179        let bytes = [
180            0, 0, 0, 12, // throttle time
181            0, 0, 0, 2, // topic count
182            0, 6, b'o', b'r', b'd', b'e', b'r', b's', // topic
183            0, 0, // success
184            0xff, 0xff, // null error message
185            0, 8, b'p', b'a', b'y', b'm', b'e', b'n', b't', b's', // topic
186            0, 36, // topic already exists
187            0, 6, b'e', b'x', b'i', b's', b't', b's', // error message
188        ];
189        let mut decoder = Decoder::new(&bytes);
190
191        let response = CreateTopicsResponseV2::decode_body(&mut decoder).unwrap();
192
193        assert_eq!(response.throttle_time_ms, 12);
194        assert_eq!(response.topics.len(), 2);
195        assert_eq!(response.topics[0].name, "orders");
196        assert_eq!(response.topics[0].error_code, 0);
197        assert_eq!(response.topics[0].error_message, None);
198        assert_eq!(response.topics[1].name, "payments");
199        assert_eq!(response.topics[1].error_code, 36);
200        assert_eq!(response.topics[1].error_message.as_deref(), Some("exists"));
201        assert!(decoder.is_empty());
202    }
203}
kafrust_protocol/api/create_topics.rs

kafrust_protocol/api/
create_topics.rs
