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/// OffsetCommit v10, which uses topic UUIDs instead of topic names.
110///
111/// Kafka v10 is the topic-ID form used by the current consumer protocol. The
112/// request keeps the flexible v8/v9 body shape and changes only the topic
113/// identity field.
114#[derive(Debug, Clone, PartialEq, Eq)]
115pub struct OffsetCommitRequestV10 {
116    pub correlation_id: i32,
117    pub client_id: Option<String>,
118    pub group_id: String,
119    pub generation_id_or_member_epoch: i32,
120    pub member_id: String,
121    pub group_instance_id: Option<String>,
122    pub topics: Vec<OffsetCommitTopicV10>,
123}
124
125impl OffsetCommitRequestV10 {
126    pub fn encode(&self) -> Result<Vec<u8>> {
127        let mut encoder = Encoder::new();
128        RequestHeader {
129            api_key: API_KEY,
130            api_version: 10,
131            correlation_id: self.correlation_id,
132            client_id: self.client_id.clone(),
133        }
134        .encode_v2(&mut encoder)?;
135        encoder.write_compact_string(&self.group_id)?;
136        encoder.write_i32(self.generation_id_or_member_epoch);
137        encoder.write_compact_string(&self.member_id)?;
138        encoder.write_compact_nullable_string(self.group_instance_id.as_deref())?;
139        encoder.write_compact_array(Some(self.topics.as_slice()), |encoder, topic| {
140            topic.encode(encoder)
141        })?;
142        encoder.write_empty_tagged_fields();
143        Ok(encoder.into_bytes())
144    }
145}
146
147#[derive(Debug, Clone, PartialEq, Eq)]
148pub struct OffsetCommitTopic {
149    pub name: String,
150    pub partitions: Vec<OffsetCommitPartition>,
151}
152
153impl OffsetCommitTopic {
154    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
155        encoder.write_string(&self.name)?;
156        encoder.write_array(Some(self.partitions.as_slice()), |encoder, partition| {
157            partition.encode(encoder)
158        })
159    }
160}
161
162#[derive(Debug, Clone, PartialEq, Eq)]
163pub struct OffsetCommitTopicV7 {
164    pub name: String,
165    pub partitions: Vec<OffsetCommitPartitionV7>,
166}
167
168impl OffsetCommitTopicV7 {
169    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
170        encoder.write_string(&self.name)?;
171        encoder.write_array(Some(self.partitions.as_slice()), |encoder, partition| {
172            partition.encode(encoder)
173        })
174    }
175}
176
177#[derive(Debug, Clone, PartialEq, Eq)]
178pub struct OffsetCommitTopicV9 {
179    pub name: String,
180    pub partitions: Vec<OffsetCommitPartitionV9>,
181}
182
183impl OffsetCommitTopicV9 {
184    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
185        encoder.write_compact_string(&self.name)?;
186        encoder.write_compact_array(Some(self.partitions.as_slice()), |encoder, partition| {
187            partition.encode(encoder)
188        })?;
189        encoder.write_empty_tagged_fields();
190        Ok(())
191    }
192}
193
194#[derive(Debug, Clone, PartialEq, Eq)]
195pub struct OffsetCommitTopicV10 {
196    pub topic_id: [u8; 16],
197    pub partitions: Vec<OffsetCommitPartitionV10>,
198}
199
200impl OffsetCommitTopicV10 {
201    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
202        encoder.write_uuid(&self.topic_id);
203        encoder.write_compact_array(Some(self.partitions.as_slice()), |encoder, partition| {
204            partition.encode(encoder)
205        })?;
206        encoder.write_empty_tagged_fields();
207        Ok(())
208    }
209}
210
211#[derive(Debug, Clone, PartialEq, Eq)]
212pub struct OffsetCommitPartition {
213    pub partition_index: i32,
214    pub committed_offset: i64,
215    pub committed_metadata: Option<String>,
216}
217
218impl OffsetCommitPartition {
219    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
220        encoder.write_i32(self.partition_index);
221        encoder.write_i64(self.committed_offset);
222        encoder.write_nullable_string(self.committed_metadata.as_deref())
223    }
224}
225
226#[derive(Debug, Clone, PartialEq, Eq)]
227pub struct OffsetCommitPartitionV7 {
228    pub partition_index: i32,
229    pub committed_offset: i64,
230    pub committed_leader_epoch: i32,
231    pub committed_metadata: Option<String>,
232}
233
234impl OffsetCommitPartitionV7 {
235    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
236        encoder.write_i32(self.partition_index);
237        encoder.write_i64(self.committed_offset);
238        encoder.write_i32(self.committed_leader_epoch);
239        encoder.write_nullable_string(self.committed_metadata.as_deref())
240    }
241}
242
243#[derive(Debug, Clone, PartialEq, Eq)]
244pub struct OffsetCommitPartitionV9 {
245    pub partition_index: i32,
246    pub committed_offset: i64,
247    pub committed_leader_epoch: i32,
248    pub committed_metadata: Option<String>,
249}
250
251impl OffsetCommitPartitionV9 {
252    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
253        encoder.write_i32(self.partition_index);
254        encoder.write_i64(self.committed_offset);
255        encoder.write_i32(self.committed_leader_epoch);
256        encoder.write_compact_nullable_string(self.committed_metadata.as_deref())?;
257        encoder.write_empty_tagged_fields();
258        Ok(())
259    }
260}
261
262#[derive(Debug, Clone, PartialEq, Eq)]
263pub struct OffsetCommitPartitionV10 {
264    pub partition_index: i32,
265    pub committed_offset: i64,
266    pub committed_leader_epoch: i32,
267    pub committed_metadata: Option<String>,
268}
269
270impl OffsetCommitPartitionV10 {
271    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
272        encoder.write_i32(self.partition_index);
273        encoder.write_i64(self.committed_offset);
274        encoder.write_i32(self.committed_leader_epoch);
275        encoder.write_compact_nullable_string(self.committed_metadata.as_deref())?;
276        encoder.write_empty_tagged_fields();
277        Ok(())
278    }
279}
280
281#[derive(Debug, Clone, PartialEq, Eq)]
282pub struct OffsetCommitResponseV2 {
283    pub topics: Vec<OffsetCommitTopicResponse>,
284}
285
286impl OffsetCommitResponseV2 {
287    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
288        Ok(Self {
289            topics: decoder
290                .read_array(
291                    "offset commit topic responses",
292                    OffsetCommitTopicResponse::decode,
293                )?
294                .unwrap_or_default(),
295        })
296    }
297}
298
299#[derive(Debug, Clone, PartialEq, Eq)]
300pub struct OffsetCommitResponseV7 {
301    pub throttle_time_ms: i32,
302    pub topics: Vec<OffsetCommitTopicResponse>,
303}
304
305impl OffsetCommitResponseV7 {
306    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
307        Ok(Self {
308            throttle_time_ms: decoder.read_i32()?,
309            topics: decoder
310                .read_array(
311                    "offset commit topic responses",
312                    OffsetCommitTopicResponse::decode,
313                )?
314                .unwrap_or_default(),
315        })
316    }
317}
318
319#[derive(Debug, Clone, PartialEq, Eq)]
320pub struct OffsetCommitResponseV9 {
321    pub throttle_time_ms: i32,
322    pub topics: Vec<OffsetCommitTopicResponse>,
323}
324
325impl OffsetCommitResponseV9 {
326    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
327        let throttle_time_ms = decoder.read_i32()?;
328        let topics = decoder
329            .read_compact_array("offset commit topic responses", |decoder| {
330                let name = decoder.read_compact_string()?;
331                let partitions = decoder
332                    .read_compact_array("offset commit partition responses", |decoder| {
333                        let partition_index = decoder.read_i32()?;
334                        let error_code = decoder.read_i16()?;
335                        decoder.read_tagged_fields()?;
336                        Ok(OffsetCommitPartitionResponse {
337                            partition_index,
338                            error_code,
339                        })
340                    })?
341                    .unwrap_or_default();
342                decoder.read_tagged_fields()?;
343                Ok(OffsetCommitTopicResponse { name, partitions })
344            })?
345            .unwrap_or_default();
346        decoder.read_tagged_fields()?;
347        Ok(Self {
348            throttle_time_ms,
349            topics,
350        })
351    }
352}
353
354/// OffsetCommit v10 response with topic UUIDs.
355#[derive(Debug, Clone, PartialEq, Eq)]
356pub struct OffsetCommitResponseV10 {
357    pub throttle_time_ms: i32,
358    pub topics: Vec<OffsetCommitTopicResponseV10>,
359}
360
361impl OffsetCommitResponseV10 {
362    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
363        let throttle_time_ms = decoder.read_i32()?;
364        let topics = decoder
365            .read_compact_array("offset commit topic UUID responses", |decoder| {
366                let topic_id = decoder.read_uuid()?;
367                let partitions = decoder
368                    .read_compact_array("offset commit partition responses", |decoder| {
369                        let partition_index = decoder.read_i32()?;
370                        let error_code = decoder.read_i16()?;
371                        decoder.read_tagged_fields()?;
372                        Ok(OffsetCommitPartitionResponse {
373                            partition_index,
374                            error_code,
375                        })
376                    })?
377                    .unwrap_or_default();
378                decoder.read_tagged_fields()?;
379                Ok(OffsetCommitTopicResponseV10 {
380                    topic_id,
381                    partitions,
382                })
383            })?
384            .unwrap_or_default();
385        decoder.read_tagged_fields()?;
386        Ok(Self {
387            throttle_time_ms,
388            topics,
389        })
390    }
391}
392
393#[derive(Debug, Clone, PartialEq, Eq)]
394pub struct OffsetCommitTopicResponse {
395    pub name: String,
396    pub partitions: Vec<OffsetCommitPartitionResponse>,
397}
398
399#[derive(Debug, Clone, PartialEq, Eq)]
400pub struct OffsetCommitTopicResponseV10 {
401    pub topic_id: [u8; 16],
402    pub partitions: Vec<OffsetCommitPartitionResponse>,
403}
404
405impl OffsetCommitTopicResponse {
406    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
407        Ok(Self {
408            name: decoder.read_string()?,
409            partitions: decoder
410                .read_array(
411                    "offset commit partition responses",
412                    OffsetCommitPartitionResponse::decode,
413                )?
414                .unwrap_or_default(),
415        })
416    }
417}
418
419#[derive(Debug, Clone, PartialEq, Eq)]
420pub struct OffsetCommitPartitionResponse {
421    pub partition_index: i32,
422    pub error_code: i16,
423}
424
425impl OffsetCommitPartitionResponse {
426    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
427        Ok(Self {
428            partition_index: decoder.read_i32()?,
429            error_code: decoder.read_i16()?,
430        })
431    }
432}
433
434#[cfg(test)]
435#[allow(clippy::unwrap_used)]
436mod tests {
437    use super::{
438        OffsetCommitPartition, OffsetCommitPartitionResponse, OffsetCommitPartitionV10,
439        OffsetCommitPartitionV7, OffsetCommitPartitionV9, OffsetCommitRequestV10,
440        OffsetCommitRequestV2, OffsetCommitRequestV7, OffsetCommitRequestV9,
441        OffsetCommitResponseV10, OffsetCommitResponseV2, OffsetCommitResponseV7,
442        OffsetCommitResponseV9, OffsetCommitTopic, OffsetCommitTopicResponse, OffsetCommitTopicV10,
443        OffsetCommitTopicV7, OffsetCommitTopicV9,
444    };
445    use crate::codec::{Decoder, Encoder};
446
447    #[test]
448    fn encodes_offset_commit_v2_request() {
449        let request = OffsetCommitRequestV2 {
450            correlation_id: 23,
451            client_id: Some("kafrust".to_owned()),
452            group_id: "orders-group".to_owned(),
453            generation_id_or_member_epoch: 7,
454            member_id: "member-a".to_owned(),
455            retention_time_ms: 86_400_000,
456            topics: vec![OffsetCommitTopic {
457                name: "orders".to_owned(),
458                partitions: vec![OffsetCommitPartition {
459                    partition_index: 0,
460                    committed_offset: 42,
461                    committed_metadata: Some("processed".to_owned()),
462                }],
463            }],
464        };
465
466        assert_eq!(
467            request.encode().unwrap(),
468            [
469                0, 8, // api key
470                0, 2, // api version
471                0, 0, 0, 23, // correlation id
472                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
473                0, 12, b'o', b'r', b'd', b'e', b'r', b's', b'-', b'g', b'r', b'o', b'u',
474                b'p', // group id
475                0, 0, 0, 7, // generation id
476                0, 8, b'm', b'e', b'm', b'b', b'e', b'r', b'-', b'a', // member id
477                0, 0, 0, 0, 5, 38, 92, 0, // retention time
478                0, 0, 0, 1, // topic count
479                0, 6, b'o', b'r', b'd', b'e', b'r', b's', // topic
480                0, 0, 0, 1, // partition count
481                0, 0, 0, 0, // partition
482                0, 0, 0, 0, 0, 0, 0, 42, // committed offset
483                0, 9, b'p', b'r', b'o', b'c', b'e', b's', b's', b'e', b'd', // metadata
484            ]
485        );
486    }
487
488    #[test]
489    fn decodes_offset_commit_v2_response() {
490        let mut bytes = Encoder::new();
491        bytes.write_i32(1);
492        bytes.write_string("orders").unwrap();
493        bytes.write_i32(1);
494        bytes.write_i32(0);
495        bytes.write_i16(0);
496        let bytes = bytes.into_bytes();
497
498        let mut decoder = Decoder::new(&bytes);
499        let response = OffsetCommitResponseV2::decode_body(&mut decoder).unwrap();
500
501        assert_eq!(
502            response.topics,
503            vec![OffsetCommitTopicResponse {
504                name: "orders".to_owned(),
505                partitions: vec![OffsetCommitPartitionResponse {
506                    partition_index: 0,
507                    error_code: 0,
508                }],
509            }]
510        );
511        assert!(decoder.is_empty());
512    }
513
514    #[test]
515    fn encodes_offset_commit_v7_request_with_static_member() {
516        let request = OffsetCommitRequestV7 {
517            correlation_id: 23,
518            client_id: Some("kafrust".to_owned()),
519            group_id: "orders-group".to_owned(),
520            generation_id_or_member_epoch: 7,
521            member_id: "member-a".to_owned(),
522            group_instance_id: Some("orders-reader-1".to_owned()),
523            topics: vec![OffsetCommitTopicV7 {
524                name: "orders".to_owned(),
525                partitions: vec![OffsetCommitPartitionV7 {
526                    partition_index: 0,
527                    committed_offset: 42,
528                    committed_leader_epoch: -1,
529                    committed_metadata: None,
530                }],
531            }],
532        };
533
534        let encoded = request.encode().unwrap();
535        assert_eq!(&encoded[0..4], &[0, 8, 0, 7]);
536        assert!(encoded
537            .windows(17)
538            .any(|bytes| bytes == b"\0\x0forders-reader-1"));
539        assert!(encoded.windows(4).any(|bytes| bytes == [u8::MAX; 4]));
540    }
541
542    #[test]
543    fn decodes_offset_commit_v7_response() {
544        let mut bytes = Encoder::new();
545        bytes.write_i32(12);
546        bytes.write_i32(1);
547        bytes.write_string("orders").unwrap();
548        bytes.write_i32(1);
549        bytes.write_i32(0);
550        bytes.write_i16(0);
551        let bytes = bytes.into_bytes();
552
553        let mut decoder = Decoder::new(&bytes);
554        let response = OffsetCommitResponseV7::decode_body(&mut decoder).unwrap();
555
556        assert_eq!(response.throttle_time_ms, 12);
557        assert_eq!(response.topics[0].partitions[0].error_code, 0);
558        assert!(decoder.is_empty());
559    }
560
561    #[test]
562    fn encodes_offset_commit_v9_request_for_consumer_protocol() {
563        let request = OffsetCommitRequestV9 {
564            correlation_id: 23,
565            client_id: Some("kafrust".to_owned()),
566            group_id: "orders-group".to_owned(),
567            generation_id_or_member_epoch: 7,
568            member_id: "member-a".to_owned(),
569            group_instance_id: Some("orders-reader-1".to_owned()),
570            topics: vec![OffsetCommitTopicV9 {
571                name: "orders".to_owned(),
572                partitions: vec![OffsetCommitPartitionV9 {
573                    partition_index: 0,
574                    committed_offset: 42,
575                    committed_leader_epoch: -1,
576                    committed_metadata: None,
577                }],
578            }],
579        };
580
581        let encoded = request.encode().unwrap();
582        assert_eq!(&encoded[0..4], &[0, 8, 0, 9]);
583        assert_eq!(
584            &encoded[8..18],
585            &[0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', 0]
586        );
587
588        let mut decoder = Decoder::new(&encoded[18..]);
589        assert_eq!(decoder.read_compact_string().unwrap(), "orders-group");
590        assert_eq!(decoder.read_i32().unwrap(), 7);
591        assert_eq!(decoder.read_compact_string().unwrap(), "member-a");
592        assert_eq!(
593            decoder.read_compact_nullable_string().unwrap(),
594            Some("orders-reader-1".to_owned())
595        );
596        let topics = decoder
597            .read_compact_array("offset commit topics", |decoder| {
598                let name = decoder.read_compact_string()?;
599                let partitions = decoder
600                    .read_compact_array("offset commit partitions", |decoder| {
601                        let partition_index = decoder.read_i32()?;
602                        let committed_offset = decoder.read_i64()?;
603                        let committed_leader_epoch = decoder.read_i32()?;
604                        let committed_metadata = decoder.read_compact_nullable_string()?;
605                        decoder.read_tagged_fields()?;
606                        Ok((
607                            partition_index,
608                            committed_offset,
609                            committed_leader_epoch,
610                            committed_metadata,
611                        ))
612                    })?
613                    .unwrap_or_default();
614                decoder.read_tagged_fields()?;
615                Ok((name, partitions))
616            })
617            .unwrap()
618            .unwrap();
619        assert_eq!(topics[0].0, "orders");
620        assert_eq!(topics[0].1[0].0, 0);
621        assert_eq!(topics[0].1[0].1, 42);
622        assert_eq!(topics[0].1[0].2, -1);
623        assert_eq!(topics[0].1[0].3, None);
624        decoder.read_tagged_fields().unwrap();
625        assert!(decoder.is_empty());
626    }
627
628    #[test]
629    fn decodes_offset_commit_v9_response() {
630        let mut bytes = Encoder::new();
631        bytes.write_i32(12);
632        bytes
633            .write_compact_array(Some(&[()]), |encoder, ()| {
634                encoder.write_compact_string("orders")?;
635                encoder.write_compact_array(Some(&[()]), |encoder, ()| {
636                    encoder.write_i32(0);
637                    encoder.write_i16(0);
638                    encoder.write_empty_tagged_fields();
639                    Ok(())
640                })?;
641                encoder.write_empty_tagged_fields();
642                Ok(())
643            })
644            .unwrap();
645        bytes.write_empty_tagged_fields();
646
647        let bytes = bytes.into_bytes();
648        let mut decoder = Decoder::new(&bytes);
649        let response = OffsetCommitResponseV9::decode_body(&mut decoder).unwrap();
650
651        assert_eq!(response.throttle_time_ms, 12);
652        assert_eq!(response.topics[0].name, "orders");
653        assert_eq!(response.topics[0].partitions[0].partition_index, 0);
654        assert_eq!(response.topics[0].partitions[0].error_code, 0);
655        assert!(decoder.is_empty());
656    }
657
658    #[test]
659    fn encodes_offset_commit_v10_request_with_topic_uuid() {
660        let request = OffsetCommitRequestV10 {
661            correlation_id: 37,
662            client_id: Some("kafrust".to_owned()),
663            group_id: "orders-group".to_owned(),
664            generation_id_or_member_epoch: 7,
665            member_id: "member-a".to_owned(),
666            group_instance_id: None,
667            topics: vec![OffsetCommitTopicV10 {
668                topic_id: [7; 16],
669                partitions: vec![OffsetCommitPartitionV10 {
670                    partition_index: 2,
671                    committed_offset: 42,
672                    committed_leader_epoch: 9,
673                    committed_metadata: Some("processed".to_owned()),
674                }],
675            }],
676        };
677
678        let encoded = request.encode().unwrap();
679        assert_eq!(&encoded[0..4], &[0, 8, 0, 10]);
680        let mut decoder = Decoder::new(&encoded[18..]);
681        assert_eq!(decoder.read_compact_string().unwrap(), "orders-group");
682        assert_eq!(decoder.read_i32().unwrap(), 7);
683        assert_eq!(decoder.read_compact_string().unwrap(), "member-a");
684        assert_eq!(decoder.read_compact_nullable_string().unwrap(), None);
685        let topics = decoder
686            .read_compact_array("offset commit topics", |decoder| {
687                let topic_id = decoder.read_uuid()?;
688                let partitions = decoder
689                    .read_compact_array("offset commit partitions", |decoder| {
690                        let partition_index = decoder.read_i32()?;
691                        let committed_offset = decoder.read_i64()?;
692                        let committed_leader_epoch = decoder.read_i32()?;
693                        let committed_metadata = decoder.read_compact_nullable_string()?;
694                        decoder.read_tagged_fields()?;
695                        Ok((
696                            partition_index,
697                            committed_offset,
698                            committed_leader_epoch,
699                            committed_metadata,
700                        ))
701                    })?
702                    .unwrap_or_default();
703                decoder.read_tagged_fields()?;
704                Ok((topic_id, partitions))
705            })
706            .unwrap()
707            .unwrap();
708        assert_eq!(topics[0].0, [7; 16]);
709        assert_eq!(topics[0].1[0].0, 2);
710        assert_eq!(topics[0].1[0].1, 42);
711        assert_eq!(topics[0].1[0].2, 9);
712        assert_eq!(topics[0].1[0].3, Some("processed".to_owned()));
713        decoder.read_tagged_fields().unwrap();
714        assert!(decoder.is_empty());
715    }
716
717    #[test]
718    fn decodes_offset_commit_v10_response_with_topic_uuid() {
719        let mut bytes = Encoder::new();
720        bytes.write_i32(12);
721        bytes
722            .write_compact_array(Some(&[()]), |encoder, ()| {
723                encoder.write_uuid(&[8; 16]);
724                encoder.write_compact_array(Some(&[()]), |encoder, ()| {
725                    encoder.write_i32(0);
726                    encoder.write_i16(0);
727                    encoder.write_empty_tagged_fields();
728                    Ok(())
729                })?;
730                encoder.write_empty_tagged_fields();
731                Ok(())
732            })
733            .unwrap();
734        bytes.write_empty_tagged_fields();
735
736        let bytes = bytes.into_bytes();
737        let mut decoder = Decoder::new(&bytes);
738        let response = OffsetCommitResponseV10::decode_body(&mut decoder).unwrap();
739        assert_eq!(response.throttle_time_ms, 12);
740        assert_eq!(response.topics[0].topic_id, [8; 16]);
741        assert_eq!(response.topics[0].partitions[0].partition_index, 0);
742        assert_eq!(response.topics[0].partitions[0].error_code, 0);
743        assert!(decoder.is_empty());
744    }
745}
kafrust_protocol/api/offset_commit.rs

kafrust_protocol/api/
offset_commit.rs
