kacrab_protocol/
version.rs

1//! API version resolution + header version selection.
2//!
3//! Kafka clients negotiate the highest mutually-supported API version per
4//! request type with the broker; the resolved version then determines whether
5//! the request/response uses the flexible (varint-prefixed, tagged-fields)
6//! header layout or the older fixed-width one.
7
8pub mod error;
9
10pub use self::error::{UnsupportedFieldVersion, UnsupportedVersion};
11pub use crate::generated::{ApiInfo, ApiKey, client_api_info};
12
13/// Result alias for version operations.
14pub type Result<T> = core::result::Result<T, UnsupportedVersion>;
15
16/// Kafka `ApiVersions` API key (`18`). Special-cased by KIP-511 for response
17/// header version selection.
18pub const API_VERSIONS_KEY: i16 = 18;
19
20/// A contiguous range of supported API versions, inclusive on both ends.
21#[derive(Debug, Clone, Copy, PartialEq, Eq)]
22pub struct ApiVersionRange {
23    /// Lowest supported version.
24    pub min_version: i16,
25    /// Highest supported version.
26    pub max_version: i16,
27}
28
29/// Resolve the highest API version supported by both client and broker.
30///
31/// Returns `None` when the broker range is empty (`min_version > max_version`),
32/// when `api_key` is not a known API key, or when the client and broker ranges
33/// are disjoint.
34#[must_use]
35pub fn resolve_api_version(api_key: i16, broker_range: ApiVersionRange) -> Option<i16> {
36    if broker_range.min_version > broker_range.max_version {
37        return None;
38    }
39
40    let api_key = ApiKey::from_i16(api_key)?;
41    let info = client_api_info(api_key);
42    let min_version = broker_range.min_version.max(info.min_version);
43    let max_version = broker_range.max_version.min(info.max_version);
44    if min_version > max_version {
45        return None;
46    }
47
48    Some(max_version)
49}
50
51/// Request header version for `(api_key, api_version)`.
52///
53/// Flexible versions use header v2; most non-flexible versions use header v1.
54/// `ControlledShutdown` v0 is the legacy request-header v0 exception.
55#[must_use]
56pub fn request_header_version(api_key: i16, api_version: i16) -> i16 {
57    if api_key == ApiKey::ControlledShutdown as i16 && api_version == 0 {
58        return 0;
59    }
60    if is_flexible_version(api_key, api_version) {
61        2
62    } else {
63        1
64    }
65}
66
67/// Response header version for `(api_key, api_version)`.
68///
69/// Flexible versions use header v1; non-flexible use header v0. Special case
70/// (KIP-511): `ApiVersions` always uses header v0 because the client doesn't
71/// yet know the broker's supported versions when it parses this response.
72#[must_use]
73pub fn response_header_version(api_key: i16, api_version: i16) -> i16 {
74    if api_key == API_VERSIONS_KEY {
75        return 0;
76    }
77    i16::from(is_flexible_version(api_key, api_version))
78}
79
80fn is_flexible_version(api_key: i16, api_version: i16) -> bool {
81    let Some(api_key) = ApiKey::from_i16(api_key) else {
82        return false;
83    };
84    let info = client_api_info(api_key);
85    api_version >= info.flexible_versions_start
86        && api_version >= info.min_version
87        && api_version <= info.max_version
88}
89
90#[cfg(test)]
91mod tests {
92    use super::{
93        API_VERSIONS_KEY, ApiVersionRange, request_header_version, resolve_api_version,
94        response_header_version,
95    };
96    use crate::generated::ApiKey;
97
98    #[test]
99    fn resolve_api_version_intersects_client_and_broker_ranges() {
100        let metadata = ApiKey::Metadata as i16;
101
102        assert_eq!(
103            resolve_api_version(
104                metadata,
105                ApiVersionRange {
106                    min_version: 0,
107                    max_version: 99,
108                },
109            ),
110            Some(13),
111        );
112        assert_eq!(
113            resolve_api_version(
114                metadata,
115                ApiVersionRange {
116                    min_version: 0,
117                    max_version: 2,
118                },
119            ),
120            Some(2),
121        );
122        assert_eq!(
123            resolve_api_version(
124                metadata,
125                ApiVersionRange {
126                    min_version: 99,
127                    max_version: 100,
128                },
129            ),
130            None,
131        );
132        assert_eq!(
133            resolve_api_version(
134                -32,
135                ApiVersionRange {
136                    min_version: 0,
137                    max_version: 1,
138                },
139            ),
140            None,
141        );
142    }
143
144    #[test]
145    fn header_versions_follow_flexible_metadata_and_special_cases() {
146        let metadata = ApiKey::Metadata as i16;
147        let controlled_shutdown = ApiKey::ControlledShutdown as i16;
148
149        assert_eq!(request_header_version(controlled_shutdown, 0), 0);
150        assert_eq!(request_header_version(metadata, 8), 1);
151        assert_eq!(request_header_version(metadata, 9), 2);
152
153        assert_eq!(response_header_version(API_VERSIONS_KEY, 3), 0);
154        assert_eq!(response_header_version(metadata, 8), 0);
155        assert_eq!(response_header_version(metadata, 9), 1);
156    }
157}
kacrab_protocol/version.rs

kacrab_protocol/
version.rs
