kafka_protocol/messages/
add_offsets_to_txn_request.rs

1//! AddOffsetsToTxnRequest
2//!
3//! See the schema for this message [here](https://github.com/apache/kafka/blob/trunk/clients/src/main/resources/common/message/AddOffsetsToTxnRequest.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: 0-4
21#[non_exhaustive]
22#[derive(Debug, Clone, PartialEq)]
23pub struct AddOffsetsToTxnRequest {
24    /// The transactional id corresponding to the transaction.
25    ///
26    /// Supported API versions: 0-4
27    pub transactional_id: super::TransactionalId,
28
29    /// Current producer id in use by the transactional id.
30    ///
31    /// Supported API versions: 0-4
32    pub producer_id: super::ProducerId,
33
34    /// Current epoch associated with the producer id.
35    ///
36    /// Supported API versions: 0-4
37    pub producer_epoch: i16,
38
39    /// The unique group identifier.
40    ///
41    /// Supported API versions: 0-4
42    pub group_id: super::GroupId,
43
44    /// Other tagged fields
45    pub unknown_tagged_fields: BTreeMap<i32, Bytes>,
46}
47
48impl AddOffsetsToTxnRequest {
49    /// Sets `transactional_id` to the passed value.
50    ///
51    /// The transactional id corresponding to the transaction.
52    ///
53    /// Supported API versions: 0-4
54    pub fn with_transactional_id(mut self, value: super::TransactionalId) -> Self {
55        self.transactional_id = value;
56        self
57    }
58    /// Sets `producer_id` to the passed value.
59    ///
60    /// Current producer id in use by the transactional id.
61    ///
62    /// Supported API versions: 0-4
63    pub fn with_producer_id(mut self, value: super::ProducerId) -> Self {
64        self.producer_id = value;
65        self
66    }
67    /// Sets `producer_epoch` to the passed value.
68    ///
69    /// Current epoch associated with the producer id.
70    ///
71    /// Supported API versions: 0-4
72    pub fn with_producer_epoch(mut self, value: i16) -> Self {
73        self.producer_epoch = value;
74        self
75    }
76    /// Sets `group_id` to the passed value.
77    ///
78    /// The unique group identifier.
79    ///
80    /// Supported API versions: 0-4
81    pub fn with_group_id(mut self, value: super::GroupId) -> Self {
82        self.group_id = value;
83        self
84    }
85    /// Sets unknown_tagged_fields to the passed value.
86    pub fn with_unknown_tagged_fields(mut self, value: BTreeMap<i32, Bytes>) -> Self {
87        self.unknown_tagged_fields = value;
88        self
89    }
90    /// Inserts an entry into unknown_tagged_fields.
91    pub fn with_unknown_tagged_field(mut self, key: i32, value: Bytes) -> Self {
92        self.unknown_tagged_fields.insert(key, value);
93        self
94    }
95}
96
97#[cfg(feature = "client")]
98impl Encodable for AddOffsetsToTxnRequest {
99    fn encode<B: ByteBufMut>(&self, buf: &mut B, version: i16) -> Result<()> {
100        if version < 0 || version > 4 {
101            bail!("specified version not supported by this message type");
102        }
103        if version >= 3 {
104            types::CompactString.encode(buf, &self.transactional_id)?;
105        } else {
106            types::String.encode(buf, &self.transactional_id)?;
107        }
108        types::Int64.encode(buf, &self.producer_id)?;
109        types::Int16.encode(buf, &self.producer_epoch)?;
110        if version >= 3 {
111            types::CompactString.encode(buf, &self.group_id)?;
112        } else {
113            types::String.encode(buf, &self.group_id)?;
114        }
115        if version >= 3 {
116            let num_tagged_fields = self.unknown_tagged_fields.len();
117            if num_tagged_fields > std::u32::MAX as usize {
118                bail!(
119                    "Too many tagged fields to encode ({} fields)",
120                    num_tagged_fields
121                );
122            }
123            types::UnsignedVarInt.encode(buf, num_tagged_fields as u32)?;
124
125            write_unknown_tagged_fields(buf, 0.., &self.unknown_tagged_fields)?;
126        }
127        Ok(())
128    }
129    fn compute_size(&self, version: i16) -> Result<usize> {
130        let mut total_size = 0;
131        if version >= 3 {
132            total_size += types::CompactString.compute_size(&self.transactional_id)?;
133        } else {
134            total_size += types::String.compute_size(&self.transactional_id)?;
135        }
136        total_size += types::Int64.compute_size(&self.producer_id)?;
137        total_size += types::Int16.compute_size(&self.producer_epoch)?;
138        if version >= 3 {
139            total_size += types::CompactString.compute_size(&self.group_id)?;
140        } else {
141            total_size += types::String.compute_size(&self.group_id)?;
142        }
143        if version >= 3 {
144            let num_tagged_fields = self.unknown_tagged_fields.len();
145            if num_tagged_fields > std::u32::MAX as usize {
146                bail!(
147                    "Too many tagged fields to encode ({} fields)",
148                    num_tagged_fields
149                );
150            }
151            total_size += types::UnsignedVarInt.compute_size(num_tagged_fields as u32)?;
152
153            total_size += compute_unknown_tagged_fields_size(&self.unknown_tagged_fields)?;
154        }
155        Ok(total_size)
156    }
157}
158
159#[cfg(feature = "broker")]
160impl Decodable for AddOffsetsToTxnRequest {
161    fn decode<B: ByteBuf>(buf: &mut B, version: i16) -> Result<Self> {
162        if version < 0 || version > 4 {
163            bail!("specified version not supported by this message type");
164        }
165        let transactional_id = if version >= 3 {
166            types::CompactString.decode(buf)?
167        } else {
168            types::String.decode(buf)?
169        };
170        let producer_id = types::Int64.decode(buf)?;
171        let producer_epoch = types::Int16.decode(buf)?;
172        let group_id = if version >= 3 {
173            types::CompactString.decode(buf)?
174        } else {
175            types::String.decode(buf)?
176        };
177        let mut unknown_tagged_fields = BTreeMap::new();
178        if version >= 3 {
179            let num_tagged_fields = types::UnsignedVarInt.decode(buf)?;
180            for _ in 0..num_tagged_fields {
181                let tag: u32 = types::UnsignedVarInt.decode(buf)?;
182                let size: u32 = types::UnsignedVarInt.decode(buf)?;
183                let unknown_value = buf.try_get_bytes(size as usize)?;
184                unknown_tagged_fields.insert(tag as i32, unknown_value);
185            }
186        }
187        Ok(Self {
188            transactional_id,
189            producer_id,
190            producer_epoch,
191            group_id,
192            unknown_tagged_fields,
193        })
194    }
195}
196
197impl Default for AddOffsetsToTxnRequest {
198    fn default() -> Self {
199        Self {
200            transactional_id: Default::default(),
201            producer_id: (0).into(),
202            producer_epoch: 0,
203            group_id: Default::default(),
204            unknown_tagged_fields: BTreeMap::new(),
205        }
206    }
207}
208
209impl Message for AddOffsetsToTxnRequest {
210    const VERSIONS: VersionRange = VersionRange { min: 0, max: 4 };
211    const DEPRECATED_VERSIONS: Option<VersionRange> = None;
212}
213
214impl HeaderVersion for AddOffsetsToTxnRequest {
215    fn header_version(version: i16) -> i16 {
216        if version >= 3 {
217            2
218        } else {
219            1
220        }
221    }
222}
kafka_protocol/messages/add_offsets_to_txn_request.rs

kafka_protocol/messages/
add_offsets_to_txn_request.rs
