kafrust_protocol/api/
init_producer_id.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 22;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct InitProducerIdRequestV0 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub transactional_id: Option<String>,
12    pub transaction_timeout_ms: i32,
13}
14
15impl InitProducerIdRequestV0 {
16    pub fn encode(&self) -> Result<Vec<u8>> {
17        let mut encoder = Encoder::new();
18        RequestHeader {
19            api_key: API_KEY,
20            api_version: 0,
21            correlation_id: self.correlation_id,
22            client_id: self.client_id.clone(),
23        }
24        .encode_v1(&mut encoder)?;
25        encoder.write_nullable_string(self.transactional_id.as_deref())?;
26        encoder.write_i32(self.transaction_timeout_ms);
27        Ok(encoder.into_bytes())
28    }
29}
30
31#[derive(Debug, Clone, PartialEq, Eq)]
32pub struct InitProducerIdResponseV0 {
33    pub throttle_time_ms: i32,
34    pub error_code: i16,
35    pub producer_id: i64,
36    pub producer_epoch: i16,
37}
38
39impl InitProducerIdResponseV0 {
40    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
41        Ok(Self {
42            throttle_time_ms: decoder.read_i32()?,
43            error_code: decoder.read_i16()?,
44            producer_id: decoder.read_i64()?,
45            producer_epoch: decoder.read_i16()?,
46        })
47    }
48}
49
50/// InitProducerId v2, the flexible form used by current Kafka brokers.
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct InitProducerIdRequestV2 {
53    pub correlation_id: i32,
54    pub client_id: Option<String>,
55    pub transactional_id: Option<String>,
56    pub transaction_timeout_ms: i32,
57}
58
59impl InitProducerIdRequestV2 {
60    pub fn encode(&self) -> Result<Vec<u8>> {
61        let mut encoder = Encoder::new();
62        RequestHeader {
63            api_key: API_KEY,
64            api_version: 2,
65            correlation_id: self.correlation_id,
66            client_id: self.client_id.clone(),
67        }
68        .encode_v2(&mut encoder)?;
69        encoder.write_compact_nullable_string(self.transactional_id.as_deref())?;
70        encoder.write_i32(self.transaction_timeout_ms);
71        encoder.write_empty_tagged_fields();
72        Ok(encoder.into_bytes())
73    }
74}
75
76/// InitProducerId v2 response with flexible tagged fields.
77#[derive(Debug, Clone, PartialEq, Eq)]
78pub struct InitProducerIdResponseV2 {
79    pub throttle_time_ms: i32,
80    pub error_code: i16,
81    pub producer_id: i64,
82    pub producer_epoch: i16,
83}
84
85impl InitProducerIdResponseV2 {
86    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
87        let response = Self {
88            throttle_time_ms: decoder.read_i32()?,
89            error_code: decoder.read_i16()?,
90            producer_id: decoder.read_i64()?,
91            producer_epoch: decoder.read_i16()?,
92        };
93        decoder.read_tagged_fields()?;
94        Ok(response)
95    }
96}
97
98#[cfg(test)]
99#[allow(clippy::unwrap_used)]
100mod tests {
101    use super::{
102        InitProducerIdRequestV0, InitProducerIdRequestV2, InitProducerIdResponseV0,
103        InitProducerIdResponseV2, API_KEY,
104    };
105    use crate::codec::Decoder;
106
107    #[test]
108    fn encodes_non_transactional_init_producer_id_v0_request() {
109        let request = InitProducerIdRequestV0 {
110            correlation_id: 23,
111            client_id: Some("kafrust".to_owned()),
112            transactional_id: None,
113            transaction_timeout_ms: 60_000,
114        };
115
116        assert_eq!(
117            request.encode().unwrap(),
118            [
119                0, 22, // api key
120                0, 0, // api version
121                0, 0, 0, 23, // correlation id
122                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client id
123                0xff, 0xff, // null transactional id
124                0, 0, 0xea, 0x60, // transaction timeout
125            ]
126        );
127        assert_eq!(API_KEY, 22);
128    }
129
130    #[test]
131    fn encodes_transactional_init_producer_id_v0_request() {
132        let request = InitProducerIdRequestV0 {
133            correlation_id: 24,
134            client_id: None,
135            transactional_id: Some("orders-tx".to_owned()),
136            transaction_timeout_ms: 30_000,
137        };
138
139        assert_eq!(
140            request.encode().unwrap(),
141            [
142                0, 22, // api key
143                0, 0, // api version
144                0, 0, 0, 24, // correlation id
145                0xff, 0xff, // null client id
146                0, 9, b'o', b'r', b'd', b'e', b'r', b's', b'-', b't', b'x', 0, 0, 0x75,
147                0x30, // transaction timeout
148            ]
149        );
150    }
151
152    #[test]
153    fn decodes_init_producer_id_v0_response() {
154        let bytes = [
155            0, 0, 0, 12, // throttle time
156            0, 0, // error code
157            0, 0, 0, 0, 0, 0, 0, 42, // producer id
158            0, 3, // producer epoch
159        ];
160        let mut decoder = Decoder::new(&bytes);
161        let response = InitProducerIdResponseV0::decode_body(&mut decoder).unwrap();
162
163        assert_eq!(response.throttle_time_ms, 12);
164        assert_eq!(response.error_code, 0);
165        assert_eq!(response.producer_id, 42);
166        assert_eq!(response.producer_epoch, 3);
167        assert!(decoder.is_empty());
168    }
169
170    #[test]
171    fn encodes_transactional_init_producer_id_v2_request_with_flexible_fields() {
172        let request = InitProducerIdRequestV2 {
173            correlation_id: 25,
174            client_id: Some("kafrust".to_owned()),
175            transactional_id: Some("orders-tx".to_owned()),
176            transaction_timeout_ms: 30_000,
177        };
178        let encoded = request.encode().unwrap();
179
180        assert_eq!(&encoded[0..8], &[0, 22, 0, 2, 0, 0, 0, 25]);
181        assert!(encoded
182            .windows(b"orders-tx".len())
183            .any(|window| window == b"orders-tx"));
184        assert_eq!(encoded.last(), Some(&0));
185    }
186
187    #[test]
188    fn decodes_init_producer_id_v2_response_with_tagged_fields() {
189        let bytes = [
190            0, 0, 0, 12, // throttle time
191            0, 0, // error code
192            0, 0, 0, 0, 0, 0, 0, 42, // producer id
193            0, 3, // producer epoch
194            0, // response tagged fields
195        ];
196        let mut decoder = Decoder::new(&bytes);
197        let response = InitProducerIdResponseV2::decode_body(&mut decoder).unwrap();
198
199        assert_eq!(response.throttle_time_ms, 12);
200        assert_eq!(response.error_code, 0);
201        assert_eq!(response.producer_id, 42);
202        assert_eq!(response.producer_epoch, 3);
203        assert!(decoder.is_empty());
204    }
205}
kafrust_protocol/api/init_producer_id.rs

kafrust_protocol/api/
init_producer_id.rs
