kafrust_protocol/api/
offset_commit.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 8;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct OffsetCommitRequestV2 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub group_id: String,
12    pub generation_id_or_member_epoch: i32,
13    pub member_id: String,
14    pub retention_time_ms: i64,
15    pub topics: Vec<OffsetCommitTopic>,
16}
17
18impl OffsetCommitRequestV2 {
19    pub fn encode(&self) -> Result<Vec<u8>> {
20        let mut encoder = Encoder::new();
21        RequestHeader {
22            api_key: API_KEY,
23            api_version: 2,
24            correlation_id: self.correlation_id,
25            client_id: self.client_id.clone(),
26        }
27        .encode_v1(&mut encoder)?;
28        encoder.write_string(&self.group_id)?;
29        encoder.write_i32(self.generation_id_or_member_epoch);
30        encoder.write_string(&self.member_id)?;
31        encoder.write_i64(self.retention_time_ms);
32        encoder.write_array(Some(self.topics.as_slice()), |encoder, topic| {
33            topic.encode(encoder)
34        })?;
35        Ok(encoder.into_bytes())
36    }
37}
38
39#[derive(Debug, Clone, PartialEq, Eq)]
40pub struct OffsetCommitRequestV7 {
41    pub correlation_id: i32,
42    pub client_id: Option<String>,
43    pub group_id: String,
44    pub generation_id_or_member_epoch: i32,
45    pub member_id: String,
46    pub group_instance_id: Option<String>,
47    pub topics: Vec<OffsetCommitTopicV7>,
48}
49
50impl OffsetCommitRequestV7 {
51    pub fn encode(&self) -> Result<Vec<u8>> {
52        let mut encoder = Encoder::new();
53        RequestHeader {
54            api_key: API_KEY,
55            api_version: 7,
56            correlation_id: self.correlation_id,
57            client_id: self.client_id.clone(),
58        }
59        .encode_v1(&mut encoder)?;
60        encoder.write_string(&self.group_id)?;
61        encoder.write_i32(self.generation_id_or_member_epoch);
62        encoder.write_string(&self.member_id)?;
63        encoder.write_nullable_string(self.group_instance_id.as_deref())?;
64        encoder.write_array(Some(self.topics.as_slice()), |encoder, topic| {
65            topic.encode(encoder)
66        })?;
67        Ok(encoder.into_bytes())
68    }
69}
70
71#[derive(Debug, Clone, PartialEq, Eq)]
72pub struct OffsetCommitTopic {
73    pub name: String,
74    pub partitions: Vec<OffsetCommitPartition>,
75}
76
77impl OffsetCommitTopic {
78    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
79        encoder.write_string(&self.name)?;
80        encoder.write_array(Some(self.partitions.as_slice()), |encoder, partition| {
81            partition.encode(encoder)
82        })
83    }
84}
85
86#[derive(Debug, Clone, PartialEq, Eq)]
87pub struct OffsetCommitTopicV7 {
88    pub name: String,
89    pub partitions: Vec<OffsetCommitPartitionV7>,
90}
91
92impl OffsetCommitTopicV7 {
93    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
94        encoder.write_string(&self.name)?;
95        encoder.write_array(Some(self.partitions.as_slice()), |encoder, partition| {
96            partition.encode(encoder)
97        })
98    }
99}
100
101#[derive(Debug, Clone, PartialEq, Eq)]
102pub struct OffsetCommitPartition {
103    pub partition_index: i32,
104    pub committed_offset: i64,
105    pub committed_metadata: Option<String>,
106}
107
108impl OffsetCommitPartition {
109    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
110        encoder.write_i32(self.partition_index);
111        encoder.write_i64(self.committed_offset);
112        encoder.write_nullable_string(self.committed_metadata.as_deref())
113    }
114}
115
116#[derive(Debug, Clone, PartialEq, Eq)]
117pub struct OffsetCommitPartitionV7 {
118    pub partition_index: i32,
119    pub committed_offset: i64,
120    pub committed_leader_epoch: i32,
121    pub committed_metadata: Option<String>,
122}
123
124impl OffsetCommitPartitionV7 {
125    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
126        encoder.write_i32(self.partition_index);
127        encoder.write_i64(self.committed_offset);
128        encoder.write_i32(self.committed_leader_epoch);
129        encoder.write_nullable_string(self.committed_metadata.as_deref())
130    }
131}
132
133#[derive(Debug, Clone, PartialEq, Eq)]
134pub struct OffsetCommitResponseV2 {
135    pub topics: Vec<OffsetCommitTopicResponse>,
136}
137
138impl OffsetCommitResponseV2 {
139    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
140        Ok(Self {
141            topics: decoder
142                .read_array(
143                    "offset commit topic responses",
144                    OffsetCommitTopicResponse::decode,
145                )?
146                .unwrap_or_default(),
147        })
148    }
149}
150
151#[derive(Debug, Clone, PartialEq, Eq)]
152pub struct OffsetCommitResponseV7 {
153    pub throttle_time_ms: i32,
154    pub topics: Vec<OffsetCommitTopicResponse>,
155}
156
157impl OffsetCommitResponseV7 {
158    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
159        Ok(Self {
160            throttle_time_ms: decoder.read_i32()?,
161            topics: decoder
162                .read_array(
163                    "offset commit topic responses",
164                    OffsetCommitTopicResponse::decode,
165                )?
166                .unwrap_or_default(),
167        })
168    }
169}
170
171#[derive(Debug, Clone, PartialEq, Eq)]
172pub struct OffsetCommitTopicResponse {
173    pub name: String,
174    pub partitions: Vec<OffsetCommitPartitionResponse>,
175}
176
177impl OffsetCommitTopicResponse {
178    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
179        Ok(Self {
180            name: decoder.read_string()?,
181            partitions: decoder
182                .read_array(
183                    "offset commit partition responses",
184                    OffsetCommitPartitionResponse::decode,
185                )?
186                .unwrap_or_default(),
187        })
188    }
189}
190
191#[derive(Debug, Clone, PartialEq, Eq)]
192pub struct OffsetCommitPartitionResponse {
193    pub partition_index: i32,
194    pub error_code: i16,
195}
196
197impl OffsetCommitPartitionResponse {
198    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
199        Ok(Self {
200            partition_index: decoder.read_i32()?,
201            error_code: decoder.read_i16()?,
202        })
203    }
204}
205
206#[cfg(test)]
207#[allow(clippy::unwrap_used)]
208mod tests {
209    use super::{
210        OffsetCommitPartition, OffsetCommitPartitionResponse, OffsetCommitPartitionV7,
211        OffsetCommitRequestV2, OffsetCommitRequestV7, OffsetCommitResponseV2,
212        OffsetCommitResponseV7, OffsetCommitTopic, OffsetCommitTopicResponse, OffsetCommitTopicV7,
213    };
214    use crate::codec::{Decoder, Encoder};
215
216    #[test]
217    fn encodes_offset_commit_v2_request() {
218        let request = OffsetCommitRequestV2 {
219            correlation_id: 23,
220            client_id: Some("kafrust".to_owned()),
221            group_id: "orders-group".to_owned(),
222            generation_id_or_member_epoch: 7,
223            member_id: "member-a".to_owned(),
224            retention_time_ms: 86_400_000,
225            topics: vec![OffsetCommitTopic {
226                name: "orders".to_owned(),
227                partitions: vec![OffsetCommitPartition {
228                    partition_index: 0,
229                    committed_offset: 42,
230                    committed_metadata: Some("processed".to_owned()),
231                }],
232            }],
233        };
234
235        assert_eq!(
236            request.encode().unwrap(),
237            [
238                0, 8, // api key
239                0, 2, // api version
240                0, 0, 0, 23, // correlation id
241                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
242                0, 12, b'o', b'r', b'd', b'e', b'r', b's', b'-', b'g', b'r', b'o', b'u',
243                b'p', // group id
244                0, 0, 0, 7, // generation id
245                0, 8, b'm', b'e', b'm', b'b', b'e', b'r', b'-', b'a', // member id
246                0, 0, 0, 0, 5, 38, 92, 0, // retention time
247                0, 0, 0, 1, // topic count
248                0, 6, b'o', b'r', b'd', b'e', b'r', b's', // topic
249                0, 0, 0, 1, // partition count
250                0, 0, 0, 0, // partition
251                0, 0, 0, 0, 0, 0, 0, 42, // committed offset
252                0, 9, b'p', b'r', b'o', b'c', b'e', b's', b's', b'e', b'd', // metadata
253            ]
254        );
255    }
256
257    #[test]
258    fn decodes_offset_commit_v2_response() {
259        let mut bytes = Encoder::new();
260        bytes.write_i32(1);
261        bytes.write_string("orders").unwrap();
262        bytes.write_i32(1);
263        bytes.write_i32(0);
264        bytes.write_i16(0);
265        let bytes = bytes.into_bytes();
266
267        let mut decoder = Decoder::new(&bytes);
268        let response = OffsetCommitResponseV2::decode_body(&mut decoder).unwrap();
269
270        assert_eq!(
271            response.topics,
272            vec![OffsetCommitTopicResponse {
273                name: "orders".to_owned(),
274                partitions: vec![OffsetCommitPartitionResponse {
275                    partition_index: 0,
276                    error_code: 0,
277                }],
278            }]
279        );
280        assert!(decoder.is_empty());
281    }
282
283    #[test]
284    fn encodes_offset_commit_v7_request_with_static_member() {
285        let request = OffsetCommitRequestV7 {
286            correlation_id: 23,
287            client_id: Some("kafrust".to_owned()),
288            group_id: "orders-group".to_owned(),
289            generation_id_or_member_epoch: 7,
290            member_id: "member-a".to_owned(),
291            group_instance_id: Some("orders-reader-1".to_owned()),
292            topics: vec![OffsetCommitTopicV7 {
293                name: "orders".to_owned(),
294                partitions: vec![OffsetCommitPartitionV7 {
295                    partition_index: 0,
296                    committed_offset: 42,
297                    committed_leader_epoch: -1,
298                    committed_metadata: None,
299                }],
300            }],
301        };
302
303        let encoded = request.encode().unwrap();
304        assert_eq!(&encoded[0..4], &[0, 8, 0, 7]);
305        assert!(encoded
306            .windows(17)
307            .any(|bytes| bytes == b"\0\x0forders-reader-1"));
308        assert!(encoded.windows(4).any(|bytes| bytes == [u8::MAX; 4]));
309    }
310
311    #[test]
312    fn decodes_offset_commit_v7_response() {
313        let mut bytes = Encoder::new();
314        bytes.write_i32(12);
315        bytes.write_i32(1);
316        bytes.write_string("orders").unwrap();
317        bytes.write_i32(1);
318        bytes.write_i32(0);
319        bytes.write_i16(0);
320        let bytes = bytes.into_bytes();
321
322        let mut decoder = Decoder::new(&bytes);
323        let response = OffsetCommitResponseV7::decode_body(&mut decoder).unwrap();
324
325        assert_eq!(response.throttle_time_ms, 12);
326        assert_eq!(response.topics[0].partitions[0].error_code, 0);
327        assert!(decoder.is_empty());
328    }
329}
kafrust_protocol/api/offset_commit.rs

kafrust_protocol/api/
offset_commit.rs
