kafrust_protocol/api/
api_versions.rs

1use crate::codec::{Decoder, Encoder, TaggedField};
2use crate::error::Result;
3use crate::header::RequestHeader;
4
5pub const API_KEY: i16 = 18;
6
7#[derive(Debug, Clone, Copy, PartialEq, Eq)]
8pub struct ApiKeyVersion {
9    pub api_key: i16,
10    pub min_version: i16,
11    pub max_version: i16,
12}
13
14/// Common capability lookup implemented by fixed and flexible ApiVersions responses.
15pub trait ApiVersionsLookup {
16    /// Returns the highest broker-supported version not exceeding the client limit.
17    fn highest_supported_version(&self, api_key: i16, max_supported: i16) -> Option<i16>;
18}
19
20impl ApiKeyVersion {
21    fn decode(decoder: &mut Decoder<'_>) -> Result<Self> {
22        Ok(Self {
23            api_key: decoder.read_i16()?,
24            min_version: decoder.read_i16()?,
25            max_version: decoder.read_i16()?,
26        })
27    }
28
29    fn decode_flexible(decoder: &mut Decoder<'_>) -> Result<Self> {
30        let version = Self::decode(decoder)?;
31        decoder.read_tagged_fields()?;
32        Ok(version)
33    }
34}
35
36#[derive(Debug, Clone, PartialEq, Eq)]
37pub struct ApiVersionsRequestV0 {
38    pub correlation_id: i32,
39    pub client_id: Option<String>,
40}
41
42impl ApiVersionsRequestV0 {
43    pub fn encode(&self) -> Result<Vec<u8>> {
44        let mut encoder = Encoder::new();
45        RequestHeader {
46            api_key: API_KEY,
47            api_version: 0,
48            correlation_id: self.correlation_id,
49            client_id: self.client_id.clone(),
50        }
51        .encode_v1(&mut encoder)?;
52        Ok(encoder.into_bytes())
53    }
54}
55
56/// ApiVersions v3 request with client software identification.
57#[derive(Debug, Clone, PartialEq, Eq)]
58pub struct ApiVersionsRequestV3 {
59    /// Request correlation ID.
60    pub correlation_id: i32,
61    /// Optional Kafka client ID from the request header.
62    pub client_id: Option<String>,
63    /// Client software name reported through KIP-511.
64    pub client_software_name: String,
65    /// Client software version reported through KIP-511.
66    pub client_software_version: String,
67}
68
69impl ApiVersionsRequestV3 {
70    /// Encodes an ApiVersions v3 request frame without the outer frame length.
71    pub fn encode(&self) -> Result<Vec<u8>> {
72        let mut encoder = Encoder::new();
73        RequestHeader {
74            api_key: API_KEY,
75            api_version: 3,
76            correlation_id: self.correlation_id,
77            client_id: self.client_id.clone(),
78        }
79        .encode_v2(&mut encoder)?;
80        encoder.write_compact_string(&self.client_software_name)?;
81        encoder.write_compact_string(&self.client_software_version)?;
82        encoder.write_empty_tagged_fields();
83        Ok(encoder.into_bytes())
84    }
85}
86
87#[derive(Debug, Clone, PartialEq, Eq)]
88pub struct ApiVersionsResponseV0 {
89    pub error_code: i16,
90    pub api_keys: Vec<ApiKeyVersion>,
91}
92
93impl ApiVersionsResponseV0 {
94    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
95        let error_code = decoder.read_i16()?;
96        let api_keys = decoder
97            .read_array("api versions", ApiKeyVersion::decode)?
98            .unwrap_or_default();
99        Ok(Self {
100            error_code,
101            api_keys,
102        })
103    }
104
105    pub fn highest_supported_version(&self, api_key: i16, max_supported: i16) -> Option<i16> {
106        self.api_keys
107            .iter()
108            .find(|version| version.api_key == api_key)
109            .and_then(|version| {
110                let selected = version.max_version.min(max_supported);
111                (selected >= version.min_version).then_some(selected)
112            })
113    }
114}
115
116impl ApiVersionsLookup for ApiVersionsResponseV0 {
117    fn highest_supported_version(&self, api_key: i16, max_supported: i16) -> Option<i16> {
118        highest_supported_version(&self.api_keys, api_key, max_supported)
119    }
120}
121
122/// ApiVersions v3 response with flexible encoding and forward-compatible tags.
123#[derive(Debug, Clone, PartialEq, Eq)]
124pub struct ApiVersionsResponseV3 {
125    /// Top-level Kafka error code.
126    pub error_code: i16,
127    /// API version ranges advertised by the broker.
128    pub api_keys: Vec<ApiKeyVersion>,
129    /// Broker throttle duration in milliseconds.
130    pub throttle_time_ms: i32,
131    /// Unknown or future top-level tagged fields preserved for inspection.
132    pub tagged_fields: Vec<TaggedField>,
133}
134
135impl ApiVersionsResponseV3 {
136    /// Decodes an ApiVersions v3 response body after its response header.
137    pub fn decode_body(decoder: &mut Decoder<'_>) -> Result<Self> {
138        let error_code = decoder.read_i16()?;
139        let api_keys = decoder
140            .read_compact_array("api versions", ApiKeyVersion::decode_flexible)?
141            .unwrap_or_default();
142        let throttle_time_ms = decoder.read_i32()?;
143        let tagged_fields = decoder.read_tagged_fields()?;
144        Ok(Self {
145            error_code,
146            api_keys,
147            throttle_time_ms,
148            tagged_fields,
149        })
150    }
151
152    /// Returns the highest broker-supported version not exceeding the client limit.
153    pub fn highest_supported_version(&self, api_key: i16, max_supported: i16) -> Option<i16> {
154        highest_supported_version(&self.api_keys, api_key, max_supported)
155    }
156}
157
158impl ApiVersionsLookup for ApiVersionsResponseV3 {
159    fn highest_supported_version(&self, api_key: i16, max_supported: i16) -> Option<i16> {
160        highest_supported_version(&self.api_keys, api_key, max_supported)
161    }
162}
163
164fn highest_supported_version(
165    api_keys: &[ApiKeyVersion],
166    api_key: i16,
167    max_supported: i16,
168) -> Option<i16> {
169    api_keys
170        .iter()
171        .find(|version| version.api_key == api_key)
172        .and_then(|version| {
173            let selected = version.max_version.min(max_supported);
174            (selected >= version.min_version).then_some(selected)
175        })
176}
177
178#[cfg(test)]
179#[allow(clippy::unwrap_used)]
180mod tests {
181    use super::{
182        ApiVersionsRequestV0, ApiVersionsRequestV3, ApiVersionsResponseV0, ApiVersionsResponseV3,
183        API_KEY,
184    };
185    use crate::codec::Decoder;
186
187    #[test]
188    fn encodes_api_versions_request_v0() {
189        let request = ApiVersionsRequestV0 {
190            correlation_id: 42,
191            client_id: Some("kafrust".to_owned()),
192        };
193        assert_eq!(
194            request.encode().unwrap(),
195            [
196                0, 18, // api key
197                0, 0, // api version
198                0, 0, 0, 42, // correlation id
199                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't',
200            ]
201        );
202        assert_eq!(API_KEY, 18);
203    }
204
205    #[test]
206    fn decodes_api_versions_response_v0() {
207        let bytes = [
208            0, 0, // error code
209            0, 0, 0, 2, // api key count
210            0, 18, 0, 0, 0, 4, // ApiVersions min/max
211            0, 3, 0, 1, 0, 9, // Metadata min/max
212        ];
213        let mut decoder = Decoder::new(&bytes);
214        let response = ApiVersionsResponseV0::decode_body(&mut decoder).unwrap();
215
216        assert_eq!(response.error_code, 0);
217        assert_eq!(response.api_keys.len(), 2);
218        assert_eq!(response.highest_supported_version(18, 3), Some(3));
219        assert_eq!(response.highest_supported_version(3, 12), Some(9));
220        assert_eq!(response.highest_supported_version(1, 1), None);
221        assert!(decoder.is_empty());
222    }
223
224    #[test]
225    fn encodes_api_versions_request_v3() {
226        let request = ApiVersionsRequestV3 {
227            correlation_id: 42,
228            client_id: Some("kafrust".to_owned()),
229            client_software_name: "kafrust".to_owned(),
230            client_software_version: "0.3.0".to_owned(),
231        };
232        assert_eq!(
233            request.encode().unwrap(),
234            [
235                0, 18, // api key
236                0, 3, // api version
237                0, 0, 0, 42, // correlation id
238                0, 7, b'k', b'a', b'f', b'r', b'u', b's', b't', // nullable client id
239                0,    // request tagged fields
240                8, b'k', b'a', b'f', b'r', b'u', b's', b't', // software name
241                6, b'0', b'.', b'3', b'.', b'0', // software version
242                0,    // request body tagged fields
243            ]
244        );
245    }
246
247    #[test]
248    fn decodes_api_versions_response_v3() {
249        let bytes = [
250            0, 0, // error code
251            3, // compact api key count: two entries
252            0, 18, 0, 0, 0, 4, 0, // ApiVersions entry + tagged fields
253            0, 3, 0, 1, 0, 9, 0, // Metadata entry + tagged fields
254            0, 0, 0, 17, // throttle time
255            0,  // top-level tagged fields
256        ];
257        let mut decoder = Decoder::new(&bytes);
258        let response = ApiVersionsResponseV3::decode_body(&mut decoder).unwrap();
259
260        assert_eq!(response.error_code, 0);
261        assert_eq!(response.throttle_time_ms, 17);
262        assert_eq!(response.api_keys.len(), 2);
263        assert_eq!(response.highest_supported_version(18, 3), Some(3));
264        assert_eq!(response.highest_supported_version(3, 12), Some(9));
265        assert_eq!(response.highest_supported_version(1, 1), None);
266        assert!(response.tagged_fields.is_empty());
267        assert!(decoder.is_empty());
268    }
269}
kafrust_protocol/api/api_versions.rs

kafrust_protocol/api/
api_versions.rs
