krabka_raft/server/
api_versions.rs

1//! The `ApiVersions` handshake that every controller-listener connection begins
2//! with: the advertised API table, the supported and finalized feature ranges,
3//! and the request checks of Kafka's `ControllerApis.handleApiVersionsRequest`.
4
5pub use self::{
6    client_software::is_valid_client_info,
7    features::{finalized_feature_keys, supported_feature_key, supported_feature_keys},
8};
9use {
10    self::table::CONTROLLER_LISTENER_APIS,
11    crate::{
12        ControllerAdminRouter, ControllerApiVersion, UnstableApiVersions, UnstableFeatureVersions,
13        error::RaftError, kafka_4_3_1_max,
14    },
15    bytes::{Bytes, BytesMut},
16    krabka_protocol::{
17        Decode, Encode,
18        owned::{
19            api_versions_request::{self, ApiVersionsRequest},
20            api_versions_response::{ApiVersion as ApiVersionEntry, ApiVersionsResponse},
21        },
22    },
23};
24
25mod client_software;
26mod features;
27pub(super) mod table;
28
29/// Kafka's `ApiVersions` API key. The controller TCP listener answers this
30/// because `krabka_client_core::Connection::connect` performs an `ApiVersions`
31/// handshake before any other request.
32pub(super) const API_KEY_API_VERSIONS: i16 = 18;
33
34/// Lowest `ApiVersions` request version this listener speaks.
35const API_VERSIONS_MIN_VERSION: i16 = api_versions_request::MIN_VERSION;
36/// Highest `ApiVersions` request version this listener decodes: the generated
37/// maximum. [`api_versions_max_version`] is what it serves.
38const API_VERSIONS_MAX_VERSION: i16 = api_versions_request::MAX_VERSION;
39
40/// Highest `ApiVersions` version a listener serves under `unstable`.
41///
42/// It is the same maximum its `api_keys` table advertises for API 18: Kafka 4.3.1's v4 by
43/// default, and trunk's KIP-1242 v5 under `unstable.api.versions.enable`.
44#[must_use]
45pub fn api_versions_max_version(unstable: UnstableApiVersions) -> i16 {
46    match unstable {
47        UnstableApiVersions::Enabled => API_VERSIONS_MAX_VERSION,
48        UnstableApiVersions::Disabled => {
49            kafka_4_3_1_max(API_KEY_API_VERSIONS).unwrap_or(API_VERSIONS_MAX_VERSION)
50        }
51    }
52}
53/// First `ApiVersions` version that carries the KIP-511 client software name
54/// and version.
55const API_VERSIONS_CLIENT_SOFTWARE_MIN_VERSION: i16 = 3;
56/// First `ApiVersions` version that carries the KIP-1242 cluster id and node
57/// id.
58const API_VERSIONS_ROUTING_MIN_VERSION: i16 = 5;
59const API_VERSIONS_UNSUPPORTED_VERSION: i16 = 35;
60const API_VERSIONS_INVALID_REQUEST: i16 = 42;
61
62/// What one controller-listener `ApiVersions` answer is built from.
63#[derive(Clone, Copy)]
64pub struct ApiVersionsView<'a> {
65    /// The applied metadata image: its finalized features.
66    pub(crate) image: &'a krabka_metadata::MetadataImage,
67    /// The finalized-features epoch: the last contained offset of the newest
68    /// applied batch or snapshot that carried a `FeatureLevelRecord`, or `-1`
69    /// before the first. Kafka's `FeaturesPublisher` takes
70    /// `image.provenance().lastContainedOffset()` only from a delta with a
71    /// features delta, so topic records do not move it.
72    pub(crate) finalized_features_epoch: i64,
73    /// The KIP-919 Admin surface the broker attaches, if any.
74    pub(crate) admin_router: Option<&'a dyn ControllerAdminRouter>,
75    /// Kafka's `unstable.api.versions.enable`. It caps both the API table and
76    /// the supported features: the controller listener's
77    /// `SimpleApiVersionManager` is built with it and takes no
78    /// `unstable.feature.versions.enable`.
79    pub(crate) unstable: UnstableApiVersions,
80}
81
82/// Answers one controller-listener `ApiVersions` request with the response
83/// body. The body always goes out behind a v0 response header.
84///
85/// The checks follow Kafka's `ControllerApis.handleApiVersionsRequest` and
86/// `SaslServerAuthenticator.handleApiVersionsRequest`, which answer the same
87/// way:
88///
89/// 1. A version outside `MIN..=MAX` is not decoded. Kafka's `RequestContext`
90///    parses it as an empty v0 request, and `ApiVersionsRequest.getErrorResponse`
91///    answers `UNSUPPORTED_VERSION` in a v0 body with one `api_keys` entry, the
92///    `ApiVersions` range, so the client retries at a version the listener
93///    serves.
94/// 2. `ApiVersionsRequest.isValid` fails, and the answer is `INVALID_REQUEST`
95///    with an empty table. From v5 the cluster id and the node id must be
96///    both set or both unset. From v3 the client software name and version
97///    must match the KIP-511 pattern.
98/// 3. Otherwise the answer is the full controller-listener table.
99///
100/// Kafka's controller does not run the KIP-1242 `REBOOTSTRAP_REQUIRED` check.
101/// Only `KafkaApis` on the broker listener runs it.
102///
103/// # Errors
104/// Returns the decode error for a supported version whose body does not
105/// decode. Kafka's `RequestContext.parseRequest` also fails such a request,
106/// and the connection closes.
107pub fn api_versions_response(
108    req_version: i16,
109    body: &[u8],
110    view: ApiVersionsView<'_>,
111) -> Result<Bytes, RaftError> {
112    if !(API_VERSIONS_MIN_VERSION..=api_versions_max_version(view.unstable)).contains(&req_version)
113    {
114        return Ok(encode_body(&unsupported_version_response(view.unstable), 0));
115    }
116    let request = ApiVersionsRequest::decode(&mut &body[..], req_version)?;
117    if !is_valid_api_versions_request(&request, req_version) {
118        return Ok(encode_body(
119            &ApiVersionsResponse {
120                error_code: API_VERSIONS_INVALID_REQUEST,
121                ..Default::default()
122            },
123            req_version,
124        ));
125    }
126    Ok(api_versions_response_body(req_version, view))
127}
128
129/// Kafka's `ApiVersionsRequest.getErrorResponse` for `UNSUPPORTED_VERSION`
130/// (KIP-511).
131///
132/// It is error 35 and exactly one `api_keys` entry, the range of
133/// `ApiVersions` itself (`ApiVersionsResponse.toApiVersion(API_VERSIONS)`),
134/// which the client reads to pick the version it retries with.
135///
136/// Every listener, broker or controller, answers with these bytes in a v0 body.
137#[must_use]
138pub fn unsupported_version_response(unstable: UnstableApiVersions) -> ApiVersionsResponse {
139    ApiVersionsResponse {
140        error_code: API_VERSIONS_UNSUPPORTED_VERSION,
141        api_keys: vec![ApiVersionEntry {
142            api_key: API_KEY_API_VERSIONS,
143            min_version: API_VERSIONS_MIN_VERSION,
144            max_version: api_versions_max_version(unstable),
145            ..Default::default()
146        }],
147        ..Default::default()
148    }
149}
150
151/// Kafka's `ApiVersionsRequest.isValid`.
152#[must_use]
153pub fn is_valid_api_versions_request(request: &ApiVersionsRequest, version: i16) -> bool {
154    if version >= API_VERSIONS_ROUTING_MIN_VERSION
155        && (request.cluster_id.is_none() != (request.node_id == -1))
156    {
157        return false;
158    }
159    version < API_VERSIONS_CLIENT_SOFTWARE_MIN_VERSION
160        || (is_valid_client_info(&request.client_software_name)
161            && is_valid_client_info(&request.client_software_version))
162}
163
164fn encode_body(response: &ApiVersionsResponse, version: i16) -> Bytes {
165    let mut body = BytesMut::with_capacity(response.encoded_len(version));
166    // The version is in the generated range, and the fields are defaults or
167    // table entries, so the encoder has nothing to refuse.
168    let _ = response.encode(&mut body, version);
169    body.freeze()
170}
171
172/// The controller listener's advertised table under `unstable`, sorted by API
173/// key: its own APIs and the Admin router's, each capped at the version
174/// `unstable` enables, as Kafka's `ApiKeys.toApiVersion` caps it. An api key
175/// the enabled table does not have -- one Kafka 4.3.1 lacks, while unstable
176/// api versions are disabled -- is left out.
177pub(super) fn advertised_api_keys(
178    admin_router: Option<&dyn ControllerAdminRouter>,
179    unstable: UnstableApiVersions,
180) -> Vec<ApiVersionEntry> {
181    let entry = |version: &ControllerApiVersion| {
182        version
183            .enabled_max(unstable)
184            .map(|max_version| ApiVersionEntry {
185                api_key: version.api_key,
186                min_version: version.min_version,
187                max_version,
188                ..Default::default()
189            })
190    };
191    let mut api_keys: Vec<ApiVersionEntry> =
192        CONTROLLER_LISTENER_APIS.iter().filter_map(entry).collect();
193    if let Some(router) = admin_router {
194        api_keys.extend(router.api_versions().iter().filter_map(entry));
195    }
196    api_keys.sort_unstable_by_key(|version| version.api_key);
197    api_keys
198}
199
200/// Whether `version` of `api_key` is a version that `unstable` disables on
201/// this listener: one above the enabled maximum, or any version of an api key
202/// the enabled table does not have. Kafka's `Processor.parseRequestHeader`
203/// closes the connection on either (`ApiKeys.isVersionEnabled`, and
204/// `ApiKeys.forId` for a key the release does not know).
205pub(super) fn is_disabled_version(
206    api_key: i16,
207    version: i16,
208    admin_router: Option<&dyn ControllerAdminRouter>,
209    unstable: UnstableApiVersions,
210) -> bool {
211    let router_apis = admin_router.map_or(&[][..], |router| router.api_versions());
212    CONTROLLER_LISTENER_APIS
213        .iter()
214        .chain(router_apis)
215        .find(|api| api.api_key == api_key)
216        .is_some_and(|api| api.is_disabled_version(version, unstable))
217}
218
219/// `ApiVersionsResponse` advertising the controller-listener APIs.
220///
221/// A real `mirror.gcr.io/apache/kafka:4.0.0` controller dials peers with `ApiVersions v4` over a
222/// flexible (v2) request header, then consults the returned table to decide
223/// which version of `Vote`/`Fetch`/etc. to send. An EMPTY `api_keys` list made
224/// the JVM treat every raft RPC as `UNSUPPORTED_VERSION` and refuse to send
225/// `Vote` on the wire. Advertising the KIP-595 APIs at the versions Krabka's
226/// engine speaks lets compatible peers proceed to real `Vote`/`Fetch`. Those
227/// versions come from [`table::CONTROLLER_LISTENER_APIS`], which derives them
228/// from the generated message constants; the KIP-919 Admin surface the broker
229/// attaches contributes the rest.
230///
231/// The feature rows follow Kafka's `SimpleApiVersionManager`: the supported
232/// set of `BrokerFeatures.defaultSupportedFeatures(enableUnstableLastVersion)`,
233/// filtered for `alterFeatureLevel0` below v4, and the finalized levels above 0
234/// with the image offset as their epoch. The controller listener builds that
235/// manager with `unstable.api.versions.enable`, so that flag, and not
236/// `unstable.feature.versions.enable`, caps the supported levels. The broker
237/// listener caps them with the feature flag. [`features`] holds the rules the
238/// two share.
239///
240/// Body is the flexible (v3+) `ApiVersionsResponse` shape: `error_code(i16)`,
241/// `api_keys` compact-array of `{api_key(i16), min(i16), max(i16), tagged(0)}`,
242/// `throttle_time_ms(i32)`, response-level `tagged(0)`. Per the documented Kafka
243/// asymmetry, the *response header* stays v0 (no leading tagged-fields byte) —
244/// so this is written via [`super::framing::write_response`].
245pub(super) fn api_versions_response_body(req_version: i16, view: ApiVersionsView<'_>) -> Bytes {
246    let resp = ApiVersionsResponse {
247        api_keys: advertised_api_keys(view.admin_router, view.unstable),
248        supported_features: supported_feature_keys(
249            req_version,
250            UnstableFeatureVersions::from(view.unstable == UnstableApiVersions::Enabled),
251        ),
252        finalized_features_epoch: view.finalized_features_epoch,
253        finalized_features: finalized_feature_keys(view.image),
254        ..Default::default()
255    };
256    // JVM dials at v4 (flexible); Krabka's own client at v0 (non-flexible). The
257    // codec emits the correct body shape per version: req v<=2 → non-flexible
258    // v0-shaped body, req v>=3 → flexible (compact) body. The v0 ApiVersions
259    // response HEADER asymmetry lives in the framing (`write_response` with tagged fields disabled),
260    // not here.
261    encode_body(
262        &resp,
263        req_version.clamp(0, api_versions_max_version(view.unstable)),
264    )
265}
266
267/// What a test Admin router that answers only `BrokerHeartbeat` advertises.
268#[cfg(test)]
269pub(super) const BROKER_HEARTBEAT_ONLY: &[ControllerApiVersion] = {
270    use krabka_protocol::owned::broker_heartbeat_request;
271    &[ControllerApiVersion {
272        api_key: broker_heartbeat_request::API_KEY,
273        min_version: broker_heartbeat_request::MIN_VERSION,
274        max_version: broker_heartbeat_request::MAX_VERSION,
275        released_max: Some(broker_heartbeat_request::MAX_VERSION),
276        flexible_min: broker_heartbeat_request::FLEXIBLE_MIN,
277    }]
278};
279
280#[cfg(test)]
281mod tests {
282    use {
283        super::*,
284        crate::{
285            ControllerAdminRequest, ControllerAdminRouteFuture, ControllerAdminRouter,
286            ControllerApiVersion, LATEST_PRODUCTION_METADATA_VERSION, test_support::feature_image,
287        },
288        krabka_metadata::{
289            FeatureLevelRecord, MetadataRecord,
290            metadata_version::{
291                METADATA_VERSION_FEATURE, METADATA_VERSION_MAX, METADATA_VERSION_MIN,
292            },
293        },
294        krabka_protocol::{
295            Decode,
296            owned::broker_heartbeat_request::{MAX_VERSION, MIN_VERSION},
297        },
298        std::collections::BTreeSet,
299        uuid::Uuid,
300    };
301
302    /// The view a listener with no Admin router and the default unstable
303    /// setting answers from, at finalized-features epoch
304    /// `finalized_features_epoch`.
305    fn view(
306        image: &krabka_metadata::MetadataImage,
307        finalized_features_epoch: i64,
308    ) -> ApiVersionsView<'_> {
309        ApiVersionsView {
310            image,
311            finalized_features_epoch,
312            admin_router: None,
313            unstable: UnstableApiVersions::Disabled,
314        }
315    }
316
317    #[yare::parameterized(
318        legacy_shape = { 0i16 },
319        flexible_shape = { 4i16 },
320    )]
321    fn api_versions_body_advertises_kip595_set_both_shapes(req_v: i16) {
322        use krabka_protocol::{Decode, owned::api_versions_response::ApiVersionsResponse};
323        let mut image = krabka_metadata::MetadataImage::new(Uuid::nil());
324        image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
325            name: METADATA_VERSION_FEATURE.into(),
326            level: 24,
327        }));
328        let body = super::api_versions_response_body(req_v, view(&image, 41));
329        let v = req_v.clamp(0, 4);
330        let mut cur = &body[..];
331        let resp = ApiVersionsResponse::decode(&mut cur, v).expect("decode body");
332        assert2::assert!(cur.is_empty());
333        assert2::assert!(resp.error_code == 0);
334        let keys: BTreeSet<i16> = resp.api_keys.iter().map(|k| k.api_key).collect();
335        for want in [1i16, 18, 52, 53, 54, 59, 70] {
336            assert2::assert!(keys.contains(&want));
337        }
338        // `BrokerRegistration` and `BrokerHeartbeat` reach the controller
339        // listener through the Admin router, so the native table must not
340        // advertise them on its own.
341        // `an_admin_router_contributes_its_own_api_versions` covers the
342        // other half: with a router bound, a key is advertised.
343        assert2::assert!(!keys.contains(&62i16));
344        assert2::assert!(!keys.contains(&63i16));
345        // Vote is advertised over its whole schema range, as Kafka's
346        // `ApiKeys` does.
347        let vote = resp.api_keys.iter().find(|k| k.api_key == 52).unwrap();
348        assert2::assert!(vote.min_version == 0 && vote.max_version == 2);
349        if req_v >= 3 {
350            let finalized_metadata = resp
351                .finalized_features
352                .iter()
353                .find(|feature| feature.name == "metadata.version")
354                .expect("metadata.version finalized");
355            assert2::assert!(
356                (
357                    finalized_metadata.min_version_level,
358                    finalized_metadata.max_version_level
359                ) == (24, 24)
360            );
361            assert2::assert!(resp.finalized_features_epoch == 41);
362        }
363    }
364
365    /// The controller listener advertises what a bound Admin router serves, on
366    /// top of the APIs it answers itself. `BrokerHeartbeat` is the case that
367    /// matters: a broker reads this table to learn where to send it.
368    #[test]
369    fn an_admin_router_contributes_its_own_api_versions() {
370        use krabka_protocol::owned::api_versions_response::ApiVersionsResponse;
371
372        struct HeartbeatRouter;
373        impl ControllerAdminRouter for HeartbeatRouter {
374            fn api_versions(&self) -> &[ControllerApiVersion] { super::BROKER_HEARTBEAT_ONLY }
375
376            fn route(&self, _request: ControllerAdminRequest) -> ControllerAdminRouteFuture<'_> {
377                Box::pin(async { Ok(None) })
378            }
379        }
380
381        let image = krabka_metadata::MetadataImage::new(Uuid::nil());
382        let body = super::api_versions_response_body(4, ApiVersionsView {
383            admin_router: Some(&HeartbeatRouter),
384            ..view(&image, -1)
385        });
386        let resp = ApiVersionsResponse::decode(&mut &body[..], 4).expect("decode body");
387
388        let heartbeat = resp
389            .api_keys
390            .iter()
391            .find(|key| key.api_key == 63)
392            .expect("BrokerHeartbeat advertised through the Admin router");
393        assert2::assert!(
394            (heartbeat.min_version, heartbeat.max_version) == (MIN_VERSION, MAX_VERSION,)
395        );
396    }
397
398    /// A router whose one API has an unstable last version, the shape of
399    /// Kafka's `InitProducerId` v6 (`latestVersionUnstable`).
400    struct UnstableRouter;
401
402    impl ControllerAdminRouter for UnstableRouter {
403        fn api_versions(&self) -> &[ControllerApiVersion] {
404            &[ControllerApiVersion {
405                api_key: 22,
406                min_version: 0,
407                max_version: 6,
408                released_max: Some(5),
409                flexible_min: 2,
410            }]
411        }
412
413        fn route(&self, _request: ControllerAdminRequest) -> ControllerAdminRouteFuture<'_> {
414            Box::pin(async { Ok(None) })
415        }
416    }
417
418    /// #646: an unstable last version is advertised, and accepted, only under
419    /// `unstable.api.versions.enable`, as Kafka's `ApiKeys.toApiVersion` and
420    /// `ApiKeys.isVersionEnabled` gate it.
421    #[yare::parameterized(
422        disabled = { UnstableApiVersions::Disabled, 5, true },
423        enabled = { UnstableApiVersions::Enabled, 6, false },
424    )]
425    fn unstable_last_versions_follow_unstable_api_versions_enable(
426        unstable: UnstableApiVersions,
427        advertised_max: i16,
428        v6_disabled: bool,
429    ) {
430        let api_keys = advertised_api_keys(Some(&UnstableRouter), unstable);
431        let row = api_keys
432            .iter()
433            .find(|key| key.api_key == 22)
434            .expect("router api advertised");
435        assert2::check!(
436            (row.min_version, row.max_version) == (0, advertised_max),
437            "{unstable:?}"
438        );
439        assert2::check!(
440            is_disabled_version(22, 6, Some(&UnstableRouter), unstable) == v6_disabled,
441            "{unstable:?}"
442        );
443        assert2::check!(
444            !is_disabled_version(22, 5, Some(&UnstableRouter), unstable),
445            "{unstable:?}"
446        );
447        // An out-of-range version is the decoder's business, not this
448        // gate's; an api the listener does not know is not disabled.
449        assert2::check!(!is_disabled_version(22, 7, Some(&UnstableRouter), unstable));
450        assert2::check!(!is_disabled_version(22, 6, None, unstable));
451    }
452
453    /// A router api Kafka 4.3.1 does not have -- `UnregisterController` (94),
454    /// a trunk key -- is neither advertised nor accepted while unstable api
455    /// versions are disabled, and is served whole when they are enabled.
456    #[yare::parameterized(
457        disabled = { UnstableApiVersions::Disabled, None, true },
458        enabled = { UnstableApiVersions::Enabled, Some((0, 0)), false },
459    )]
460    fn an_api_kafka_4_3_1_lacks_follows_unstable_api_versions_enable(
461        unstable: UnstableApiVersions,
462        advertised: Option<(i16, i16)>,
463        disabled: bool,
464    ) {
465        struct TrunkRouter;
466        impl ControllerAdminRouter for TrunkRouter {
467            fn api_versions(&self) -> &[ControllerApiVersion] {
468                &[ControllerApiVersion {
469                    api_key: 94,
470                    min_version: 0,
471                    max_version: 0,
472                    released_max: None,
473                    flexible_min: 0,
474                }]
475            }
476
477            fn route(&self, _request: ControllerAdminRequest) -> ControllerAdminRouteFuture<'_> {
478                Box::pin(async { Ok(None) })
479            }
480        }
481
482        let row = advertised_api_keys(Some(&TrunkRouter), unstable)
483            .into_iter()
484            .find(|key| key.api_key == 94)
485            .map(|key| (key.min_version, key.max_version));
486        assert2::check!(row == advertised, "{unstable:?}");
487        assert2::check!(
488            is_disabled_version(94, 0, Some(&TrunkRouter), unstable) == disabled,
489            "{unstable:?}"
490        );
491    }
492
493    /// The controller listener's table is strictly ascending by api key, the
494    /// router's rows merged in, as Kafka's `apisForListener` `EnumSet` orders
495    /// it.
496    #[test]
497    fn the_advertised_table_is_strictly_ascending() {
498        let api_keys = advertised_api_keys(Some(&UnstableRouter), UnstableApiVersions::Disabled);
499        assert2::assert!(
500            api_keys
501                .windows(2)
502                .all(|pair| pair[0].api_key < pair[1].api_key)
503        );
504    }
505
506    /// One row per `ApiVersions` request shape: Kafka's unsupported-version
507    /// answer, the `isValid` refusals, and the full table. A request whose
508    /// version is not served carries a body that does not decode, because
509    /// Kafka does not read it.
510    #[test]
511    fn api_versions_response_runs_kafka_request_checks() {
512        use krabka_protocol::owned::api_versions_request::ApiVersionsRequest;
513
514        /// The answer a request must get, and the version it is encoded at.
515        struct Answer {
516            version: i16,
517            response: ApiVersionsResponse,
518        }
519        struct Row {
520            label: &'static str,
521            unstable: UnstableApiVersions,
522            version: i16,
523            request: Option<ApiVersionsRequest>,
524            expected: Option<Answer>,
525        }
526        let request = |name: &str, software_version: &str, cluster_id: Option<&str>, node_id| {
527            Some(ApiVersionsRequest {
528                client_software_name: name.into(),
529                client_software_version: software_version.into(),
530                cluster_id: cluster_id.map(str::to_owned),
531                node_id,
532                ..Default::default()
533            })
534        };
535        let unsupported = |max_version| {
536            Some(Answer {
537                version: 0,
538                response: ApiVersionsResponse {
539                    error_code: API_VERSIONS_UNSUPPORTED_VERSION,
540                    api_keys: vec![ApiVersionEntry {
541                        api_key: 18,
542                        min_version: 0,
543                        max_version,
544                        ..Default::default()
545                    }],
546                    ..Default::default()
547                },
548            })
549        };
550        let (strict, trunk) = (UnstableApiVersions::Disabled, UnstableApiVersions::Enabled);
551        let invalid = |version| {
552            Some(Answer {
553                version,
554                response: ApiVersionsResponse {
555                    error_code: API_VERSIONS_INVALID_REQUEST,
556                    ..Default::default()
557                },
558            })
559        };
560        let rows = [
561            Row {
562                label: "v6",
563                unstable: strict,
564                version: 6,
565                request: None,
566                expected: unsupported(4),
567            },
568            Row {
569                label: "i16::MAX",
570                unstable: strict,
571                version: i16::MAX,
572                request: None,
573                expected: unsupported(4),
574            },
575            Row {
576                label: "negative",
577                unstable: strict,
578                version: -1,
579                request: None,
580                expected: unsupported(4),
581            },
582            Row {
583                label: "v3 empty name",
584                unstable: strict,
585                version: 3,
586                request: request("", "1.0", None, -1),
587                expected: invalid(3),
588            },
589            Row {
590                label: "v4 name with a space",
591                unstable: strict,
592                version: 4,
593                request: request("a b", "1.0", None, -1),
594                expected: invalid(4),
595            },
596            Row {
597                label: "v3 empty software version",
598                unstable: strict,
599                version: 3,
600                request: request("krabka", "", None, -1),
601                expected: invalid(3),
602            },
603            Row {
604                label: "v5 cluster id without node id",
605                unstable: trunk,
606                version: 5,
607                request: request("krabka", "1.0", Some("cluster"), -1),
608                expected: invalid(5),
609            },
610            Row {
611                label: "v5 node id without cluster id",
612                unstable: trunk,
613                version: 5,
614                request: request("krabka", "1.0", None, 7),
615                expected: invalid(5),
616            },
617            Row {
618                label: "v5 another cluster and node, no KIP-1242 check",
619                unstable: trunk,
620                version: 5,
621                request: request("krabka", "1.0", Some("other"), 8),
622                expected: None,
623            },
624            Row {
625                label: "v5 valid",
626                unstable: trunk,
627                version: 5,
628                request: request("krabka", "1.0", None, -1),
629                expected: None,
630            },
631            Row {
632                label: "v0",
633                unstable: strict,
634                version: 0,
635                request: Some(ApiVersionsRequest::default()),
636                expected: None,
637            },
638            Row {
639                label: "v6 under unstable api versions",
640                unstable: trunk,
641                version: 6,
642                request: None,
643                expected: unsupported(5),
644            },
645            Row {
646                label: "v5 is Kafka 4.3.1's unsupported version",
647                unstable: strict,
648                version: 5,
649                request: request("krabka", "1.0", None, -1),
650                expected: unsupported(4),
651            },
652        ];
653
654        let image = krabka_metadata::MetadataImage::new(Uuid::nil());
655        for row in rows {
656            let body = row.request.map_or_else(
657                || bytes::Bytes::from_static(&[0xff]),
658                |request| {
659                    let mut body = BytesMut::new();
660                    request.encode(&mut body, row.version).expect("encode");
661                    body.freeze()
662                },
663            );
664            let view = ApiVersionsView {
665                unstable: row.unstable,
666                ..view(&image, -1)
667            };
668            let answer = super::api_versions_response(row.version, &body, view).expect("answer");
669            let Answer {
670                version,
671                response: expected,
672            } = row.expected.unwrap_or_else(|| {
673                let full = super::api_versions_response_body(row.version, view);
674                Answer {
675                    version: row.version,
676                    response: ApiVersionsResponse::decode(&mut &full[..], row.version)
677                        .expect("full"),
678                }
679            });
680            let mut cur = &answer[..];
681            let decoded = ApiVersionsResponse::decode(&mut cur, version).expect("decode");
682            assert2::check!(cur.is_empty(), "{}", row.label);
683            assert2::check!(decoded == expected, "{}", row.label);
684            if expected.error_code == 0 {
685                assert2::check!(!decoded.api_keys.is_empty(), "{}", row.label);
686            }
687        }
688    }
689
690    /// A served version whose body does not decode fails the request, as
691    /// Kafka's `RequestContext.parseRequest` does.
692    #[test]
693    fn api_versions_response_refuses_a_malformed_body() {
694        let image = krabka_metadata::MetadataImage::new(Uuid::nil());
695        assert2::assert!(super::api_versions_response(3, &[0xff], view(&image, -1)).is_err());
696    }
697
698    /// #783 on the controller listener, table-driven over the issue's rows:
699    /// the supported rows follow Kafka's minimums and `alterFeatureLevel0`,
700    /// the finalized rows omit level 0 (`kraft.version` included), and the
701    /// epoch is the metadata offset the view carries.
702    #[test]
703    fn feature_fields_match_kafka_on_the_controller_listener() {
704        use krabka_protocol::owned::api_versions_response::{
705            FinalizedFeatureKey, SupportedFeatureKey,
706        };
707
708        struct Row<'a> {
709            version: i16,
710            kraft_version: u16,
711            levels: &'a [(&'a str, i16)],
712            want_supported: &'a Vec<SupportedFeatureKey>,
713            want_finalized: Vec<FinalizedFeatureKey>,
714        }
715
716        krabka_macros::supported_feature_fixture!(supported);
717        krabka_macros::supported_features_fixture!(modern_supported, supported);
718        krabka_macros::finalized_feature_fixture!(finalized);
719        // The listener's default, `unstable.api.versions.enable=false`, caps
720        // metadata.version at 4.3.1's latest production level.
721        let metadata_max = LATEST_PRODUCTION_METADATA_VERSION;
722        let legacy = vec![supported(
723            "metadata.version",
724            METADATA_VERSION_MIN,
725            metadata_max,
726        )];
727        let modern = modern_supported(
728            supported("metadata.version", METADATA_VERSION_MIN, metadata_max),
729            1,
730        );
731        let rows = [
732            Row {
733                version: 3,
734                kraft_version: 0,
735                levels: &[("metadata.version", 25)],
736                want_supported: &legacy,
737                want_finalized: vec![finalized("metadata.version", 25)],
738            },
739            Row {
740                version: 4,
741                kraft_version: 0,
742                levels: &[("metadata.version", 25), ("group.version", 1)],
743                want_supported: &modern,
744                want_finalized: vec![
745                    finalized("group.version", 1),
746                    finalized("metadata.version", 25),
747                ],
748            },
749            Row {
750                version: 4,
751                kraft_version: 1,
752                levels: &[("metadata.version", 25)],
753                want_supported: &modern,
754                want_finalized: vec![
755                    finalized("kraft.version", 1),
756                    finalized("metadata.version", 25),
757                ],
758            },
759            Row {
760                version: 4,
761                kraft_version: 1,
762                levels: &[("metadata.version", 30), ("krabka.version", 1)],
763                want_supported: &modern,
764                want_finalized: vec![
765                    finalized("krabka.version", 1),
766                    finalized("kraft.version", 1),
767                    finalized("metadata.version", 30),
768                ],
769            },
770        ];
771        for Row {
772            version,
773            kraft_version,
774            levels,
775            want_supported,
776            want_finalized,
777        } in rows
778        {
779            let image = feature_image(kraft_version, levels);
780            let body = super::api_versions_response_body(version, view(&image, 1234));
781            let resp = ApiVersionsResponse::decode(&mut &body[..], version).expect("decode");
782            assert2::check!(
783                (
784                    &resp.supported_features,
785                    &resp.finalized_features,
786                    resp.finalized_features_epoch
787                ) == (want_supported, &want_finalized, 1234),
788                "v{version}, kraft.version {kraft_version}"
789            );
790        }
791    }
792
793    /// #1344: before the first feature delta Kafka's controller answers from
794    /// the initial state of `FeaturesPublisher`,
795    /// `FinalizedFeatures.fromKRaftVersion(MINIMUM_VERSION)`, which finalizes
796    /// `metadata.version` at its minimum with epoch -1. `ControllerServer`
797    /// adds `kraft.version` through `setFinalizedLevel`, which drops level 0.
798    #[yare::parameterized(
799        kraft_version_0 = { 0, &[("metadata.version", METADATA_VERSION_MIN)] },
800        kraft_version_1 = {
801            1,
802            &[("kraft.version", 1), ("metadata.version", METADATA_VERSION_MIN)],
803        },
804    )]
805    fn a_fresh_controller_advertises_kafkas_minimum_metadata_version(
806        kraft_version: u16,
807        want: &[(&str, i16)],
808    ) {
809        krabka_macros::finalized_feature_fixture!(finalized);
810        let image = feature_image(kraft_version, &[]);
811        let body = super::api_versions_response_body(4, view(&image, -1));
812        let resp = ApiVersionsResponse::decode(&mut &body[..], 4).expect("decode");
813        let want_finalized: Vec<_> = want
814            .iter()
815            .map(|(name, level)| finalized(name, *level))
816            .collect();
817        assert2::check!(
818            (resp.finalized_features, resp.finalized_features_epoch) == (want_finalized, -1)
819        );
820    }
821
822    /// Kafka's `ControllerServer` builds the listener's
823    /// `SimpleApiVersionManager` with `unstable.api.versions.enable`, so that
824    /// flag alone caps the supported `metadata.version` here. With only
825    /// `unstable.feature.versions.enable` set the listener stays at the latest
826    /// production level, and with only the API flag set it advertises the
827    /// highest level krabka knows. The listener takes no feature flag at all:
828    /// its view has none.
829    #[yare::parameterized(
830        disabled = { UnstableApiVersions::Disabled, LATEST_PRODUCTION_METADATA_VERSION },
831        enabled = {
832            UnstableApiVersions::Enabled,
833            METADATA_VERSION_MAX,
834        },
835    )]
836    fn the_controller_listener_caps_supported_features_with_the_api_flag(
837        unstable: UnstableApiVersions,
838        want_max: i16,
839    ) {
840        let image = krabka_metadata::MetadataImage::new(Uuid::nil());
841        let view = ApiVersionsView {
842            unstable,
843            ..view(&image, 0)
844        };
845        let body = super::api_versions_response_body(4, view);
846        let resp = ApiVersionsResponse::decode(&mut &body[..], 4).expect("decode");
847        let metadata_version = resp
848            .supported_features
849            .iter()
850            .find(|feature| feature.name == "metadata.version")
851            .expect("metadata.version is supported");
852        assert2::check!(
853            metadata_version.max_version == want_max,
854            "unstable.api.versions.enable {unstable:?}"
855        );
856    }
857}
krabka_raft/server/api_versions.rs

krabka_raft/server/
api_versions.rs
