kafrust_protocol/api/
share_group_state.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5/// Kafka InitializeShareGroupState API key.
6pub const INITIALIZE_API_KEY: i16 = 83;
7/// Kafka ReadShareGroupState API key.
8pub const READ_API_KEY: i16 = 84;
9/// Kafka WriteShareGroupState API key.
10pub const WRITE_API_KEY: i16 = 85;
11/// Kafka DeleteShareGroupState API key.
12pub const DELETE_API_KEY: i16 = 86;
13/// Kafka ReadShareGroupStateSummary API key.
14pub const SUMMARY_API_KEY: i16 = 87;
15
16/// One partition in an InitializeShareGroupState request.
17#[derive(Debug, Clone, Copy, PartialEq, Eq)]
18pub struct InitializeShareGroupStatePartition {
19    pub partition: i32,
20    pub state_epoch: i32,
21    pub start_offset: i64,
22}
23
24/// One topic in an InitializeShareGroupState request.
25#[derive(Debug, Clone, PartialEq, Eq)]
26pub struct InitializeShareGroupStateTopic {
27    pub topic_id: [u8; 16],
28    pub partitions: Vec<InitializeShareGroupStatePartition>,
29}
30
31/// InitializeShareGroupState v0 request.
32#[derive(Debug, Clone, PartialEq, Eq)]
33pub struct InitializeShareGroupStateRequestV0 {
34    pub correlation_id: i32,
35    pub client_id: Option<String>,
36    pub group_id: String,
37    pub topics: Vec<InitializeShareGroupStateTopic>,
38}
39
40impl InitializeShareGroupStateRequestV0 {
41    /// Encodes the flexible request, including its request header.
42    pub fn encode(&self) -> Result<Vec<u8>> {
43        encode_request(
44            INITIALIZE_API_KEY,
45            0,
46            self.correlation_id,
47            self.client_id.clone(),
48            &self.group_id,
49            &self.topics,
50            |encoder, topic| {
51                encoder.write_uuid(&topic.topic_id);
52                encoder.write_compact_array(Some(&topic.partitions), |encoder, partition| {
53                    encoder.write_i32(partition.partition);
54                    encoder.write_i32(partition.state_epoch);
55                    encoder.write_i64(partition.start_offset);
56                    encoder.write_empty_tagged_fields();
57                    Ok(())
58                })?;
59                encoder.write_empty_tagged_fields();
60                Ok(())
61            },
62        )
63    }
64}
65
66/// One partition in a ReadShareGroupState or ReadShareGroupStateSummary request.
67#[derive(Debug, Clone, Copy, PartialEq, Eq)]
68pub struct ReadShareGroupStatePartition {
69    pub partition: i32,
70    pub leader_epoch: i32,
71}
72
73/// One topic in a ReadShareGroupState or ReadShareGroupStateSummary request.
74#[derive(Debug, Clone, PartialEq, Eq)]
75pub struct ReadShareGroupStateTopic {
76    pub topic_id: [u8; 16],
77    pub partitions: Vec<ReadShareGroupStatePartition>,
78}
79
80fn encode_read_request(
81    api_key: i16,
82    api_version: i16,
83    correlation_id: i32,
84    client_id: Option<String>,
85    group_id: &str,
86    topics: &[ReadShareGroupStateTopic],
87) -> Result<Vec<u8>> {
88    encode_request(
89        api_key,
90        api_version,
91        correlation_id,
92        client_id,
93        group_id,
94        topics,
95        |encoder, topic| {
96            encoder.write_uuid(&topic.topic_id);
97            encoder.write_compact_array(Some(&topic.partitions), |encoder, partition| {
98                encoder.write_i32(partition.partition);
99                encoder.write_i32(partition.leader_epoch);
100                encoder.write_empty_tagged_fields();
101                Ok(())
102            })?;
103            encoder.write_empty_tagged_fields();
104            Ok(())
105        },
106    )
107}
108
109fn encode_request<T>(
110    api_key: i16,
111    api_version: i16,
112    correlation_id: i32,
113    client_id: Option<String>,
114    group_id: &str,
115    topics: &[T],
116    mut encode_topic: impl FnMut(&mut Encoder, &T) -> Result<()>,
117) -> Result<Vec<u8>> {
118    let mut encoder = Encoder::new();
119    RequestHeader {
120        api_key,
121        api_version,
122        correlation_id,
123        client_id,
124    }
125    .encode_v2(&mut encoder)?;
126    encoder.write_compact_string(group_id)?;
127    encoder.write_compact_array(Some(topics), |encoder, topic| encode_topic(encoder, topic))?;
128    encoder.write_empty_tagged_fields();
129    Ok(encoder.into_bytes())
130}
131
132/// ReadShareGroupState v0 request.
133#[derive(Debug, Clone, PartialEq, Eq)]
134pub struct ReadShareGroupStateRequestV0 {
135    pub correlation_id: i32,
136    pub client_id: Option<String>,
137    pub group_id: String,
138    pub topics: Vec<ReadShareGroupStateTopic>,
139}
140
141impl ReadShareGroupStateRequestV0 {
142    /// Encodes the flexible request, including its request header.
143    pub fn encode(&self) -> Result<Vec<u8>> {
144        encode_read_request(
145            READ_API_KEY,
146            0,
147            self.correlation_id,
148            self.client_id.clone(),
149            &self.group_id,
150            &self.topics,
151        )
152    }
153}
154
155/// One state batch returned by ReadShareGroupState.
156#[derive(Debug, Clone, Copy, PartialEq, Eq)]
157pub struct ShareGroupStateBatch {
158    pub first_offset: i64,
159    pub last_offset: i64,
160    pub delivery_state: i8,
161    pub delivery_count: i16,
162}
163
164/// One partition result returned by an Initialize, Write, or Delete operation.
165#[derive(Debug, Clone, PartialEq, Eq)]
166pub struct ShareGroupStatePartitionResult {
167    pub partition: i32,
168    pub error_code: i16,
169    pub error_message: Option<String>,
170}
171
172/// One topic result returned by an Initialize, Write, or Delete operation.
173#[derive(Debug, Clone, PartialEq, Eq)]
174pub struct ShareGroupStateTopicResult {
175    pub topic_id: [u8; 16],
176    pub partitions: Vec<ShareGroupStatePartitionResult>,
177}
178
179fn decode_state_results(
180    decoder: &mut Decoder<'_>,
181    kind: &'static str,
182) -> Result<Vec<ShareGroupStateTopicResult>> {
183    let results = decoder
184        .read_compact_array(kind, |decoder| {
185            let topic_id = decoder.read_uuid()?;
186            let partitions = decoder
187                .read_compact_array("share group state result partitions", |decoder| {
188                    let result = ShareGroupStatePartitionResult {
189                        partition: decoder.read_i32()?,
190                        error_code: decoder.read_i16()?,
191                        error_message: decoder.read_compact_nullable_string()?,
192                    };
193                    decoder.read_tagged_fields()?;
194                    Ok(result)
195                })?
196                .unwrap_or_default();
197            decoder.read_tagged_fields()?;
198            Ok(ShareGroupStateTopicResult {
199                topic_id,
200                partitions,
201            })
202        })?
203        .unwrap_or_default();
204    Ok(results)
205}
206
207/// Response body shared by InitializeShareGroupState, WriteShareGroupState,
208/// and DeleteShareGroupState v0/v1 responses.
209#[derive(Debug, Clone, PartialEq, Eq)]
210pub struct ShareGroupStateResultResponse {
211    pub results: Vec<ShareGroupStateTopicResult>,
212}
213
214impl ShareGroupStateResultResponse {
215    /// Decodes the flexible response body after the response header.
216    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
217        let results = decode_state_results(decoder, "share group state results")?;
218        decoder.read_tagged_fields()?;
219        Ok(Self { results })
220    }
221}
222
223/// InitializeShareGroupState v0 response.
224pub type InitializeShareGroupStateResponseV0 = ShareGroupStateResultResponse;
225
226/// One partition result returned by ReadShareGroupState.
227#[derive(Debug, Clone, PartialEq, Eq)]
228pub struct ReadShareGroupStatePartitionResult {
229    pub partition: i32,
230    pub error_code: i16,
231    pub error_message: Option<String>,
232    pub state_epoch: i32,
233    pub start_offset: i64,
234    pub state_batches: Vec<ShareGroupStateBatch>,
235}
236
237/// One topic result returned by ReadShareGroupState.
238#[derive(Debug, Clone, PartialEq, Eq)]
239pub struct ReadShareGroupStateTopicResult {
240    pub topic_id: [u8; 16],
241    pub partitions: Vec<ReadShareGroupStatePartitionResult>,
242}
243
244/// ReadShareGroupState v0 response.
245#[derive(Debug, Clone, PartialEq, Eq)]
246pub struct ReadShareGroupStateResponseV0 {
247    pub results: Vec<ReadShareGroupStateTopicResult>,
248}
249
250impl ReadShareGroupStateResponseV0 {
251    /// Decodes the flexible response body after the response header.
252    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
253        Ok(Self {
254            results: decode_read_results(decoder, "read share group state results")?,
255        })
256    }
257}
258
259fn decode_read_results(
260    decoder: &mut Decoder<'_>,
261    kind: &'static str,
262) -> Result<Vec<ReadShareGroupStateTopicResult>> {
263    let results = decoder
264        .read_compact_array(kind, |decoder| {
265            let topic_id = decoder.read_uuid()?;
266            let partitions = decoder
267                .read_compact_array("read share group state partitions", |decoder| {
268                    let partition = ReadShareGroupStatePartitionResult {
269                        partition: decoder.read_i32()?,
270                        error_code: decoder.read_i16()?,
271                        error_message: decoder.read_compact_nullable_string()?,
272                        state_epoch: decoder.read_i32()?,
273                        start_offset: decoder.read_i64()?,
274                        state_batches: decoder
275                            .read_compact_array("share group state batches", |decoder| {
276                                let batch = ShareGroupStateBatch {
277                                    first_offset: decoder.read_i64()?,
278                                    last_offset: decoder.read_i64()?,
279                                    delivery_state: decoder.read_i8()?,
280                                    delivery_count: decoder.read_i16()?,
281                                };
282                                decoder.read_tagged_fields()?;
283                                Ok(batch)
284                            })?
285                            .unwrap_or_default(),
286                    };
287                    decoder.read_tagged_fields()?;
288                    Ok(partition)
289                })?
290                .unwrap_or_default();
291            decoder.read_tagged_fields()?;
292            Ok(ReadShareGroupStateTopicResult {
293                topic_id,
294                partitions,
295            })
296        })?
297        .unwrap_or_default();
298    decoder.read_tagged_fields()?;
299    Ok(results)
300}
301
302/// One partition in a WriteShareGroupState v0 request.
303#[derive(Debug, Clone, PartialEq, Eq)]
304pub struct WriteShareGroupStatePartitionV0 {
305    pub partition: i32,
306    pub state_epoch: i32,
307    pub leader_epoch: i32,
308    pub start_offset: i64,
309    pub state_batches: Vec<ShareGroupStateBatch>,
310}
311
312/// One topic in a WriteShareGroupState v0 request.
313#[derive(Debug, Clone, PartialEq, Eq)]
314pub struct WriteShareGroupStateTopicV0 {
315    pub topic_id: [u8; 16],
316    pub partitions: Vec<WriteShareGroupStatePartitionV0>,
317}
318
319/// One partition in a WriteShareGroupState v1 request.
320#[derive(Debug, Clone, PartialEq, Eq)]
321pub struct WriteShareGroupStatePartitionV1 {
322    pub partition: i32,
323    pub state_epoch: i32,
324    pub leader_epoch: i32,
325    pub start_offset: i64,
326    pub delivery_complete_count: i32,
327    pub state_batches: Vec<ShareGroupStateBatch>,
328}
329
330/// One topic in a WriteShareGroupState v1 request.
331#[derive(Debug, Clone, PartialEq, Eq)]
332pub struct WriteShareGroupStateTopicV1 {
333    pub topic_id: [u8; 16],
334    pub partitions: Vec<WriteShareGroupStatePartitionV1>,
335}
336
337fn encode_write_partition(
338    encoder: &mut Encoder,
339    partition: i32,
340    state_epoch: i32,
341    leader_epoch: i32,
342    start_offset: i64,
343    delivery_complete_count: Option<i32>,
344    state_batches: &[ShareGroupStateBatch],
345) -> Result<()> {
346    encoder.write_i32(partition);
347    encoder.write_i32(state_epoch);
348    encoder.write_i32(leader_epoch);
349    encoder.write_i64(start_offset);
350    if let Some(delivery_complete_count) = delivery_complete_count {
351        encoder.write_i32(delivery_complete_count);
352    }
353    encoder.write_compact_array(Some(state_batches), |encoder, batch| {
354        encoder.write_i64(batch.first_offset);
355        encoder.write_i64(batch.last_offset);
356        encoder.write_i8(batch.delivery_state);
357        encoder.write_i16(batch.delivery_count);
358        encoder.write_empty_tagged_fields();
359        Ok(())
360    })?;
361    encoder.write_empty_tagged_fields();
362    Ok(())
363}
364
365/// WriteShareGroupState v0 request.
366#[derive(Debug, Clone, PartialEq, Eq)]
367pub struct WriteShareGroupStateRequestV0 {
368    pub correlation_id: i32,
369    pub client_id: Option<String>,
370    pub group_id: String,
371    pub topics: Vec<WriteShareGroupStateTopicV0>,
372}
373
374impl WriteShareGroupStateRequestV0 {
375    /// Encodes the flexible request, including its request header.
376    pub fn encode(&self) -> Result<Vec<u8>> {
377        encode_request(
378            WRITE_API_KEY,
379            0,
380            self.correlation_id,
381            self.client_id.clone(),
382            &self.group_id,
383            &self.topics,
384            |encoder, topic| {
385                encoder.write_uuid(&topic.topic_id);
386                encoder.write_compact_array(Some(&topic.partitions), |encoder, partition| {
387                    encode_write_partition(
388                        encoder,
389                        partition.partition,
390                        partition.state_epoch,
391                        partition.leader_epoch,
392                        partition.start_offset,
393                        None,
394                        &partition.state_batches,
395                    )
396                })?;
397                encoder.write_empty_tagged_fields();
398                Ok(())
399            },
400        )
401    }
402}
403
404/// WriteShareGroupState v1 request.
405#[derive(Debug, Clone, PartialEq, Eq)]
406pub struct WriteShareGroupStateRequestV1 {
407    pub correlation_id: i32,
408    pub client_id: Option<String>,
409    pub group_id: String,
410    pub topics: Vec<WriteShareGroupStateTopicV1>,
411}
412
413impl WriteShareGroupStateRequestV1 {
414    /// Encodes the flexible request, including its request header.
415    pub fn encode(&self) -> Result<Vec<u8>> {
416        encode_request(
417            WRITE_API_KEY,
418            1,
419            self.correlation_id,
420            self.client_id.clone(),
421            &self.group_id,
422            &self.topics,
423            |encoder, topic| {
424                encoder.write_uuid(&topic.topic_id);
425                encoder.write_compact_array(Some(&topic.partitions), |encoder, partition| {
426                    encode_write_partition(
427                        encoder,
428                        partition.partition,
429                        partition.state_epoch,
430                        partition.leader_epoch,
431                        partition.start_offset,
432                        Some(partition.delivery_complete_count),
433                        &partition.state_batches,
434                    )
435                })?;
436                encoder.write_empty_tagged_fields();
437                Ok(())
438            },
439        )
440    }
441}
442
443/// WriteShareGroupState v0 response.
444pub type WriteShareGroupStateResponseV0 = ShareGroupStateResultResponse;
445/// WriteShareGroupState v1 response.
446pub type WriteShareGroupStateResponseV1 = ShareGroupStateResultResponse;
447
448/// One topic in a DeleteShareGroupState request.
449#[derive(Debug, Clone, PartialEq, Eq)]
450pub struct DeleteShareGroupStateTopic {
451    pub topic_id: [u8; 16],
452    pub partitions: Vec<i32>,
453}
454
455/// DeleteShareGroupState v0 request.
456#[derive(Debug, Clone, PartialEq, Eq)]
457pub struct DeleteShareGroupStateRequestV0 {
458    pub correlation_id: i32,
459    pub client_id: Option<String>,
460    pub group_id: String,
461    pub topics: Vec<DeleteShareGroupStateTopic>,
462}
463
464impl DeleteShareGroupStateRequestV0 {
465    /// Encodes the flexible request, including its request header.
466    pub fn encode(&self) -> Result<Vec<u8>> {
467        encode_request(
468            DELETE_API_KEY,
469            0,
470            self.correlation_id,
471            self.client_id.clone(),
472            &self.group_id,
473            &self.topics,
474            |encoder, topic| {
475                encoder.write_uuid(&topic.topic_id);
476                encoder.write_compact_array(Some(&topic.partitions), |encoder, partition| {
477                    encoder.write_i32(*partition);
478                    encoder.write_empty_tagged_fields();
479                    Ok(())
480                })?;
481                encoder.write_empty_tagged_fields();
482                Ok(())
483            },
484        )
485    }
486}
487
488/// DeleteShareGroupState v0 response.
489pub type DeleteShareGroupStateResponseV0 = ShareGroupStateResultResponse;
490
491/// ReadShareGroupStateSummary v0 request.
492#[derive(Debug, Clone, PartialEq, Eq)]
493pub struct ReadShareGroupStateSummaryRequestV0 {
494    pub correlation_id: i32,
495    pub client_id: Option<String>,
496    pub group_id: String,
497    pub topics: Vec<ReadShareGroupStateTopic>,
498}
499
500impl ReadShareGroupStateSummaryRequestV0 {
501    /// Encodes the flexible request, including its request header.
502    pub fn encode(&self) -> Result<Vec<u8>> {
503        encode_read_request(
504            SUMMARY_API_KEY,
505            0,
506            self.correlation_id,
507            self.client_id.clone(),
508            &self.group_id,
509            &self.topics,
510        )
511    }
512}
513
514/// ReadShareGroupStateSummary v1 request.
515#[derive(Debug, Clone, PartialEq, Eq)]
516pub struct ReadShareGroupStateSummaryRequestV1 {
517    pub correlation_id: i32,
518    pub client_id: Option<String>,
519    pub group_id: String,
520    pub topics: Vec<ReadShareGroupStateTopic>,
521}
522
523impl ReadShareGroupStateSummaryRequestV1 {
524    /// Encodes the flexible request, including its request header.
525    pub fn encode(&self) -> Result<Vec<u8>> {
526        encode_read_request(
527            SUMMARY_API_KEY,
528            1,
529            self.correlation_id,
530            self.client_id.clone(),
531            &self.group_id,
532            &self.topics,
533        )
534    }
535}
536
537/// One partition result returned by ReadShareGroupStateSummary.
538#[derive(Debug, Clone, PartialEq, Eq)]
539pub struct ReadShareGroupStateSummaryPartitionResult {
540    pub partition: i32,
541    pub error_code: i16,
542    pub error_message: Option<String>,
543    pub state_epoch: i32,
544    pub leader_epoch: i32,
545    pub start_offset: i64,
546    pub delivery_complete_count: Option<i32>,
547}
548
549/// One topic result returned by ReadShareGroupStateSummary.
550#[derive(Debug, Clone, PartialEq, Eq)]
551pub struct ReadShareGroupStateSummaryTopicResult {
552    pub topic_id: [u8; 16],
553    pub partitions: Vec<ReadShareGroupStateSummaryPartitionResult>,
554}
555
556fn decode_summary_results(
557    decoder: &mut Decoder<'_>,
558    api_version: i16,
559) -> Result<Vec<ReadShareGroupStateSummaryTopicResult>> {
560    let results = decoder
561        .read_compact_array("read share group state summary results", |decoder| {
562            let topic_id = decoder.read_uuid()?;
563            let partitions = decoder
564                .read_compact_array("share group state summary partitions", |decoder| {
565                    let partition = ReadShareGroupStateSummaryPartitionResult {
566                        partition: decoder.read_i32()?,
567                        error_code: decoder.read_i16()?,
568                        error_message: decoder.read_compact_nullable_string()?,
569                        state_epoch: decoder.read_i32()?,
570                        leader_epoch: decoder.read_i32()?,
571                        start_offset: decoder.read_i64()?,
572                        delivery_complete_count: (api_version >= 1)
573                            .then(|| decoder.read_i32())
574                            .transpose()?,
575                    };
576                    decoder.read_tagged_fields()?;
577                    Ok(partition)
578                })?
579                .unwrap_or_default();
580            decoder.read_tagged_fields()?;
581            Ok(ReadShareGroupStateSummaryTopicResult {
582                topic_id,
583                partitions,
584            })
585        })?
586        .unwrap_or_default();
587    decoder.read_tagged_fields()?;
588    Ok(results)
589}
590
591/// ReadShareGroupStateSummary v0 response.
592#[derive(Debug, Clone, PartialEq, Eq)]
593pub struct ReadShareGroupStateSummaryResponseV0 {
594    pub results: Vec<ReadShareGroupStateSummaryTopicResult>,
595}
596
597impl ReadShareGroupStateSummaryResponseV0 {
598    /// Decodes the flexible response body after the response header.
599    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
600        Ok(Self {
601            results: decode_summary_results(decoder, 0)?,
602        })
603    }
604}
605
606/// ReadShareGroupStateSummary v1 response.
607#[derive(Debug, Clone, PartialEq, Eq)]
608pub struct ReadShareGroupStateSummaryResponseV1 {
609    pub results: Vec<ReadShareGroupStateSummaryTopicResult>,
610}
611
612impl ReadShareGroupStateSummaryResponseV1 {
613    /// Decodes the flexible response body after the response header.
614    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
615        Ok(Self {
616            results: decode_summary_results(decoder, 1)?,
617        })
618    }
619}
620
621#[cfg(test)]
622#[allow(clippy::unwrap_used)]
623mod tests {
624    use super::*;
625    use crate::codec::{Decoder, Encoder};
626
627    fn read_header(decoder: &mut Decoder<'_>, api_key: i16, api_version: i16) {
628        assert_eq!(decoder.read_i16().unwrap(), api_key);
629        assert_eq!(decoder.read_i16().unwrap(), api_version);
630        assert_eq!(decoder.read_i32().unwrap(), 7);
631        assert_eq!(
632            decoder.read_nullable_string().unwrap().as_deref(),
633            Some("kafrust")
634        );
635        assert!(decoder.read_tagged_fields().unwrap().is_empty());
636    }
637
638    fn batch() -> ShareGroupStateBatch {
639        ShareGroupStateBatch {
640            first_offset: 10,
641            last_offset: 12,
642            delivery_state: 2,
643            delivery_count: 3,
644        }
645    }
646
647    #[test]
648    fn encodes_initialize_share_group_state_v0() {
649        let request = InitializeShareGroupStateRequestV0 {
650            correlation_id: 7,
651            client_id: Some("kafrust".to_owned()),
652            group_id: "share-orders".to_owned(),
653            topics: vec![InitializeShareGroupStateTopic {
654                topic_id: [1; 16],
655                partitions: vec![InitializeShareGroupStatePartition {
656                    partition: 2,
657                    state_epoch: 4,
658                    start_offset: 10,
659                }],
660            }],
661        };
662        let encoded = request.encode().unwrap();
663        let mut decoder = Decoder::new(&encoded);
664        read_header(&mut decoder, INITIALIZE_API_KEY, 0);
665        assert_eq!(decoder.read_compact_string().unwrap(), "share-orders");
666        assert_eq!(decoder.read_unsigned_varint().unwrap(), 2);
667        assert_eq!(decoder.read_uuid().unwrap(), [1; 16]);
668        assert_eq!(decoder.read_unsigned_varint().unwrap(), 2);
669        assert_eq!(decoder.read_i32().unwrap(), 2);
670        assert_eq!(decoder.read_i32().unwrap(), 4);
671        assert_eq!(decoder.read_i64().unwrap(), 10);
672        assert!(decoder.read_tagged_fields().unwrap().is_empty());
673        assert!(decoder.read_tagged_fields().unwrap().is_empty());
674        assert!(decoder.read_tagged_fields().unwrap().is_empty());
675        assert!(decoder.is_empty());
676    }
677
678    #[test]
679    fn encodes_write_v1_delivery_complete_count_without_v0_field() {
680        let topic = WriteShareGroupStateTopicV1 {
681            topic_id: [2; 16],
682            partitions: vec![WriteShareGroupStatePartitionV1 {
683                partition: 1,
684                state_epoch: 3,
685                leader_epoch: 4,
686                start_offset: 5,
687                delivery_complete_count: 6,
688                state_batches: vec![batch()],
689            }],
690        };
691        let request = WriteShareGroupStateRequestV1 {
692            correlation_id: 7,
693            client_id: Some("kafrust".to_owned()),
694            group_id: "share-orders".to_owned(),
695            topics: vec![topic],
696        };
697        let encoded = request.encode().unwrap();
698        let mut decoder = Decoder::new(&encoded);
699        read_header(&mut decoder, WRITE_API_KEY, 1);
700        assert_eq!(decoder.read_compact_string().unwrap(), "share-orders");
701        assert_eq!(decoder.read_unsigned_varint().unwrap(), 2);
702        assert_eq!(decoder.read_uuid().unwrap(), [2; 16]);
703        assert_eq!(decoder.read_unsigned_varint().unwrap(), 2);
704        assert_eq!(decoder.read_i32().unwrap(), 1);
705        assert_eq!(decoder.read_i32().unwrap(), 3);
706        assert_eq!(decoder.read_i32().unwrap(), 4);
707        assert_eq!(decoder.read_i64().unwrap(), 5);
708        assert_eq!(decoder.read_i32().unwrap(), 6);
709        assert_eq!(decoder.read_unsigned_varint().unwrap(), 2);
710        assert_eq!(decoder.read_i64().unwrap(), 10);
711        assert_eq!(decoder.read_i64().unwrap(), 12);
712        assert_eq!(decoder.read_i8().unwrap(), 2);
713        assert_eq!(decoder.read_i16().unwrap(), 3);
714        assert!(decoder.read_tagged_fields().unwrap().is_empty());
715        assert!(decoder.read_tagged_fields().unwrap().is_empty());
716        assert!(decoder.read_tagged_fields().unwrap().is_empty());
717        assert!(decoder.read_tagged_fields().unwrap().is_empty());
718        assert!(decoder.is_empty());
719    }
720
721    #[test]
722    fn decodes_read_and_summary_results_by_version() -> crate::error::Result<()> {
723        let mut body = Encoder::new();
724        body.write_compact_array(Some(&[()]), |encoder, ()| {
725            encoder.write_uuid(&[3; 16]);
726            encoder.write_compact_array(Some(&[()]), |encoder, ()| {
727                encoder.write_i32(1);
728                encoder.write_i16(0);
729                encoder.write_compact_nullable_string(None)?;
730                encoder.write_i32(2);
731                encoder.write_i64(10);
732                encoder.write_compact_array(Some(&[()]), |encoder, ()| {
733                    encoder.write_i64(10);
734                    encoder.write_i64(12);
735                    encoder.write_i8(0);
736                    encoder.write_i16(1);
737                    encoder.write_empty_tagged_fields();
738                    Ok(())
739                })?;
740                encoder.write_empty_tagged_fields();
741                Ok(())
742            })?;
743            encoder.write_empty_tagged_fields();
744            Ok(())
745        })?;
746        body.write_empty_tagged_fields();
747        let encoded = body.into_bytes();
748        let mut decoder = Decoder::new(&encoded);
749        let response = ReadShareGroupStateResponseV0::decode_body(&mut decoder)?;
750        assert_eq!(
751            response.results[0].partitions[0].state_batches[0],
752            ShareGroupStateBatch {
753                first_offset: 10,
754                last_offset: 12,
755                delivery_state: 0,
756                delivery_count: 1,
757            }
758        );
759        assert!(decoder.is_empty());
760
761        let mut summary = Encoder::new();
762        summary.write_compact_array(Some(&[()]), |encoder, ()| {
763            encoder.write_uuid(&[4; 16]);
764            encoder.write_compact_array(Some(&[()]), |encoder, ()| {
765                encoder.write_i32(1);
766                encoder.write_i16(0);
767                encoder.write_compact_nullable_string(None)?;
768                encoder.write_i32(2);
769                encoder.write_i32(3);
770                encoder.write_i64(10);
771                encoder.write_i32(9);
772                encoder.write_empty_tagged_fields();
773                Ok(())
774            })?;
775            encoder.write_empty_tagged_fields();
776            Ok(())
777        })?;
778        summary.write_empty_tagged_fields();
779        let encoded = summary.into_bytes();
780        let mut decoder = Decoder::new(&encoded);
781        let response = ReadShareGroupStateSummaryResponseV1::decode_body(&mut decoder)?;
782        assert_eq!(
783            response.results[0].partitions[0].delivery_complete_count,
784            Some(9)
785        );
786        assert!(decoder.is_empty());
787        Ok(())
788    }
789
790    #[test]
791    fn encodes_delete_share_group_state_v0() {
792        let request = DeleteShareGroupStateRequestV0 {
793            correlation_id: 7,
794            client_id: Some("kafrust".to_owned()),
795            group_id: "share-orders".to_owned(),
796            topics: vec![DeleteShareGroupStateTopic {
797                topic_id: [5; 16],
798                partitions: vec![0, 2],
799            }],
800        };
801        let encoded = request.encode().unwrap();
802        let mut decoder = Decoder::new(&encoded);
803        read_header(&mut decoder, DELETE_API_KEY, 0);
804        assert_eq!(decoder.read_compact_string().unwrap(), "share-orders");
805        assert_eq!(decoder.read_unsigned_varint().unwrap(), 2);
806        assert_eq!(decoder.read_uuid().unwrap(), [5; 16]);
807        assert_eq!(decoder.read_unsigned_varint().unwrap(), 3);
808        assert_eq!(decoder.read_i32().unwrap(), 0);
809        assert!(decoder.read_tagged_fields().unwrap().is_empty());
810        assert_eq!(decoder.read_i32().unwrap(), 2);
811        assert!(decoder.read_tagged_fields().unwrap().is_empty());
812        assert!(decoder.read_tagged_fields().unwrap().is_empty());
813        assert!(decoder.read_tagged_fields().unwrap().is_empty());
814        assert!(decoder.is_empty());
815    }
816}
kafrust_protocol/api/share_group_state.rs

kafrust_protocol/api/
share_group_state.rs
