kafrust_protocol/api/
offset_delete.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 47;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct OffsetDeleteRequestV0 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub group_id: String,
12    pub topics: Vec<OffsetDeleteRequestTopicV0>,
13}
14
15impl OffsetDeleteRequestV0 {
16    pub fn encode(&self) -> Result<Vec<u8>> {
17        let mut encoder = Encoder::new();
18        RequestHeader {
19            api_key: API_KEY,
20            api_version: 0,
21            correlation_id: self.correlation_id,
22            client_id: self.client_id.clone(),
23        }
24        .encode_v1(&mut encoder)?;
25        encoder.write_string(&self.group_id)?;
26        encoder.write_array(Some(&self.topics), |encoder, topic| {
27            encoder.write_string(&topic.name)?;
28            encoder.write_array(Some(&topic.partitions), |encoder, partition| {
29                encoder.write_i32(partition.partition_index);
30                Ok(())
31            })
32        })?;
33        Ok(encoder.into_bytes())
34    }
35}
36
37#[derive(Debug, Clone, PartialEq, Eq)]
38pub struct OffsetDeleteRequestTopicV0 {
39    pub name: String,
40    pub partitions: Vec<OffsetDeleteRequestPartitionV0>,
41}
42
43#[derive(Debug, Clone, Copy, PartialEq, Eq)]
44pub struct OffsetDeleteRequestPartitionV0 {
45    pub partition_index: i32,
46}
47
48#[derive(Debug, Clone, PartialEq, Eq)]
49pub struct OffsetDeleteResponseV0 {
50    pub error_code: i16,
51    pub throttle_time_ms: i32,
52    pub topics: Vec<OffsetDeleteResponseTopicV0>,
53}
54
55impl OffsetDeleteResponseV0 {
56    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
57        Ok(Self {
58            error_code: decoder.read_i16()?,
59            throttle_time_ms: decoder.read_i32()?,
60            topics: decoder
61                .read_array("offset delete topics", OffsetDeleteResponseTopicV0::decode)?
62                .unwrap_or_default(),
63        })
64    }
65}
66
67#[derive(Debug, Clone, PartialEq, Eq)]
68pub struct OffsetDeleteResponseTopicV0 {
69    pub name: String,
70    pub partitions: Vec<OffsetDeleteResponsePartitionV0>,
71}
72
73impl OffsetDeleteResponseTopicV0 {
74    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
75        Ok(Self {
76            name: decoder.read_string()?,
77            partitions: decoder
78                .read_array(
79                    "offset delete partitions",
80                    OffsetDeleteResponsePartitionV0::decode,
81                )?
82                .unwrap_or_default(),
83        })
84    }
85}
86
87#[derive(Debug, Clone, Copy, PartialEq, Eq)]
88pub struct OffsetDeleteResponsePartitionV0 {
89    pub partition_index: i32,
90    pub error_code: i16,
91}
92
93impl OffsetDeleteResponsePartitionV0 {
94    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
95        Ok(Self {
96            partition_index: decoder.read_i32()?,
97            error_code: decoder.read_i16()?,
98        })
99    }
100}
101
102#[cfg(test)]
103#[allow(clippy::unwrap_used)]
104mod tests {
105    use super::{
106        OffsetDeleteRequestPartitionV0, OffsetDeleteRequestTopicV0, OffsetDeleteRequestV0,
107        OffsetDeleteResponseV0, API_KEY,
108    };
109    use crate::codec::Decoder;
110
111    #[test]
112    fn encodes_offset_delete_v0_request() {
113        let request = OffsetDeleteRequestV0 {
114            correlation_id: 12,
115            client_id: Some("kafrust".to_owned()),
116            group_id: "orders-group".to_owned(),
117            topics: vec![OffsetDeleteRequestTopicV0 {
118                name: "orders".to_owned(),
119                partitions: vec![
120                    OffsetDeleteRequestPartitionV0 { partition_index: 0 },
121                    OffsetDeleteRequestPartitionV0 { partition_index: 2 },
122                ],
123            }],
124        };
125
126        assert_eq!(
127            request.encode().unwrap(),
128            [
129                0, 47, // API key
130                0, 0, // API version
131                0, 0, 0, 12, // correlation ID
132                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client ID
133                0, 12, b'o', b'r', b'd', b'e', b'r', b's', b'-', b'g', b'r', b'o', b'u', b'p', 0,
134                0, 0, 1, // topic count
135                0, 6, b'o', b'r', b'd', b'e', b'r', b's', // topic name
136                0, 0, 0, 2, // partition count
137                0, 0, 0, 0, // partition 0
138                0, 0, 0, 2, // partition 2
139            ]
140        );
141        assert_eq!(API_KEY, 47);
142    }
143
144    #[test]
145    fn decodes_offset_delete_v0_response_in_schema_order() {
146        let bytes = [
147            0, 69, // group ID not found
148            0, 0, 0, 5, // throttle time
149            0, 0, 0, 1, // topic count
150            0, 6, b'o', b'r', b'd', b'e', b'r', b's', // topic name
151            0, 0, 0, 2, // partition count
152            0, 0, 0, 0, // partition 0
153            0, 0, // success
154            0, 0, 0, 2, // partition 2
155            0, 86, // group subscribed to topic
156        ];
157        let mut decoder = Decoder::new(&bytes);
158
159        let response = OffsetDeleteResponseV0::decode_body(&mut decoder).unwrap();
160
161        assert_eq!(response.error_code, 69);
162        assert_eq!(response.throttle_time_ms, 5);
163        assert_eq!(response.topics.len(), 1);
164        assert_eq!(response.topics[0].name, "orders");
165        assert_eq!(response.topics[0].partitions.len(), 2);
166        assert_eq!(response.topics[0].partitions[0].partition_index, 0);
167        assert_eq!(response.topics[0].partitions[0].error_code, 0);
168        assert_eq!(response.topics[0].partitions[1].partition_index, 2);
169        assert_eq!(response.topics[0].partitions[1].error_code, 86);
170        assert!(decoder.is_empty());
171    }
172}
kafrust_protocol/api/offset_delete.rs

kafrust_protocol/api/
offset_delete.rs
