krabka_client_core/
version.rs

1//! `ApiVersionTable`: broker-advertised version ranges per API key,
2//! plus client-side negotiation.
3
4use std::collections::{BTreeMap, HashMap};
5
6use api_versions_response::ApiVersionsResponse;
7use krabka_protocol::owned::api_versions_response;
8
9use crate::{error::ClientError, request::ProtocolRequest};
10
11/// Kafka's `ApiVersionsResponse.UNKNOWN_FINALIZED_FEATURES_EPOCH`.
12pub const UNKNOWN_FINALIZED_FEATURES_EPOCH: i64 = -1;
13
14/// The finalized feature levels of a broker (KIP-584), from `ApiVersions` v3
15/// and later, as Kafka's `ApiVersions.FinalizedFeaturesInfo` holds them.
16#[derive(Clone, Debug, Eq, PartialEq)]
17pub struct FinalizedFeatures {
18    /// The epoch of the levels. [`UNKNOWN_FINALIZED_FEATURES_EPOCH`] when the
19    /// broker sent none.
20    pub epoch: i64,
21    /// The finalized maximum version level of each feature, for example
22    /// `transaction.version`.
23    pub levels: BTreeMap<String, i16>,
24}
25
26impl Default for FinalizedFeatures {
27    fn default() -> Self {
28        Self {
29            epoch: UNKNOWN_FINALIZED_FEATURES_EPOCH,
30            levels: BTreeMap::new(),
31        }
32    }
33}
34
35impl FinalizedFeatures {
36    /// The finalized level of `feature`, or `None` when the broker did not
37    /// finalize it.
38    #[must_use]
39    pub fn level(&self, feature: &str) -> Option<i16> {
40        self.levels.get(feature).copied()
41    }
42}
43
44#[derive(Debug, Clone, Default)]
45pub struct ApiVersionTable {
46    by_key: HashMap<i16, (i16, i16)>,
47    /// The supported `(min, max)` version of each feature of the broker.
48    supported_features: BTreeMap<String, (i16, i16)>,
49    finalized_features: FinalizedFeatures,
50}
51
52impl ApiVersionTable {
53    /// Build from a sequence of `(api_key, broker_min, broker_max)` tuples.
54    /// Used when seeding the table from a decoded `ApiVersionsResponse`.
55    #[must_use]
56    pub fn from_entries(entries: impl IntoIterator<Item = (i16, i16, i16)>) -> Self {
57        let mut by_key = HashMap::new();
58        for (k, lo, hi) in entries {
59            by_key.insert(k, (lo, hi));
60        }
61        Self {
62            by_key,
63            ..Self::default()
64        }
65    }
66
67    /// Build from a decoded `ApiVersionsResponse`, with its feature data.
68    #[must_use]
69    pub fn from_response(response: &ApiVersionsResponse) -> Self {
70        let mut table = Self::from_entries(
71            response
72                .api_keys
73                .iter()
74                .map(|key| (key.api_key, key.min_version, key.max_version)),
75        );
76        table.supported_features = response
77            .supported_features
78            .iter()
79            .map(|feature| {
80                (
81                    feature.name.clone(),
82                    (feature.min_version, feature.max_version),
83                )
84            })
85            .collect();
86        table.finalized_features = FinalizedFeatures {
87            epoch: response.finalized_features_epoch,
88            levels: response
89                .finalized_features
90                .iter()
91                .map(|feature| (feature.name.clone(), feature.max_version_level))
92                .collect(),
93        };
94        table
95    }
96
97    /// The finalized feature levels that the broker sent.
98    #[must_use]
99    pub const fn finalized_features(&self) -> &FinalizedFeatures {
100        &self.finalized_features
101    }
102
103    /// The supported `(min, max)` version of each feature of the broker.
104    #[must_use]
105    pub const fn supported_features(&self) -> &BTreeMap<String, (i16, i16)> {
106        &self.supported_features
107    }
108
109    /// Highest version both sides support for `R`, or
110    /// [`ClientError::IncompatibleVersion`] if the ranges do not overlap.
111    ///
112    /// The client side ends at `R::LATEST_STABLE_VERSION`. Kafka's
113    /// `AbstractRequest.Builder` uses `ApiKeys.latestVersion(false)`, which
114    /// leaves out a `latestVersionUnstable` version, and
115    /// `NodeApiVersions.latestUsableVersion` picks the highest version of that
116    /// range that the broker also supports.
117    /// # Errors
118    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
119    pub fn negotiate<R: ProtocolRequest>(&self) -> Result<i16, ClientError> {
120        let api_key = R::API_KEY;
121        let client_min = R::MIN_VERSION;
122        let client_max = R::LATEST_STABLE_VERSION;
123        let (broker_min, broker_max) = self.by_key.get(&api_key).copied().unwrap_or((0, 0));
124        let chosen = client_max.min(broker_max);
125        if chosen < client_min || chosen < broker_min {
126            return Err(ClientError::IncompatibleVersion {
127                api_key,
128                broker_min,
129                broker_max,
130                client_min,
131                client_max,
132            });
133        }
134        Ok(chosen)
135    }
136
137    /// Return the broker-advertised `(min, max)` version range for `api_key`,
138    /// or `None` if the broker did not advertise it.
139    #[must_use]
140    pub fn broker_range(&self, api_key: i16) -> Option<(i16, i16)> {
141        self.by_key.get(&api_key).copied()
142    }
143
144    /// Returns `true` if the table contains no entries.
145    #[must_use]
146    pub fn is_empty(&self) -> bool {
147        self.by_key.is_empty()
148    }
149}
150
151#[cfg(test)]
152mod tests {
153    use assert2::assert;
154    use krabka_protocol::owned::api_versions_request::ApiVersionsRequest;
155
156    use super::*;
157
158    // `ApiVersionsRequest` acts as a sample `ProtocolRequest`. We only
159    // need the trait's constants here; the impl comes from codegen.
160
161    /// A request whose latest version is unstable, as Kafka's
162    /// `InitProducerIdRequest.json` marks v6.
163    struct UnstableLatest;
164
165    impl krabka_protocol::Encode for UnstableLatest {
166        fn encode<B: bytes::BufMut>(
167            &self,
168            _buf: &mut B,
169            _version: i16,
170        ) -> Result<(), krabka_protocol::ProtocolError> {
171            Ok(())
172        }
173
174        fn encoded_len(&self, _version: i16) -> usize {
175            0
176        }
177    }
178
179    impl ProtocolRequest for UnstableLatest {
180        const API_KEY: i16 = 22;
181        const MIN_VERSION: i16 = 0;
182        const MAX_VERSION: i16 = 6;
183        const LATEST_STABLE_VERSION: i16 = 5;
184        const FLEXIBLE_MIN: i16 = 2;
185        type Response = api_versions_response::ApiVersionsResponse;
186    }
187
188    /// Negotiation stops at the latest stable version, as Kafka's
189    /// `latestVersion(false)` does, and an unstable-only overlap fails.
190    #[test]
191    fn negotiate_leaves_out_an_unstable_latest_version() {
192        let cases = [
193            ("broker supports the unstable version", (0, 6), Ok(5)),
194            ("broker stops at the stable version", (0, 5), Ok(5)),
195            ("broker stops below", (0, 3), Ok(3)),
196            (
197                "broker supports only the unstable version",
198                (6, 6),
199                Err((6, 6)),
200            ),
201        ];
202        for (name, (broker_min, broker_max), expected) in cases {
203            let table = ApiVersionTable::from_entries([(22, broker_min, broker_max)]);
204            let actual = table
205                .negotiate::<UnstableLatest>()
206                .map_err(|error| match error {
207                    ClientError::IncompatibleVersion {
208                        broker_min,
209                        broker_max,
210                        ..
211                    } => (broker_min, broker_max),
212                    other => panic!("{name}: {other}"),
213                });
214            assert!(actual == expected, "{name}");
215        }
216    }
217
218    #[test]
219    fn negotiate_takes_min_of_max() {
220        let t = ApiVersionTable::from_entries([(
221            ApiVersionsRequest::API_KEY,
222            0,
223            ApiVersionsRequest::MAX_VERSION,
224        )]);
225        // Sanity: client max wins if broker max is higher.
226        let _ = t.negotiate::<ApiVersionsRequest>().unwrap();
227    }
228
229    #[test]
230    fn negotiate_errors_when_disjoint() {
231        let t = ApiVersionTable::from_entries([(ApiVersionsRequest::API_KEY, 99, 100)]);
232        assert!(matches!(
233            t.negotiate::<ApiVersionsRequest>(),
234            Err(ClientError::IncompatibleVersion { .. })
235        ));
236    }
237
238    #[test]
239    fn negotiate_picks_lowest_supported_when_broker_caps_low() {
240        let t = ApiVersionTable::from_entries([(ApiVersionsRequest::API_KEY, 0, 0)]);
241        // Both sides support 0; that's what's chosen.
242        assert!(t.negotiate::<ApiVersionsRequest>().unwrap() == 0);
243    }
244
245    #[test]
246    fn broker_range_reports_exact_advertised_bounds() {
247        let t = ApiVersionTable::from_entries([(ApiVersionsRequest::API_KEY, 2, 4)]);
248
249        assert!(t.broker_range(ApiVersionsRequest::API_KEY) == Some((2, 4)));
250        assert!(t.broker_range(ApiVersionsRequest::API_KEY + 1).is_none());
251    }
252
253    #[test]
254    fn is_empty_reflects_whether_any_versions_were_advertised() {
255        assert!(ApiVersionTable::default().is_empty());
256        assert!(!ApiVersionTable::from_entries([(ApiVersionsRequest::API_KEY, 0, 1)]).is_empty());
257    }
258}
krabka_client_core/version.rs

krabka_client_core/
version.rs
