kafrust_protocol/api/
add_partitions_to_txn.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 24;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct AddPartitionsToTxnRequestV0 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub transactional_id: String,
12    pub producer_id: i64,
13    pub producer_epoch: i16,
14    pub topics: Vec<AddPartitionsToTxnTopic>,
15}
16
17impl AddPartitionsToTxnRequestV0 {
18    pub fn encode(&self) -> Result<Vec<u8>> {
19        let mut encoder = Encoder::new();
20        RequestHeader {
21            api_key: API_KEY,
22            api_version: 0,
23            correlation_id: self.correlation_id,
24            client_id: self.client_id.clone(),
25        }
26        .encode_v1(&mut encoder)?;
27        encoder.write_string(&self.transactional_id)?;
28        encoder.write_i64(self.producer_id);
29        encoder.write_i16(self.producer_epoch);
30        encoder.write_array(Some(&self.topics), |encoder, topic| topic.encode(encoder))?;
31        Ok(encoder.into_bytes())
32    }
33}
34
35#[derive(Debug, Clone, PartialEq, Eq)]
36pub struct AddPartitionsToTxnTopic {
37    pub name: String,
38    pub partitions: Vec<i32>,
39}
40
41impl AddPartitionsToTxnTopic {
42    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
43        encoder.write_string(&self.name)?;
44        encoder.write_array(Some(&self.partitions), |encoder, partition| {
45            encoder.write_i32(*partition);
46            Ok(())
47        })
48    }
49}
50
51#[derive(Debug, Clone, PartialEq, Eq)]
52pub struct AddPartitionsToTxnResponseV0 {
53    pub throttle_time_ms: i32,
54    pub errors: Vec<AddPartitionsToTxnTopicResult>,
55}
56
57impl AddPartitionsToTxnResponseV0 {
58    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
59        Ok(Self {
60            throttle_time_ms: decoder.read_i32()?,
61            errors: decoder
62                .read_array(
63                    "add partitions to transaction topic results",
64                    AddPartitionsToTxnTopicResult::decode,
65                )?
66                .unwrap_or_default(),
67        })
68    }
69}
70
71#[derive(Debug, Clone, PartialEq, Eq)]
72pub struct AddPartitionsToTxnTopicResult {
73    pub name: String,
74    pub partitions: Vec<AddPartitionsToTxnPartitionResult>,
75}
76
77impl AddPartitionsToTxnTopicResult {
78    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
79        Ok(Self {
80            name: decoder.read_string()?,
81            partitions: decoder
82                .read_array(
83                    "add partitions to transaction partition results",
84                    AddPartitionsToTxnPartitionResult::decode,
85                )?
86                .unwrap_or_default(),
87        })
88    }
89}
90
91#[derive(Debug, Clone, PartialEq, Eq)]
92pub struct AddPartitionsToTxnPartitionResult {
93    pub partition_index: i32,
94    pub error_code: i16,
95}
96
97impl AddPartitionsToTxnPartitionResult {
98    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
99        Ok(Self {
100            partition_index: decoder.read_i32()?,
101            error_code: decoder.read_i16()?,
102        })
103    }
104}
105
106#[cfg(test)]
107#[allow(clippy::unwrap_used)]
108mod tests {
109    use super::{
110        AddPartitionsToTxnRequestV0, AddPartitionsToTxnResponseV0, AddPartitionsToTxnTopic, API_KEY,
111    };
112    use crate::codec::Decoder;
113
114    #[test]
115    fn encodes_add_partitions_to_txn_v0_request() {
116        let request = AddPartitionsToTxnRequestV0 {
117            correlation_id: 41,
118            client_id: Some("kafrust".to_owned()),
119            transactional_id: "orders-tx".to_owned(),
120            producer_id: 42,
121            producer_epoch: 3,
122            topics: vec![AddPartitionsToTxnTopic {
123                name: "orders".to_owned(),
124                partitions: vec![0, 2],
125            }],
126        };
127        let encoded = request.encode().unwrap();
128
129        assert_eq!(&encoded[0..8], &[0, 24, 0, 0, 0, 0, 0, 41]);
130        assert_eq!(
131            &encoded[encoded.len() - 12..],
132            &[0, 0, 0, 2, 0, 0, 0, 0, 0, 0, 0, 2]
133        );
134        assert_eq!(API_KEY, 24);
135    }
136
137    #[test]
138    fn decodes_add_partitions_to_txn_v0_response() {
139        let bytes = [
140            0, 0, 0, 7, // throttle time
141            0, 0, 0, 1, // topic count
142            0, 6, b'o', b'r', b'd', b'e', b'r', b's', // topic
143            0, 0, 0, 2, // partition count
144            0, 0, 0, 0, 0, 0, // partition 0, success
145            0, 0, 0, 2, 0, 47, // partition 2, invalid producer epoch
146        ];
147        let mut decoder = Decoder::new(&bytes);
148        let response = AddPartitionsToTxnResponseV0::decode_body(&mut decoder).unwrap();
149
150        assert_eq!(response.throttle_time_ms, 7);
151        assert_eq!(response.errors[0].name, "orders");
152        assert_eq!(response.errors[0].partitions[0].error_code, 0);
153        assert_eq!(response.errors[0].partitions[1].partition_index, 2);
154        assert_eq!(response.errors[0].partitions[1].error_code, 47);
155        assert!(decoder.is_empty());
156    }
157}
kafrust_protocol/api/add_partitions_to_txn.rs

kafrust_protocol/api/
add_partitions_to_txn.rs
