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/// OffsetCommit v9 for Kafka's KIP-848 consumer group protocol.
72///
73/// Version 9 has the same logical fields as v8, but is the first version
74/// accepted for consumer-protocol group members. Its body uses flexible
75/// encodings and tagged fields throughout.
76#[derive(Debug, Clone, PartialEq, Eq)]
77pub struct OffsetCommitRequestV9 {
78    pub correlation_id: i32,
79    pub client_id: Option<String>,
80    pub group_id: String,
81    pub generation_id_or_member_epoch: i32,
82    pub member_id: String,
83    pub group_instance_id: Option<String>,
84    pub topics: Vec<OffsetCommitTopicV9>,
85}
86
87impl OffsetCommitRequestV9 {
88    pub fn encode(&self) -> Result<Vec<u8>> {
89        let mut encoder = Encoder::new();
90        RequestHeader {
91            api_key: API_KEY,
92            api_version: 9,
93            correlation_id: self.correlation_id,
94            client_id: self.client_id.clone(),
95        }
96        .encode_v2(&mut encoder)?;
97        encoder.write_compact_string(&self.group_id)?;
98        encoder.write_i32(self.generation_id_or_member_epoch);
99        encoder.write_compact_string(&self.member_id)?;
100        encoder.write_compact_nullable_string(self.group_instance_id.as_deref())?;
101        encoder.write_compact_array(Some(self.topics.as_slice()), |encoder, topic| {
102            topic.encode(encoder)
103        })?;
104        encoder.write_empty_tagged_fields();
105        Ok(encoder.into_bytes())
106    }
107}
108
109#[derive(Debug, Clone, PartialEq, Eq)]
110pub struct OffsetCommitTopic {
111    pub name: String,
112    pub partitions: Vec<OffsetCommitPartition>,
113}
114
115impl OffsetCommitTopic {
116    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
117        encoder.write_string(&self.name)?;
118        encoder.write_array(Some(self.partitions.as_slice()), |encoder, partition| {
119            partition.encode(encoder)
120        })
121    }
122}
123
124#[derive(Debug, Clone, PartialEq, Eq)]
125pub struct OffsetCommitTopicV7 {
126    pub name: String,
127    pub partitions: Vec<OffsetCommitPartitionV7>,
128}
129
130impl OffsetCommitTopicV7 {
131    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
132        encoder.write_string(&self.name)?;
133        encoder.write_array(Some(self.partitions.as_slice()), |encoder, partition| {
134            partition.encode(encoder)
135        })
136    }
137}
138
139#[derive(Debug, Clone, PartialEq, Eq)]
140pub struct OffsetCommitTopicV9 {
141    pub name: String,
142    pub partitions: Vec<OffsetCommitPartitionV9>,
143}
144
145impl OffsetCommitTopicV9 {
146    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
147        encoder.write_compact_string(&self.name)?;
148        encoder.write_compact_array(Some(self.partitions.as_slice()), |encoder, partition| {
149            partition.encode(encoder)
150        })?;
151        encoder.write_empty_tagged_fields();
152        Ok(())
153    }
154}
155
156#[derive(Debug, Clone, PartialEq, Eq)]
157pub struct OffsetCommitPartition {
158    pub partition_index: i32,
159    pub committed_offset: i64,
160    pub committed_metadata: Option<String>,
161}
162
163impl OffsetCommitPartition {
164    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
165        encoder.write_i32(self.partition_index);
166        encoder.write_i64(self.committed_offset);
167        encoder.write_nullable_string(self.committed_metadata.as_deref())
168    }
169}
170
171#[derive(Debug, Clone, PartialEq, Eq)]
172pub struct OffsetCommitPartitionV7 {
173    pub partition_index: i32,
174    pub committed_offset: i64,
175    pub committed_leader_epoch: i32,
176    pub committed_metadata: Option<String>,
177}
178
179impl OffsetCommitPartitionV7 {
180    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
181        encoder.write_i32(self.partition_index);
182        encoder.write_i64(self.committed_offset);
183        encoder.write_i32(self.committed_leader_epoch);
184        encoder.write_nullable_string(self.committed_metadata.as_deref())
185    }
186}
187
188#[derive(Debug, Clone, PartialEq, Eq)]
189pub struct OffsetCommitPartitionV9 {
190    pub partition_index: i32,
191    pub committed_offset: i64,
192    pub committed_leader_epoch: i32,
193    pub committed_metadata: Option<String>,
194}
195
196impl OffsetCommitPartitionV9 {
197    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
198        encoder.write_i32(self.partition_index);
199        encoder.write_i64(self.committed_offset);
200        encoder.write_i32(self.committed_leader_epoch);
201        encoder.write_compact_nullable_string(self.committed_metadata.as_deref())?;
202        encoder.write_empty_tagged_fields();
203        Ok(())
204    }
205}
206
207#[derive(Debug, Clone, PartialEq, Eq)]
208pub struct OffsetCommitResponseV2 {
209    pub topics: Vec<OffsetCommitTopicResponse>,
210}
211
212impl OffsetCommitResponseV2 {
213    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
214        Ok(Self {
215            topics: decoder
216                .read_array(
217                    "offset commit topic responses",
218                    OffsetCommitTopicResponse::decode,
219                )?
220                .unwrap_or_default(),
221        })
222    }
223}
224
225#[derive(Debug, Clone, PartialEq, Eq)]
226pub struct OffsetCommitResponseV7 {
227    pub throttle_time_ms: i32,
228    pub topics: Vec<OffsetCommitTopicResponse>,
229}
230
231impl OffsetCommitResponseV7 {
232    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
233        Ok(Self {
234            throttle_time_ms: decoder.read_i32()?,
235            topics: decoder
236                .read_array(
237                    "offset commit topic responses",
238                    OffsetCommitTopicResponse::decode,
239                )?
240                .unwrap_or_default(),
241        })
242    }
243}
244
245#[derive(Debug, Clone, PartialEq, Eq)]
246pub struct OffsetCommitResponseV9 {
247    pub throttle_time_ms: i32,
248    pub topics: Vec<OffsetCommitTopicResponse>,
249}
250
251impl OffsetCommitResponseV9 {
252    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
253        let throttle_time_ms = decoder.read_i32()?;
254        let topics = decoder
255            .read_compact_array("offset commit topic responses", |decoder| {
256                let name = decoder.read_compact_string()?;
257                let partitions = decoder
258                    .read_compact_array("offset commit partition responses", |decoder| {
259                        let partition_index = decoder.read_i32()?;
260                        let error_code = decoder.read_i16()?;
261                        decoder.read_tagged_fields()?;
262                        Ok(OffsetCommitPartitionResponse {
263                            partition_index,
264                            error_code,
265                        })
266                    })?
267                    .unwrap_or_default();
268                decoder.read_tagged_fields()?;
269                Ok(OffsetCommitTopicResponse { name, partitions })
270            })?
271            .unwrap_or_default();
272        decoder.read_tagged_fields()?;
273        Ok(Self {
274            throttle_time_ms,
275            topics,
276        })
277    }
278}
279
280#[derive(Debug, Clone, PartialEq, Eq)]
281pub struct OffsetCommitTopicResponse {
282    pub name: String,
283    pub partitions: Vec<OffsetCommitPartitionResponse>,
284}
285
286impl OffsetCommitTopicResponse {
287    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
288        Ok(Self {
289            name: decoder.read_string()?,
290            partitions: decoder
291                .read_array(
292                    "offset commit partition responses",
293                    OffsetCommitPartitionResponse::decode,
294                )?
295                .unwrap_or_default(),
296        })
297    }
298}
299
300#[derive(Debug, Clone, PartialEq, Eq)]
301pub struct OffsetCommitPartitionResponse {
302    pub partition_index: i32,
303    pub error_code: i16,
304}
305
306impl OffsetCommitPartitionResponse {
307    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
308        Ok(Self {
309            partition_index: decoder.read_i32()?,
310            error_code: decoder.read_i16()?,
311        })
312    }
313}
314
315#[cfg(test)]
316#[allow(clippy::unwrap_used)]
317mod tests {
318    use super::{
319        OffsetCommitPartition, OffsetCommitPartitionResponse, OffsetCommitPartitionV7,
320        OffsetCommitPartitionV9, OffsetCommitRequestV2, OffsetCommitRequestV7,
321        OffsetCommitRequestV9, OffsetCommitResponseV2, OffsetCommitResponseV7,
322        OffsetCommitResponseV9, OffsetCommitTopic, OffsetCommitTopicResponse, OffsetCommitTopicV7,
323        OffsetCommitTopicV9,
324    };
325    use crate::codec::{Decoder, Encoder};
326
327    #[test]
328    fn encodes_offset_commit_v2_request() {
329        let request = OffsetCommitRequestV2 {
330            correlation_id: 23,
331            client_id: Some("kafrust".to_owned()),
332            group_id: "orders-group".to_owned(),
333            generation_id_or_member_epoch: 7,
334            member_id: "member-a".to_owned(),
335            retention_time_ms: 86_400_000,
336            topics: vec![OffsetCommitTopic {
337                name: "orders".to_owned(),
338                partitions: vec![OffsetCommitPartition {
339                    partition_index: 0,
340                    committed_offset: 42,
341                    committed_metadata: Some("processed".to_owned()),
342                }],
343            }],
344        };
345
346        assert_eq!(
347            request.encode().unwrap(),
348            [
349                0, 8, // api key
350                0, 2, // api version
351                0, 0, 0, 23, // correlation id
352                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
353                0, 12, b'o', b'r', b'd', b'e', b'r', b's', b'-', b'g', b'r', b'o', b'u',
354                b'p', // group id
355                0, 0, 0, 7, // generation id
356                0, 8, b'm', b'e', b'm', b'b', b'e', b'r', b'-', b'a', // member id
357                0, 0, 0, 0, 5, 38, 92, 0, // retention time
358                0, 0, 0, 1, // topic count
359                0, 6, b'o', b'r', b'd', b'e', b'r', b's', // topic
360                0, 0, 0, 1, // partition count
361                0, 0, 0, 0, // partition
362                0, 0, 0, 0, 0, 0, 0, 42, // committed offset
363                0, 9, b'p', b'r', b'o', b'c', b'e', b's', b's', b'e', b'd', // metadata
364            ]
365        );
366    }
367
368    #[test]
369    fn decodes_offset_commit_v2_response() {
370        let mut bytes = Encoder::new();
371        bytes.write_i32(1);
372        bytes.write_string("orders").unwrap();
373        bytes.write_i32(1);
374        bytes.write_i32(0);
375        bytes.write_i16(0);
376        let bytes = bytes.into_bytes();
377
378        let mut decoder = Decoder::new(&bytes);
379        let response = OffsetCommitResponseV2::decode_body(&mut decoder).unwrap();
380
381        assert_eq!(
382            response.topics,
383            vec![OffsetCommitTopicResponse {
384                name: "orders".to_owned(),
385                partitions: vec![OffsetCommitPartitionResponse {
386                    partition_index: 0,
387                    error_code: 0,
388                }],
389            }]
390        );
391        assert!(decoder.is_empty());
392    }
393
394    #[test]
395    fn encodes_offset_commit_v7_request_with_static_member() {
396        let request = OffsetCommitRequestV7 {
397            correlation_id: 23,
398            client_id: Some("kafrust".to_owned()),
399            group_id: "orders-group".to_owned(),
400            generation_id_or_member_epoch: 7,
401            member_id: "member-a".to_owned(),
402            group_instance_id: Some("orders-reader-1".to_owned()),
403            topics: vec![OffsetCommitTopicV7 {
404                name: "orders".to_owned(),
405                partitions: vec![OffsetCommitPartitionV7 {
406                    partition_index: 0,
407                    committed_offset: 42,
408                    committed_leader_epoch: -1,
409                    committed_metadata: None,
410                }],
411            }],
412        };
413
414        let encoded = request.encode().unwrap();
415        assert_eq!(&encoded[0..4], &[0, 8, 0, 7]);
416        assert!(encoded
417            .windows(17)
418            .any(|bytes| bytes == b"\0\x0forders-reader-1"));
419        assert!(encoded.windows(4).any(|bytes| bytes == [u8::MAX; 4]));
420    }
421
422    #[test]
423    fn decodes_offset_commit_v7_response() {
424        let mut bytes = Encoder::new();
425        bytes.write_i32(12);
426        bytes.write_i32(1);
427        bytes.write_string("orders").unwrap();
428        bytes.write_i32(1);
429        bytes.write_i32(0);
430        bytes.write_i16(0);
431        let bytes = bytes.into_bytes();
432
433        let mut decoder = Decoder::new(&bytes);
434        let response = OffsetCommitResponseV7::decode_body(&mut decoder).unwrap();
435
436        assert_eq!(response.throttle_time_ms, 12);
437        assert_eq!(response.topics[0].partitions[0].error_code, 0);
438        assert!(decoder.is_empty());
439    }
440
441    #[test]
442    fn encodes_offset_commit_v9_request_for_consumer_protocol() {
443        let request = OffsetCommitRequestV9 {
444            correlation_id: 23,
445            client_id: Some("kafrust".to_owned()),
446            group_id: "orders-group".to_owned(),
447            generation_id_or_member_epoch: 7,
448            member_id: "member-a".to_owned(),
449            group_instance_id: Some("orders-reader-1".to_owned()),
450            topics: vec![OffsetCommitTopicV9 {
451                name: "orders".to_owned(),
452                partitions: vec![OffsetCommitPartitionV9 {
453                    partition_index: 0,
454                    committed_offset: 42,
455                    committed_leader_epoch: -1,
456                    committed_metadata: None,
457                }],
458            }],
459        };
460
461        let encoded = request.encode().unwrap();
462        assert_eq!(&encoded[0..4], &[0, 8, 0, 9]);
463        assert_eq!(
464            &encoded[8..18],
465            &[0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', 0]
466        );
467
468        let mut decoder = Decoder::new(&encoded[18..]);
469        assert_eq!(decoder.read_compact_string().unwrap(), "orders-group");
470        assert_eq!(decoder.read_i32().unwrap(), 7);
471        assert_eq!(decoder.read_compact_string().unwrap(), "member-a");
472        assert_eq!(
473            decoder.read_compact_nullable_string().unwrap(),
474            Some("orders-reader-1".to_owned())
475        );
476        let topics = decoder
477            .read_compact_array("offset commit topics", |decoder| {
478                let name = decoder.read_compact_string()?;
479                let partitions = decoder
480                    .read_compact_array("offset commit partitions", |decoder| {
481                        let partition_index = decoder.read_i32()?;
482                        let committed_offset = decoder.read_i64()?;
483                        let committed_leader_epoch = decoder.read_i32()?;
484                        let committed_metadata = decoder.read_compact_nullable_string()?;
485                        decoder.read_tagged_fields()?;
486                        Ok((
487                            partition_index,
488                            committed_offset,
489                            committed_leader_epoch,
490                            committed_metadata,
491                        ))
492                    })?
493                    .unwrap_or_default();
494                decoder.read_tagged_fields()?;
495                Ok((name, partitions))
496            })
497            .unwrap()
498            .unwrap();
499        assert_eq!(topics[0].0, "orders");
500        assert_eq!(topics[0].1[0].0, 0);
501        assert_eq!(topics[0].1[0].1, 42);
502        assert_eq!(topics[0].1[0].2, -1);
503        assert_eq!(topics[0].1[0].3, None);
504        decoder.read_tagged_fields().unwrap();
505        assert!(decoder.is_empty());
506    }
507
508    #[test]
509    fn decodes_offset_commit_v9_response() {
510        let mut bytes = Encoder::new();
511        bytes.write_i32(12);
512        bytes
513            .write_compact_array(Some(&[()]), |encoder, ()| {
514                encoder.write_compact_string("orders")?;
515                encoder.write_compact_array(Some(&[()]), |encoder, ()| {
516                    encoder.write_i32(0);
517                    encoder.write_i16(0);
518                    encoder.write_empty_tagged_fields();
519                    Ok(())
520                })?;
521                encoder.write_empty_tagged_fields();
522                Ok(())
523            })
524            .unwrap();
525        bytes.write_empty_tagged_fields();
526
527        let bytes = bytes.into_bytes();
528        let mut decoder = Decoder::new(&bytes);
529        let response = OffsetCommitResponseV9::decode_body(&mut decoder).unwrap();
530
531        assert_eq!(response.throttle_time_ms, 12);
532        assert_eq!(response.topics[0].name, "orders");
533        assert_eq!(response.topics[0].partitions[0].partition_index, 0);
534        assert_eq!(response.topics[0].partitions[0].error_code, 0);
535        assert!(decoder.is_empty());
536    }
537}
kafrust_protocol/api/offset_commit.rs

kafrust_protocol/api/
offset_commit.rs
