kafrust_protocol/api/
delete_records.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 21;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct DeleteRecordsRequestV1 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub topics: Vec<DeleteRecordsTopicV1>,
12    pub timeout_ms: i32,
13}
14
15impl DeleteRecordsRequestV1 {
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.topics), |encoder, topic| topic.encode(encoder))?;
26        encoder.write_i32(self.timeout_ms);
27        Ok(encoder.into_bytes())
28    }
29}
30
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct DeleteRecordsTopicV1 {
33    pub name: String,
34    pub partitions: Vec<DeleteRecordsPartitionV1>,
35}
36
37impl DeleteRecordsTopicV1 {
38    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
39        encoder.write_string(&self.name)?;
40        encoder.write_array(Some(&self.partitions), |encoder, partition| {
41            partition.encode(encoder)
42        })
43    }
44}
45
46#[derive(Debug, Clone, PartialEq, Eq)]
47pub struct DeleteRecordsPartitionV1 {
48    pub partition_index: i32,
49    pub offset: i64,
50}
51
52impl DeleteRecordsPartitionV1 {
53    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
54        encoder.write_i32(self.partition_index);
55        encoder.write_i64(self.offset);
56        Ok(())
57    }
58}
59
60#[derive(Debug, Clone, PartialEq, Eq)]
61pub struct DeleteRecordsResponseV1 {
62    pub throttle_time_ms: i32,
63    pub topics: Vec<DeleteRecordsTopicResponseV1>,
64}
65
66impl DeleteRecordsResponseV1 {
67    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
68        Ok(Self {
69            throttle_time_ms: decoder.read_i32()?,
70            topics: decoder
71                .read_array(
72                    "delete records topics",
73                    DeleteRecordsTopicResponseV1::decode,
74                )?
75                .unwrap_or_default(),
76        })
77    }
78}
79
80#[derive(Debug, Clone, PartialEq, Eq)]
81pub struct DeleteRecordsTopicResponseV1 {
82    pub name: String,
83    pub partitions: Vec<DeleteRecordsPartitionResponseV1>,
84}
85
86impl DeleteRecordsTopicResponseV1 {
87    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
88        Ok(Self {
89            name: decoder.read_string()?,
90            partitions: decoder
91                .read_array(
92                    "delete records partition results",
93                    DeleteRecordsPartitionResponseV1::decode,
94                )?
95                .unwrap_or_default(),
96        })
97    }
98}
99
100#[derive(Debug, Clone, PartialEq, Eq)]
101pub struct DeleteRecordsPartitionResponseV1 {
102    pub partition_index: i32,
103    pub low_watermark: i64,
104    pub error_code: i16,
105}
106
107impl DeleteRecordsPartitionResponseV1 {
108    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
109        Ok(Self {
110            partition_index: decoder.read_i32()?,
111            low_watermark: decoder.read_i64()?,
112            error_code: decoder.read_i16()?,
113        })
114    }
115}
116
117#[cfg(test)]
118#[allow(clippy::unwrap_used)]
119mod tests {
120    use super::*;
121    use crate::codec::Decoder;
122
123    #[test]
124    fn encodes_delete_records_v1_request() {
125        let request = DeleteRecordsRequestV1 {
126            correlation_id: 11,
127            client_id: Some("kafrust".to_owned()),
128            topics: vec![
129                DeleteRecordsTopicV1 {
130                    name: "orders".to_owned(),
131                    partitions: vec![
132                        DeleteRecordsPartitionV1 {
133                            partition_index: 0,
134                            offset: 100,
135                        },
136                        DeleteRecordsPartitionV1 {
137                            partition_index: 1,
138                            offset: -1,
139                        },
140                    ],
141                },
142                DeleteRecordsTopicV1 {
143                    name: "payments".to_owned(),
144                    partitions: vec![DeleteRecordsPartitionV1 {
145                        partition_index: 2,
146                        offset: 40,
147                    }],
148                },
149            ],
150            timeout_ms: 30_000,
151        };
152
153        assert_eq!(
154            request.encode().unwrap(),
155            [
156                0, 21, // API key
157                0, 1, // API version
158                0, 0, 0, 11, // correlation ID
159                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client ID
160                0, 0, 0, 2, // topic count
161                0, 6, b'o', b'r', b'd', b'e', b'r', b's', // topic
162                0, 0, 0, 2, // partition count
163                0, 0, 0, 0, // partition 0
164                0, 0, 0, 0, 0, 0, 0, 100, // offset 100
165                0, 0, 0, 1, // partition 1
166                0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, // high watermark
167                0, 8, b'p', b'a', b'y', b'm', b'e', b'n', b't', b's', // topic
168                0, 0, 0, 1, // partition count
169                0, 0, 0, 2, // partition 2
170                0, 0, 0, 0, 0, 0, 0, 40, // offset 40
171                0, 0, 117, 48, // timeout
172            ]
173        );
174        assert_eq!(API_KEY, 21);
175    }
176
177    #[test]
178    fn decodes_delete_records_v1_response() {
179        let bytes = [
180            0, 0, 0, 8, // 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, 0, 2, // partition count
184            0, 0, 0, 0, // partition 0
185            0, 0, 0, 0, 0, 0, 0, 100, // low watermark
186            0, 0, // success
187            0, 0, 0, 1, // partition 1
188            0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, 0xff, // low watermark
189            0, 3, // unknown topic or partition
190            0, 8, b'p', b'a', b'y', b'm', b'e', b'n', b't', b's', // topic
191            0, 0, 0, 1, // partition count
192            0, 0, 0, 2, // partition 2
193            0, 0, 0, 0, 0, 0, 0, 40, // low watermark
194            0, 0, // success
195        ];
196        let mut decoder = Decoder::new(&bytes);
197
198        let response = DeleteRecordsResponseV1::decode_body(&mut decoder).unwrap();
199
200        assert_eq!(response.throttle_time_ms, 8);
201        assert_eq!(response.topics.len(), 2);
202        assert_eq!(response.topics[0].name, "orders");
203        assert_eq!(response.topics[0].partitions[0].low_watermark, 100);
204        assert_eq!(response.topics[0].partitions[1].error_code, 3);
205        assert_eq!(response.topics[1].partitions[0].partition_index, 2);
206        assert!(decoder.is_empty());
207    }
208}
kafrust_protocol/api/delete_records.rs

kafrust_protocol/api/
delete_records.rs
