kafrust_protocol/api/
alter_configs.rs

1use crate::codec::{Decoder, Encoder};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 33;
6
7#[derive(Debug, Clone, PartialEq, Eq)]
8pub struct AlterConfigsRequestV1 {
9    pub correlation_id: i32,
10    pub client_id: Option<String>,
11    pub resources: Vec<AlterConfigsResourceV1>,
12    pub validate_only: bool,
13}
14
15impl AlterConfigsRequestV1 {
16    pub fn encode(&self) -> Result<Vec<u8>> {
17        let mut encoder = Encoder::new();
18        RequestHeader {
19            api_key: API_KEY,
20            api_version: 1,
21            correlation_id: self.correlation_id,
22            client_id: self.client_id.clone(),
23        }
24        .encode_v1(&mut encoder)?;
25        encoder.write_array(Some(&self.resources), |encoder, resource| {
26            resource.encode(encoder)
27        })?;
28        encoder.write_bool(self.validate_only);
29        Ok(encoder.into_bytes())
30    }
31}
32
33#[derive(Debug, Clone, PartialEq, Eq)]
34pub struct AlterConfigsResourceV1 {
35    pub resource_type: i8,
36    pub resource_name: String,
37    pub configs: Vec<AlterableConfigV1>,
38}
39
40impl AlterConfigsResourceV1 {
41    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
42        encoder.write_i8(self.resource_type);
43        encoder.write_string(&self.resource_name)?;
44        encoder.write_array(Some(&self.configs), |encoder, config| {
45            config.encode(encoder)
46        })
47    }
48}
49
50#[derive(Debug, Clone, PartialEq, Eq)]
51pub struct AlterableConfigV1 {
52    pub name: String,
53    pub value: Option<String>,
54}
55
56impl AlterableConfigV1 {
57    fn encode(&self, encoder: &mut Encoder) -> Result<()> {
58        encoder.write_string(&self.name)?;
59        encoder.write_nullable_string(self.value.as_deref())
60    }
61}
62
63#[derive(Debug, Clone, PartialEq, Eq)]
64pub struct AlterConfigsResponseV1 {
65    pub throttle_time_ms: i32,
66    pub responses: Vec<AlterConfigsResourceResponseV1>,
67}
68
69impl AlterConfigsResponseV1 {
70    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
71        Ok(Self {
72            throttle_time_ms: decoder.read_i32()?,
73            responses: decoder
74                .read_array(
75                    "alter configs responses",
76                    AlterConfigsResourceResponseV1::decode,
77                )?
78                .unwrap_or_default(),
79        })
80    }
81}
82
83#[derive(Debug, Clone, PartialEq, Eq)]
84pub struct AlterConfigsResourceResponseV1 {
85    pub error_code: i16,
86    pub error_message: Option<String>,
87    pub resource_type: i8,
88    pub resource_name: String,
89}
90
91impl AlterConfigsResourceResponseV1 {
92    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
93        Ok(Self {
94            error_code: decoder.read_i16()?,
95            error_message: decoder.read_nullable_string()?,
96            resource_type: decoder.read_i8()?,
97            resource_name: decoder.read_string()?,
98        })
99    }
100}
101
102#[cfg(test)]
103#[allow(clippy::unwrap_used)]
104mod tests {
105    use super::{
106        AlterConfigsRequestV1, AlterConfigsResourceV1, AlterConfigsResponseV1, AlterableConfigV1,
107        API_KEY,
108    };
109    use crate::codec::Decoder;
110
111    #[test]
112    fn encodes_alter_configs_v1_request() {
113        let request = AlterConfigsRequestV1 {
114            correlation_id: 14,
115            client_id: Some("kafrust".to_owned()),
116            resources: vec![AlterConfigsResourceV1 {
117                resource_type: 2,
118                resource_name: "orders".to_owned(),
119                configs: vec![
120                    AlterableConfigV1 {
121                        name: "retention.ms".to_owned(),
122                        value: Some("60000".to_owned()),
123                    },
124                    AlterableConfigV1 {
125                        name: "cleanup.policy".to_owned(),
126                        value: None,
127                    },
128                ],
129            }],
130            validate_only: true,
131        };
132
133        assert_eq!(
134            request.encode().unwrap(),
135            [
136                0, 33, // API key
137                0, 1, // API version
138                0, 0, 0, 14, // correlation ID
139                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // client ID
140                0, 0, 0, 1, // resource count
141                2, // topic resource
142                0, 6, b'o', b'r', b'd', b'e', b'r', b's', // resource name
143                0, 0, 0, 2, // config count
144                0, 12, b'r', b'e', b't', b'e', b'n', b't', b'i', b'o', b'n', b'.', b'm', b's', 0,
145                5, b'6', b'0', b'0', b'0', b'0', // value
146                0, 14, b'c', b'l', b'e', b'a', b'n', b'u', b'p', b'.', b'p', b'o', b'l', b'i',
147                b'c', b'y', // config name
148                0xff, 0xff, // null value
149                1,    // validate only
150            ]
151        );
152        assert_eq!(API_KEY, 33);
153    }
154
155    #[test]
156    fn decodes_alter_configs_v1_response() {
157        let bytes = [
158            0, 0, 0, 7, // throttle time
159            0, 0, 0, 2, // response count
160            0, 0, // success
161            0xff, 0xff, // null error message
162            2,    // topic resource
163            0, 6, b'o', b'r', b'd', b'e', b'r', b's', // resource name
164            0, 40, // invalid config
165            0, 7, b'i', b'n', b'v', b'a', b'l', b'i', b'd', // error message
166            2,    // topic resource
167            0, 8, b'p', b'a', b'y', b'm', b'e', b'n', b't', b's', // resource name
168        ];
169        let mut decoder = Decoder::new(&bytes);
170
171        let response = AlterConfigsResponseV1::decode_body(&mut decoder).unwrap();
172
173        assert_eq!(response.throttle_time_ms, 7);
174        assert_eq!(response.responses.len(), 2);
175        assert_eq!(response.responses[0].resource_name, "orders");
176        assert_eq!(response.responses[0].error_code, 0);
177        assert_eq!(response.responses[1].resource_name, "payments");
178        assert_eq!(response.responses[1].error_code, 40);
179        assert_eq!(
180            response.responses[1].error_message.as_deref(),
181            Some("invalid")
182        );
183        assert!(decoder.is_empty());
184    }
185}
kafrust_protocol/api/alter_configs.rs

kafrust_protocol/api/
alter_configs.rs
