kafka_protocol/messages/
share_fetch_request.rs

1//! ShareFetchRequest
2//!
3//! See the schema for this message [here](https://github.com/apache/kafka/blob/trunk/clients/src/main/resources/common/message/ShareFetchRequest.json).
4// WARNING: the items of this module are generated and should not be edited directly
5#![allow(unused)]
6
7use std::borrow::Borrow;
8use std::collections::BTreeMap;
9
10use anyhow::{bail, Result};
11use bytes::Bytes;
12use uuid::Uuid;
13
14use crate::protocol::{
15    buf::{ByteBuf, ByteBufMut},
16    compute_unknown_tagged_fields_size, types, write_unknown_tagged_fields, Decodable, Decoder,
17    Encodable, Encoder, HeaderVersion, Message, StrBytes, VersionRange,
18};
19
20/// Valid versions: 1
21#[non_exhaustive]
22#[derive(Debug, Clone, PartialEq)]
23pub struct AcknowledgementBatch {
24    /// First offset of batch of records to acknowledge.
25    ///
26    /// Supported API versions: 1
27    pub first_offset: i64,
28
29    /// Last offset (inclusive) of batch of records to acknowledge.
30    ///
31    /// Supported API versions: 1
32    pub last_offset: i64,
33
34    /// Array of acknowledge types - 0:Gap,1:Accept,2:Release,3:Reject.
35    ///
36    /// Supported API versions: 1
37    pub acknowledge_types: Vec<i8>,
38
39    /// Other tagged fields
40    pub unknown_tagged_fields: BTreeMap<i32, Bytes>,
41}
42
43impl AcknowledgementBatch {
44    /// Sets `first_offset` to the passed value.
45    ///
46    /// First offset of batch of records to acknowledge.
47    ///
48    /// Supported API versions: 1
49    pub fn with_first_offset(mut self, value: i64) -> Self {
50        self.first_offset = value;
51        self
52    }
53    /// Sets `last_offset` to the passed value.
54    ///
55    /// Last offset (inclusive) of batch of records to acknowledge.
56    ///
57    /// Supported API versions: 1
58    pub fn with_last_offset(mut self, value: i64) -> Self {
59        self.last_offset = value;
60        self
61    }
62    /// Sets `acknowledge_types` to the passed value.
63    ///
64    /// Array of acknowledge types - 0:Gap,1:Accept,2:Release,3:Reject.
65    ///
66    /// Supported API versions: 1
67    pub fn with_acknowledge_types(mut self, value: Vec<i8>) -> Self {
68        self.acknowledge_types = value;
69        self
70    }
71    /// Sets unknown_tagged_fields to the passed value.
72    pub fn with_unknown_tagged_fields(mut self, value: BTreeMap<i32, Bytes>) -> Self {
73        self.unknown_tagged_fields = value;
74        self
75    }
76    /// Inserts an entry into unknown_tagged_fields.
77    pub fn with_unknown_tagged_field(mut self, key: i32, value: Bytes) -> Self {
78        self.unknown_tagged_fields.insert(key, value);
79        self
80    }
81}
82
83#[cfg(feature = "client")]
84impl Encodable for AcknowledgementBatch {
85    fn encode<B: ByteBufMut>(&self, buf: &mut B, version: i16) -> Result<()> {
86        if version != 1 {
87            bail!("specified version not supported by this message type");
88        }
89        types::Int64.encode(buf, &self.first_offset)?;
90        types::Int64.encode(buf, &self.last_offset)?;
91        types::CompactArray(types::Int8).encode(buf, &self.acknowledge_types)?;
92        let num_tagged_fields = self.unknown_tagged_fields.len();
93        if num_tagged_fields > std::u32::MAX as usize {
94            bail!(
95                "Too many tagged fields to encode ({} fields)",
96                num_tagged_fields
97            );
98        }
99        types::UnsignedVarInt.encode(buf, num_tagged_fields as u32)?;
100
101        write_unknown_tagged_fields(buf, 0.., &self.unknown_tagged_fields)?;
102        Ok(())
103    }
104    fn compute_size(&self, version: i16) -> Result<usize> {
105        let mut total_size = 0;
106        total_size += types::Int64.compute_size(&self.first_offset)?;
107        total_size += types::Int64.compute_size(&self.last_offset)?;
108        total_size += types::CompactArray(types::Int8).compute_size(&self.acknowledge_types)?;
109        let num_tagged_fields = self.unknown_tagged_fields.len();
110        if num_tagged_fields > std::u32::MAX as usize {
111            bail!(
112                "Too many tagged fields to encode ({} fields)",
113                num_tagged_fields
114            );
115        }
116        total_size += types::UnsignedVarInt.compute_size(num_tagged_fields as u32)?;
117
118        total_size += compute_unknown_tagged_fields_size(&self.unknown_tagged_fields)?;
119        Ok(total_size)
120    }
121}
122
123#[cfg(feature = "broker")]
124impl Decodable for AcknowledgementBatch {
125    fn decode<B: ByteBuf>(buf: &mut B, version: i16) -> Result<Self> {
126        if version != 1 {
127            bail!("specified version not supported by this message type");
128        }
129        let first_offset = types::Int64.decode(buf)?;
130        let last_offset = types::Int64.decode(buf)?;
131        let acknowledge_types = types::CompactArray(types::Int8).decode(buf)?;
132        let mut unknown_tagged_fields = BTreeMap::new();
133        let num_tagged_fields = types::UnsignedVarInt.decode(buf)?;
134        for _ in 0..num_tagged_fields {
135            let tag: u32 = types::UnsignedVarInt.decode(buf)?;
136            let size: u32 = types::UnsignedVarInt.decode(buf)?;
137            let unknown_value = buf.try_get_bytes(size as usize)?;
138            unknown_tagged_fields.insert(tag as i32, unknown_value);
139        }
140        Ok(Self {
141            first_offset,
142            last_offset,
143            acknowledge_types,
144            unknown_tagged_fields,
145        })
146    }
147}
148
149impl Default for AcknowledgementBatch {
150    fn default() -> Self {
151        Self {
152            first_offset: 0,
153            last_offset: 0,
154            acknowledge_types: Default::default(),
155            unknown_tagged_fields: BTreeMap::new(),
156        }
157    }
158}
159
160impl Message for AcknowledgementBatch {
161    const VERSIONS: VersionRange = VersionRange { min: 1, max: 1 };
162    const DEPRECATED_VERSIONS: Option<VersionRange> = None;
163}
164
165/// Valid versions: 1
166#[non_exhaustive]
167#[derive(Debug, Clone, PartialEq)]
168pub struct FetchPartition {
169    /// The partition index.
170    ///
171    /// Supported API versions: 1
172    pub partition_index: i32,
173
174    /// The maximum bytes to fetch from this partition. 0 when only acknowledgement with no fetching is required. See KIP-74 for cases where this limit may not be honored.
175    ///
176    /// Supported API versions: none
177    pub partition_max_bytes: i32,
178
179    /// Record batches to acknowledge.
180    ///
181    /// Supported API versions: 1
182    pub acknowledgement_batches: Vec<AcknowledgementBatch>,
183
184    /// Other tagged fields
185    pub unknown_tagged_fields: BTreeMap<i32, Bytes>,
186}
187
188impl FetchPartition {
189    /// Sets `partition_index` to the passed value.
190    ///
191    /// The partition index.
192    ///
193    /// Supported API versions: 1
194    pub fn with_partition_index(mut self, value: i32) -> Self {
195        self.partition_index = value;
196        self
197    }
198    /// Sets `partition_max_bytes` to the passed value.
199    ///
200    /// The maximum bytes to fetch from this partition. 0 when only acknowledgement with no fetching is required. See KIP-74 for cases where this limit may not be honored.
201    ///
202    /// Supported API versions: none
203    pub fn with_partition_max_bytes(mut self, value: i32) -> Self {
204        self.partition_max_bytes = value;
205        self
206    }
207    /// Sets `acknowledgement_batches` to the passed value.
208    ///
209    /// Record batches to acknowledge.
210    ///
211    /// Supported API versions: 1
212    pub fn with_acknowledgement_batches(mut self, value: Vec<AcknowledgementBatch>) -> Self {
213        self.acknowledgement_batches = value;
214        self
215    }
216    /// Sets unknown_tagged_fields to the passed value.
217    pub fn with_unknown_tagged_fields(mut self, value: BTreeMap<i32, Bytes>) -> Self {
218        self.unknown_tagged_fields = value;
219        self
220    }
221    /// Inserts an entry into unknown_tagged_fields.
222    pub fn with_unknown_tagged_field(mut self, key: i32, value: Bytes) -> Self {
223        self.unknown_tagged_fields.insert(key, value);
224        self
225    }
226}
227
228#[cfg(feature = "client")]
229impl Encodable for FetchPartition {
230    fn encode<B: ByteBufMut>(&self, buf: &mut B, version: i16) -> Result<()> {
231        if version != 1 {
232            bail!("specified version not supported by this message type");
233        }
234        types::Int32.encode(buf, &self.partition_index)?;
235        types::CompactArray(types::Struct { version })
236            .encode(buf, &self.acknowledgement_batches)?;
237        let num_tagged_fields = self.unknown_tagged_fields.len();
238        if num_tagged_fields > std::u32::MAX as usize {
239            bail!(
240                "Too many tagged fields to encode ({} fields)",
241                num_tagged_fields
242            );
243        }
244        types::UnsignedVarInt.encode(buf, num_tagged_fields as u32)?;
245
246        write_unknown_tagged_fields(buf, 0.., &self.unknown_tagged_fields)?;
247        Ok(())
248    }
249    fn compute_size(&self, version: i16) -> Result<usize> {
250        let mut total_size = 0;
251        total_size += types::Int32.compute_size(&self.partition_index)?;
252        total_size += types::CompactArray(types::Struct { version })
253            .compute_size(&self.acknowledgement_batches)?;
254        let num_tagged_fields = self.unknown_tagged_fields.len();
255        if num_tagged_fields > std::u32::MAX as usize {
256            bail!(
257                "Too many tagged fields to encode ({} fields)",
258                num_tagged_fields
259            );
260        }
261        total_size += types::UnsignedVarInt.compute_size(num_tagged_fields as u32)?;
262
263        total_size += compute_unknown_tagged_fields_size(&self.unknown_tagged_fields)?;
264        Ok(total_size)
265    }
266}
267
268#[cfg(feature = "broker")]
269impl Decodable for FetchPartition {
270    fn decode<B: ByteBuf>(buf: &mut B, version: i16) -> Result<Self> {
271        if version != 1 {
272            bail!("specified version not supported by this message type");
273        }
274        let partition_index = types::Int32.decode(buf)?;
275        let partition_max_bytes = 0;
276        let acknowledgement_batches = types::CompactArray(types::Struct { version }).decode(buf)?;
277        let mut unknown_tagged_fields = BTreeMap::new();
278        let num_tagged_fields = types::UnsignedVarInt.decode(buf)?;
279        for _ in 0..num_tagged_fields {
280            let tag: u32 = types::UnsignedVarInt.decode(buf)?;
281            let size: u32 = types::UnsignedVarInt.decode(buf)?;
282            let unknown_value = buf.try_get_bytes(size as usize)?;
283            unknown_tagged_fields.insert(tag as i32, unknown_value);
284        }
285        Ok(Self {
286            partition_index,
287            partition_max_bytes,
288            acknowledgement_batches,
289            unknown_tagged_fields,
290        })
291    }
292}
293
294impl Default for FetchPartition {
295    fn default() -> Self {
296        Self {
297            partition_index: 0,
298            partition_max_bytes: 0,
299            acknowledgement_batches: Default::default(),
300            unknown_tagged_fields: BTreeMap::new(),
301        }
302    }
303}
304
305impl Message for FetchPartition {
306    const VERSIONS: VersionRange = VersionRange { min: 1, max: 1 };
307    const DEPRECATED_VERSIONS: Option<VersionRange> = None;
308}
309
310/// Valid versions: 1
311#[non_exhaustive]
312#[derive(Debug, Clone, PartialEq)]
313pub struct FetchTopic {
314    /// The unique topic ID.
315    ///
316    /// Supported API versions: 1
317    pub topic_id: Uuid,
318
319    /// The partitions to fetch.
320    ///
321    /// Supported API versions: 1
322    pub partitions: Vec<FetchPartition>,
323
324    /// Other tagged fields
325    pub unknown_tagged_fields: BTreeMap<i32, Bytes>,
326}
327
328impl FetchTopic {
329    /// Sets `topic_id` to the passed value.
330    ///
331    /// The unique topic ID.
332    ///
333    /// Supported API versions: 1
334    pub fn with_topic_id(mut self, value: Uuid) -> Self {
335        self.topic_id = value;
336        self
337    }
338    /// Sets `partitions` to the passed value.
339    ///
340    /// The partitions to fetch.
341    ///
342    /// Supported API versions: 1
343    pub fn with_partitions(mut self, value: Vec<FetchPartition>) -> Self {
344        self.partitions = value;
345        self
346    }
347    /// Sets unknown_tagged_fields to the passed value.
348    pub fn with_unknown_tagged_fields(mut self, value: BTreeMap<i32, Bytes>) -> Self {
349        self.unknown_tagged_fields = value;
350        self
351    }
352    /// Inserts an entry into unknown_tagged_fields.
353    pub fn with_unknown_tagged_field(mut self, key: i32, value: Bytes) -> Self {
354        self.unknown_tagged_fields.insert(key, value);
355        self
356    }
357}
358
359#[cfg(feature = "client")]
360impl Encodable for FetchTopic {
361    fn encode<B: ByteBufMut>(&self, buf: &mut B, version: i16) -> Result<()> {
362        if version != 1 {
363            bail!("specified version not supported by this message type");
364        }
365        types::Uuid.encode(buf, &self.topic_id)?;
366        types::CompactArray(types::Struct { version }).encode(buf, &self.partitions)?;
367        let num_tagged_fields = self.unknown_tagged_fields.len();
368        if num_tagged_fields > std::u32::MAX as usize {
369            bail!(
370                "Too many tagged fields to encode ({} fields)",
371                num_tagged_fields
372            );
373        }
374        types::UnsignedVarInt.encode(buf, num_tagged_fields as u32)?;
375
376        write_unknown_tagged_fields(buf, 0.., &self.unknown_tagged_fields)?;
377        Ok(())
378    }
379    fn compute_size(&self, version: i16) -> Result<usize> {
380        let mut total_size = 0;
381        total_size += types::Uuid.compute_size(&self.topic_id)?;
382        total_size +=
383            types::CompactArray(types::Struct { version }).compute_size(&self.partitions)?;
384        let num_tagged_fields = self.unknown_tagged_fields.len();
385        if num_tagged_fields > std::u32::MAX as usize {
386            bail!(
387                "Too many tagged fields to encode ({} fields)",
388                num_tagged_fields
389            );
390        }
391        total_size += types::UnsignedVarInt.compute_size(num_tagged_fields as u32)?;
392
393        total_size += compute_unknown_tagged_fields_size(&self.unknown_tagged_fields)?;
394        Ok(total_size)
395    }
396}
397
398#[cfg(feature = "broker")]
399impl Decodable for FetchTopic {
400    fn decode<B: ByteBuf>(buf: &mut B, version: i16) -> Result<Self> {
401        if version != 1 {
402            bail!("specified version not supported by this message type");
403        }
404        let topic_id = types::Uuid.decode(buf)?;
405        let partitions = types::CompactArray(types::Struct { version }).decode(buf)?;
406        let mut unknown_tagged_fields = BTreeMap::new();
407        let num_tagged_fields = types::UnsignedVarInt.decode(buf)?;
408        for _ in 0..num_tagged_fields {
409            let tag: u32 = types::UnsignedVarInt.decode(buf)?;
410            let size: u32 = types::UnsignedVarInt.decode(buf)?;
411            let unknown_value = buf.try_get_bytes(size as usize)?;
412            unknown_tagged_fields.insert(tag as i32, unknown_value);
413        }
414        Ok(Self {
415            topic_id,
416            partitions,
417            unknown_tagged_fields,
418        })
419    }
420}
421
422impl Default for FetchTopic {
423    fn default() -> Self {
424        Self {
425            topic_id: Uuid::nil(),
426            partitions: Default::default(),
427            unknown_tagged_fields: BTreeMap::new(),
428        }
429    }
430}
431
432impl Message for FetchTopic {
433    const VERSIONS: VersionRange = VersionRange { min: 1, max: 1 };
434    const DEPRECATED_VERSIONS: Option<VersionRange> = None;
435}
436
437/// Valid versions: 1
438#[non_exhaustive]
439#[derive(Debug, Clone, PartialEq)]
440pub struct ForgottenTopic {
441    /// The unique topic ID.
442    ///
443    /// Supported API versions: 1
444    pub topic_id: Uuid,
445
446    /// The partitions indexes to forget.
447    ///
448    /// Supported API versions: 1
449    pub partitions: Vec<i32>,
450
451    /// Other tagged fields
452    pub unknown_tagged_fields: BTreeMap<i32, Bytes>,
453}
454
455impl ForgottenTopic {
456    /// Sets `topic_id` to the passed value.
457    ///
458    /// The unique topic ID.
459    ///
460    /// Supported API versions: 1
461    pub fn with_topic_id(mut self, value: Uuid) -> Self {
462        self.topic_id = value;
463        self
464    }
465    /// Sets `partitions` to the passed value.
466    ///
467    /// The partitions indexes to forget.
468    ///
469    /// Supported API versions: 1
470    pub fn with_partitions(mut self, value: Vec<i32>) -> Self {
471        self.partitions = value;
472        self
473    }
474    /// Sets unknown_tagged_fields to the passed value.
475    pub fn with_unknown_tagged_fields(mut self, value: BTreeMap<i32, Bytes>) -> Self {
476        self.unknown_tagged_fields = value;
477        self
478    }
479    /// Inserts an entry into unknown_tagged_fields.
480    pub fn with_unknown_tagged_field(mut self, key: i32, value: Bytes) -> Self {
481        self.unknown_tagged_fields.insert(key, value);
482        self
483    }
484}
485
486#[cfg(feature = "client")]
487impl Encodable for ForgottenTopic {
488    fn encode<B: ByteBufMut>(&self, buf: &mut B, version: i16) -> Result<()> {
489        if version != 1 {
490            bail!("specified version not supported by this message type");
491        }
492        types::Uuid.encode(buf, &self.topic_id)?;
493        types::CompactArray(types::Int32).encode(buf, &self.partitions)?;
494        let num_tagged_fields = self.unknown_tagged_fields.len();
495        if num_tagged_fields > std::u32::MAX as usize {
496            bail!(
497                "Too many tagged fields to encode ({} fields)",
498                num_tagged_fields
499            );
500        }
501        types::UnsignedVarInt.encode(buf, num_tagged_fields as u32)?;
502
503        write_unknown_tagged_fields(buf, 0.., &self.unknown_tagged_fields)?;
504        Ok(())
505    }
506    fn compute_size(&self, version: i16) -> Result<usize> {
507        let mut total_size = 0;
508        total_size += types::Uuid.compute_size(&self.topic_id)?;
509        total_size += types::CompactArray(types::Int32).compute_size(&self.partitions)?;
510        let num_tagged_fields = self.unknown_tagged_fields.len();
511        if num_tagged_fields > std::u32::MAX as usize {
512            bail!(
513                "Too many tagged fields to encode ({} fields)",
514                num_tagged_fields
515            );
516        }
517        total_size += types::UnsignedVarInt.compute_size(num_tagged_fields as u32)?;
518
519        total_size += compute_unknown_tagged_fields_size(&self.unknown_tagged_fields)?;
520        Ok(total_size)
521    }
522}
523
524#[cfg(feature = "broker")]
525impl Decodable for ForgottenTopic {
526    fn decode<B: ByteBuf>(buf: &mut B, version: i16) -> Result<Self> {
527        if version != 1 {
528            bail!("specified version not supported by this message type");
529        }
530        let topic_id = types::Uuid.decode(buf)?;
531        let partitions = types::CompactArray(types::Int32).decode(buf)?;
532        let mut unknown_tagged_fields = BTreeMap::new();
533        let num_tagged_fields = types::UnsignedVarInt.decode(buf)?;
534        for _ in 0..num_tagged_fields {
535            let tag: u32 = types::UnsignedVarInt.decode(buf)?;
536            let size: u32 = types::UnsignedVarInt.decode(buf)?;
537            let unknown_value = buf.try_get_bytes(size as usize)?;
538            unknown_tagged_fields.insert(tag as i32, unknown_value);
539        }
540        Ok(Self {
541            topic_id,
542            partitions,
543            unknown_tagged_fields,
544        })
545    }
546}
547
548impl Default for ForgottenTopic {
549    fn default() -> Self {
550        Self {
551            topic_id: Uuid::nil(),
552            partitions: Default::default(),
553            unknown_tagged_fields: BTreeMap::new(),
554        }
555    }
556}
557
558impl Message for ForgottenTopic {
559    const VERSIONS: VersionRange = VersionRange { min: 1, max: 1 };
560    const DEPRECATED_VERSIONS: Option<VersionRange> = None;
561}
562
563/// Valid versions: 1
564#[non_exhaustive]
565#[derive(Debug, Clone, PartialEq)]
566pub struct ShareFetchRequest {
567    /// The group identifier.
568    ///
569    /// Supported API versions: 1
570    pub group_id: Option<super::GroupId>,
571
572    /// The member ID.
573    ///
574    /// Supported API versions: 1
575    pub member_id: Option<StrBytes>,
576
577    /// The current share session epoch: 0 to open a share session; -1 to close it; otherwise increments for consecutive requests.
578    ///
579    /// Supported API versions: 1
580    pub share_session_epoch: i32,
581
582    /// The maximum time in milliseconds to wait for the response.
583    ///
584    /// Supported API versions: 1
585    pub max_wait_ms: i32,
586
587    /// The minimum bytes to accumulate in the response.
588    ///
589    /// Supported API versions: 1
590    pub min_bytes: i32,
591
592    /// The maximum bytes to fetch. See KIP-74 for cases where this limit may not be honored.
593    ///
594    /// Supported API versions: 1
595    pub max_bytes: i32,
596
597    /// The maximum number of records to fetch. This limit can be exceeded for alignment of batch boundaries.
598    ///
599    /// Supported API versions: 1
600    pub max_records: i32,
601
602    /// The optimal number of records for batches of acquired records and acknowledgements.
603    ///
604    /// Supported API versions: 1
605    pub batch_size: i32,
606
607    /// The topics to fetch.
608    ///
609    /// Supported API versions: 1
610    pub topics: Vec<FetchTopic>,
611
612    /// The partitions to remove from this share session.
613    ///
614    /// Supported API versions: 1
615    pub forgotten_topics_data: Vec<ForgottenTopic>,
616
617    /// Other tagged fields
618    pub unknown_tagged_fields: BTreeMap<i32, Bytes>,
619}
620
621impl ShareFetchRequest {
622    /// Sets `group_id` to the passed value.
623    ///
624    /// The group identifier.
625    ///
626    /// Supported API versions: 1
627    pub fn with_group_id(mut self, value: Option<super::GroupId>) -> Self {
628        self.group_id = value;
629        self
630    }
631    /// Sets `member_id` to the passed value.
632    ///
633    /// The member ID.
634    ///
635    /// Supported API versions: 1
636    pub fn with_member_id(mut self, value: Option<StrBytes>) -> Self {
637        self.member_id = value;
638        self
639    }
640    /// Sets `share_session_epoch` to the passed value.
641    ///
642    /// The current share session epoch: 0 to open a share session; -1 to close it; otherwise increments for consecutive requests.
643    ///
644    /// Supported API versions: 1
645    pub fn with_share_session_epoch(mut self, value: i32) -> Self {
646        self.share_session_epoch = value;
647        self
648    }
649    /// Sets `max_wait_ms` to the passed value.
650    ///
651    /// The maximum time in milliseconds to wait for the response.
652    ///
653    /// Supported API versions: 1
654    pub fn with_max_wait_ms(mut self, value: i32) -> Self {
655        self.max_wait_ms = value;
656        self
657    }
658    /// Sets `min_bytes` to the passed value.
659    ///
660    /// The minimum bytes to accumulate in the response.
661    ///
662    /// Supported API versions: 1
663    pub fn with_min_bytes(mut self, value: i32) -> Self {
664        self.min_bytes = value;
665        self
666    }
667    /// Sets `max_bytes` to the passed value.
668    ///
669    /// The maximum bytes to fetch. See KIP-74 for cases where this limit may not be honored.
670    ///
671    /// Supported API versions: 1
672    pub fn with_max_bytes(mut self, value: i32) -> Self {
673        self.max_bytes = value;
674        self
675    }
676    /// Sets `max_records` to the passed value.
677    ///
678    /// The maximum number of records to fetch. This limit can be exceeded for alignment of batch boundaries.
679    ///
680    /// Supported API versions: 1
681    pub fn with_max_records(mut self, value: i32) -> Self {
682        self.max_records = value;
683        self
684    }
685    /// Sets `batch_size` to the passed value.
686    ///
687    /// The optimal number of records for batches of acquired records and acknowledgements.
688    ///
689    /// Supported API versions: 1
690    pub fn with_batch_size(mut self, value: i32) -> Self {
691        self.batch_size = value;
692        self
693    }
694    /// Sets `topics` to the passed value.
695    ///
696    /// The topics to fetch.
697    ///
698    /// Supported API versions: 1
699    pub fn with_topics(mut self, value: Vec<FetchTopic>) -> Self {
700        self.topics = value;
701        self
702    }
703    /// Sets `forgotten_topics_data` to the passed value.
704    ///
705    /// The partitions to remove from this share session.
706    ///
707    /// Supported API versions: 1
708    pub fn with_forgotten_topics_data(mut self, value: Vec<ForgottenTopic>) -> Self {
709        self.forgotten_topics_data = value;
710        self
711    }
712    /// Sets unknown_tagged_fields to the passed value.
713    pub fn with_unknown_tagged_fields(mut self, value: BTreeMap<i32, Bytes>) -> Self {
714        self.unknown_tagged_fields = value;
715        self
716    }
717    /// Inserts an entry into unknown_tagged_fields.
718    pub fn with_unknown_tagged_field(mut self, key: i32, value: Bytes) -> Self {
719        self.unknown_tagged_fields.insert(key, value);
720        self
721    }
722}
723
724#[cfg(feature = "client")]
725impl Encodable for ShareFetchRequest {
726    fn encode<B: ByteBufMut>(&self, buf: &mut B, version: i16) -> Result<()> {
727        if version != 1 {
728            bail!("specified version not supported by this message type");
729        }
730        types::CompactString.encode(buf, &self.group_id)?;
731        types::CompactString.encode(buf, &self.member_id)?;
732        types::Int32.encode(buf, &self.share_session_epoch)?;
733        types::Int32.encode(buf, &self.max_wait_ms)?;
734        types::Int32.encode(buf, &self.min_bytes)?;
735        types::Int32.encode(buf, &self.max_bytes)?;
736        types::Int32.encode(buf, &self.max_records)?;
737        types::Int32.encode(buf, &self.batch_size)?;
738        types::CompactArray(types::Struct { version }).encode(buf, &self.topics)?;
739        types::CompactArray(types::Struct { version }).encode(buf, &self.forgotten_topics_data)?;
740        let num_tagged_fields = self.unknown_tagged_fields.len();
741        if num_tagged_fields > std::u32::MAX as usize {
742            bail!(
743                "Too many tagged fields to encode ({} fields)",
744                num_tagged_fields
745            );
746        }
747        types::UnsignedVarInt.encode(buf, num_tagged_fields as u32)?;
748
749        write_unknown_tagged_fields(buf, 0.., &self.unknown_tagged_fields)?;
750        Ok(())
751    }
752    fn compute_size(&self, version: i16) -> Result<usize> {
753        let mut total_size = 0;
754        total_size += types::CompactString.compute_size(&self.group_id)?;
755        total_size += types::CompactString.compute_size(&self.member_id)?;
756        total_size += types::Int32.compute_size(&self.share_session_epoch)?;
757        total_size += types::Int32.compute_size(&self.max_wait_ms)?;
758        total_size += types::Int32.compute_size(&self.min_bytes)?;
759        total_size += types::Int32.compute_size(&self.max_bytes)?;
760        total_size += types::Int32.compute_size(&self.max_records)?;
761        total_size += types::Int32.compute_size(&self.batch_size)?;
762        total_size += types::CompactArray(types::Struct { version }).compute_size(&self.topics)?;
763        total_size += types::CompactArray(types::Struct { version })
764            .compute_size(&self.forgotten_topics_data)?;
765        let num_tagged_fields = self.unknown_tagged_fields.len();
766        if num_tagged_fields > std::u32::MAX as usize {
767            bail!(
768                "Too many tagged fields to encode ({} fields)",
769                num_tagged_fields
770            );
771        }
772        total_size += types::UnsignedVarInt.compute_size(num_tagged_fields as u32)?;
773
774        total_size += compute_unknown_tagged_fields_size(&self.unknown_tagged_fields)?;
775        Ok(total_size)
776    }
777}
778
779#[cfg(feature = "broker")]
780impl Decodable for ShareFetchRequest {
781    fn decode<B: ByteBuf>(buf: &mut B, version: i16) -> Result<Self> {
782        if version != 1 {
783            bail!("specified version not supported by this message type");
784        }
785        let group_id = types::CompactString.decode(buf)?;
786        let member_id = types::CompactString.decode(buf)?;
787        let share_session_epoch = types::Int32.decode(buf)?;
788        let max_wait_ms = types::Int32.decode(buf)?;
789        let min_bytes = types::Int32.decode(buf)?;
790        let max_bytes = types::Int32.decode(buf)?;
791        let max_records = types::Int32.decode(buf)?;
792        let batch_size = types::Int32.decode(buf)?;
793        let topics = types::CompactArray(types::Struct { version }).decode(buf)?;
794        let forgotten_topics_data = types::CompactArray(types::Struct { version }).decode(buf)?;
795        let mut unknown_tagged_fields = BTreeMap::new();
796        let num_tagged_fields = types::UnsignedVarInt.decode(buf)?;
797        for _ in 0..num_tagged_fields {
798            let tag: u32 = types::UnsignedVarInt.decode(buf)?;
799            let size: u32 = types::UnsignedVarInt.decode(buf)?;
800            let unknown_value = buf.try_get_bytes(size as usize)?;
801            unknown_tagged_fields.insert(tag as i32, unknown_value);
802        }
803        Ok(Self {
804            group_id,
805            member_id,
806            share_session_epoch,
807            max_wait_ms,
808            min_bytes,
809            max_bytes,
810            max_records,
811            batch_size,
812            topics,
813            forgotten_topics_data,
814            unknown_tagged_fields,
815        })
816    }
817}
818
819impl Default for ShareFetchRequest {
820    fn default() -> Self {
821        Self {
822            group_id: None,
823            member_id: Some(Default::default()),
824            share_session_epoch: 0,
825            max_wait_ms: 0,
826            min_bytes: 0,
827            max_bytes: 0x7fffffff,
828            max_records: 0,
829            batch_size: 0,
830            topics: Default::default(),
831            forgotten_topics_data: Default::default(),
832            unknown_tagged_fields: BTreeMap::new(),
833        }
834    }
835}
836
837impl Message for ShareFetchRequest {
838    const VERSIONS: VersionRange = VersionRange { min: 1, max: 1 };
839    const DEPRECATED_VERSIONS: Option<VersionRange> = None;
840}
841
842impl HeaderVersion for ShareFetchRequest {
843    fn header_version(version: i16) -> i16 {
844        2
845    }
846}
kafka_protocol/messages/share_fetch_request.rs

kafka_protocol/messages/
share_fetch_request.rs
