krabka_raft/server/api_versions/
features.rs

1//! The KIP-584 feature rows of an `ApiVersions` response, built by one set of
2//! rules for the controller listener and the broker listener. The two pass it
3//! different caps, as Kafka does: the controller listener's
4//! `SimpleApiVersionManager` takes `unstable.api.versions.enable`, and the
5//! broker's `BrokerFeatures.createDefault` takes
6//! `unstable.feature.versions.enable`.
7//!
8//! The rules are Kafka 4.3.1's:
9//!
10//! - `BrokerFeatures.defaultSupportedFeatures` advertises each feature from its
11//!   minimum production level, which is 0 for every feature except
12//!   `metadata.version`, whose minimum is `MetadataVersion.MINIMUM_VERSION`. A
13//!   feature whose maximum is 0 is left out.
14//! - `ApiVersionsResponse.maybeFilterSupportedFeatureKeys` omits every
15//!   supported feature whose minimum is 0 when the request version is below 4
16//!   (`alterFeatureLevel0`, KAFKA-17492): older clients fail to deserialize a
17//!   zero minimum.
18//! - `ApiVersionsResponse.createFinalizedFeatureKeys` omits every finalized
19//!   level 0, and `KRaftMetadataCache.features` adds `kraft.version` only when
20//!   its level is above 0.
21//! - The finalized-features epoch is the offset of the last record the
22//!   metadata image contains, `image.highestOffsetAndEpoch().offset()`.
23
24use {
25    crate::{UnstableFeatureVersions, supported_feature_range},
26    krabka_metadata::metadata_version::{
27        KRAFT_VERSION_FEATURE, METADATA_VERSION_FEATURE, METADATA_VERSION_MIN,
28    },
29    krabka_protocol::owned::api_versions_response::{FinalizedFeatureKey, SupportedFeatureKey},
30};
31
32/// First `ApiVersions` version whose JVM client reads a zero feature minimum.
33/// Below it Kafka sets `alterFeatureLevel0`.
34const ZERO_MINIMUM_API_VERSION: i16 = 4;
35
36/// The supported-feature row for one feature at `api_version`, or `None` when
37/// Kafka leaves the feature out of the response.
38#[must_use]
39pub fn supported_feature_key(
40    name: &str,
41    min_version: i16,
42    max_version: i16,
43    api_version: i16,
44) -> Option<SupportedFeatureKey> {
45    let unregistrable = max_version == 0;
46    let alter_level_zero = api_version < ZERO_MINIMUM_API_VERSION && min_version == 0;
47    (!unregistrable && !alter_level_zero).then(|| SupportedFeatureKey {
48        name: name.to_owned(),
49        min_version,
50        max_version,
51        ..Default::default()
52    })
53}
54
55/// Every supported-feature row of the `krabka_metadata` feature registry at
56/// `api_version`, each feature capped as `unstable` caps it
57/// ([`crate::supported_feature_range`]).
58#[must_use]
59pub fn supported_feature_keys(
60    api_version: i16,
61    unstable: UnstableFeatureVersions,
62) -> Vec<SupportedFeatureKey> {
63    krabka_metadata::feature_registry()
64        .iter()
65        .filter_map(|feature| {
66            let (min_version, max_version) = supported_feature_range(*feature, unstable);
67            supported_feature_key(feature.name(), min_version, max_version, api_version)
68        })
69        .collect()
70}
71
72/// The finalized-feature rows of `image`, in name order: every finalized
73/// level above 0, `metadata.version` at Kafka's minimum before the first
74/// feature delta, and `kraft.version` when its level is above 0.
75#[must_use]
76pub fn finalized_feature_keys(image: &krabka_metadata::MetadataImage) -> Vec<FinalizedFeatureKey> {
77    let mut levels = image.finalized_features().clone();
78    // Kafka's FeaturesPublisher starts with FinalizedFeatures.fromKRaftVersion
79    // (MetadataVersion.MINIMUM_VERSION) and keeps that until a feature delta
80    // arrives. An empty Krabka image represents that same startup state.
81    levels
82        .entry(METADATA_VERSION_FEATURE.to_owned())
83        .or_insert(METADATA_VERSION_MIN);
84    levels.insert(
85        KRAFT_VERSION_FEATURE.to_owned(),
86        i16::try_from(image.kraft_version()).unwrap_or(i16::MAX),
87    );
88    levels
89        .into_iter()
90        .filter(|(_, level)| *level != 0)
91        .map(|(name, level)| FinalizedFeatureKey {
92            name,
93            // Kafka reports the finalized level as both bounds.
94            max_version_level: level,
95            min_version_level: level,
96            ..Default::default()
97        })
98        .collect()
99}
100
101#[cfg(test)]
102mod tests {
103    use {
104        super::*,
105        assert2::check,
106        krabka_metadata::metadata_version::{METADATA_VERSION_MAX, METADATA_VERSION_MIN},
107    };
108
109    krabka_macros::supported_feature_fixture!(supported);
110    krabka_macros::supported_features_fixture!(modern_supported, supported);
111
112    krabka_macros::finalized_feature_fixture!(finalized);
113
114    use crate::test_support::feature_image as image;
115
116    /// #783's table: below v4 only the features with a non-zero minimum stay;
117    /// from v4 every feature is advertised from Kafka's minimum production
118    /// level. #784: `metadata.version` tops out at 4.3.1's latest production
119    /// level, 30, unless `unstable.feature.versions.enable` is set, as
120    /// `BrokerFeatures.defaultSupportedFeatures` caps it.
121    #[yare::parameterized(
122        stable_v0 = {
123            0,
124            crate::UnstableFeatureVersions::Disabled,
125            vec![supported("metadata.version", METADATA_VERSION_MIN, 30)],
126        },
127        stable_v3 = {
128            3,
129            crate::UnstableFeatureVersions::Disabled,
130            vec![supported("metadata.version", METADATA_VERSION_MIN, 30)],
131        },
132        stable_v4 = {
133            4,
134            crate::UnstableFeatureVersions::Disabled,
135            modern_supported(supported("metadata.version", METADATA_VERSION_MIN, 30), 1),
136        },
137        stable_v5 = {
138            5,
139            crate::UnstableFeatureVersions::Disabled,
140            modern_supported(supported("metadata.version", METADATA_VERSION_MIN, 30), 1),
141        },
142        unstable_v0 = {
143            0,
144            crate::UnstableFeatureVersions::Enabled,
145            vec![supported("metadata.version", METADATA_VERSION_MIN, METADATA_VERSION_MAX)],
146        },
147        unstable_v3 = {
148            3,
149            crate::UnstableFeatureVersions::Enabled,
150            vec![supported("metadata.version", METADATA_VERSION_MIN, METADATA_VERSION_MAX)],
151        },
152        unstable_v4 = {
153            4,
154            crate::UnstableFeatureVersions::Enabled,
155            modern_supported(supported("metadata.version", METADATA_VERSION_MIN, METADATA_VERSION_MAX), 2),
156        },
157        unstable_v5 = {
158            5,
159            crate::UnstableFeatureVersions::Enabled,
160            modern_supported(supported("metadata.version", METADATA_VERSION_MIN, METADATA_VERSION_MAX), 2),
161        },
162    )]
163    fn supported_features_follow_kafkas_minimums_and_alter_level_zero(
164        api_version: i16,
165        unstable: UnstableFeatureVersions,
166        expected: Vec<SupportedFeatureKey>,
167    ) {
168        // `share.version` 2 (KIP-1191) is trunk's, so only the flag advertises it.
169        check!(
170            supported_feature_keys(api_version, unstable) == expected,
171            "v{api_version} {unstable:?}"
172        );
173    }
174
175    /// A feature whose maximum is 0 cannot be finalized above 0, and Kafka
176    /// does not advertise it at any version.
177    #[yare::parameterized(
178        v0 = { 0 },
179        v3 = { 3 },
180        v4 = { 4 },
181        v5 = { 5 },
182    )]
183    fn a_feature_with_maximum_zero_is_never_advertised(api_version: i16) {
184        check!(supported_feature_key("x.version", 0, 0, api_version) == None);
185        check!(supported_feature_key("x.version", 1, 2, 3) == Some(supported("x.version", 1, 2)));
186    }
187
188    /// Level-0 rows are never finalized rows, `kraft.version` included.
189    #[yare::parameterized(
190        empty_levels = { 0, &[][..], vec![finalized("metadata.version", METADATA_VERSION_MIN)] },
191        group_level = {
192            0,
193            &[("metadata.version", 25), ("group.version", 1)][..],
194            vec![finalized("group.version", 1), finalized("metadata.version", 25)],
195        },
196        kraft_level = {
197            1,
198            &[("metadata.version", 25)][..],
199            vec![finalized("kraft.version", 1), finalized("metadata.version", 25)],
200        },
201        zero_group_ignored = {
202            0,
203            &[("metadata.version", 25), ("group.version", 1), ("group.version", 0)][..],
204            vec![finalized("metadata.version", 25)],
205        },
206        krabka_version = {
207            1,
208            &[("metadata.version", 30), ("krabka.version", 1)][..],
209            vec![
210                finalized("krabka.version", 1),
211                finalized("kraft.version", 1),
212                finalized("metadata.version", 30),
213            ],
214        },
215        zero_krabka_ignored = {
216            0,
217            &[("metadata.version", 30), ("krabka.version", 0)][..],
218            vec![finalized("metadata.version", 30)],
219        },
220    )]
221    fn finalized_features_omit_level_zero(
222        kraft_version: u16,
223        levels: &[(&str, i16)],
224        expected: Vec<FinalizedFeatureKey>,
225    ) {
226        check!(
227            finalized_feature_keys(&image(kraft_version, levels)) == expected,
228            "kraft.version {kraft_version}, {levels:?}"
229        );
230    }
231}
krabka_raft/server/api_versions/features.rs

krabka_raft/server/api_versions/
features.rs
