kafrust_protocol/api/
incremental_alter_configs.rs

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

kafrust_protocol/api/
incremental_alter_configs.rs
