krabka_format/format/
features.rs

1//! Resolution of the bootstrap `metadata.version` and of the KIP-1022 feature
2//! levels a format finalizes.
3//!
4//! `--release-version` and `--feature` both decide what the seed
5//! `FeatureLevelRecord`s say, and the two interact: one sets the base release
6//! and the other overrides individual features, except for `metadata.version`
7//! where they conflict. The rules and the validation `kafka-storage format`
8//! performs live here, apart from the flag definitions they read.
9
10use {
11    krabka_metadata::{
12        FeatureLevelRecord,
13        metadata_version::{self, KRAFT_VERSION_FEATURE, from_version_string, is_supported_level},
14    },
15    krabka_raft::UnstableFeatureVersions,
16    std::collections::BTreeMap,
17};
18
19/// What `--release-version` and `--feature` resolve to.
20#[derive(Debug, PartialEq, Eq)]
21pub(super) struct ResolvedFeatures {
22    /// The `metadata.version` level the cluster bootstraps at.
23    pub(super) metadata_version: i16,
24    /// The levels `--feature` set, by feature name.
25    pub(super) overrides: BTreeMap<String, i16>,
26}
27
28/// The highest `metadata.version` this format accepts under `unstable`:
29/// Kafka 4.3's latest production level, or with unstable feature versions the
30/// highest level krabka knows (Kafka's `MetadataVersion.latestTesting()`).
31fn metadata_version_ceiling(unstable: UnstableFeatureVersions) -> i16 {
32    krabka_metadata::feature(metadata_version::METADATA_VERSION_FEATURE)
33        .map_or(LATEST_PRODUCTION_METADATA_VERSION, |feature| {
34            krabka_raft::supported_feature_range(feature, unstable).1
35        })
36}
37
38/// The highest `metadata.version` level the table of the emulated Kafka knows:
39/// Kafka's `MetadataVersion.latestTesting()`. It is one past the latest
40/// production level in 4.3.1's table (`4.4-IV0`, 31), and the highest level
41/// krabka knows under `unstable`. A level above it does not exist. A level
42/// between [`metadata_version_ceiling`] and it exists but is not yet stable.
43fn metadata_version_latest_testing(unstable: UnstableFeatureVersions) -> i16 {
44    match unstable {
45        UnstableFeatureVersions::Disabled => LATEST_PRODUCTION_METADATA_VERSION + 1,
46        UnstableFeatureVersions::Enabled => metadata_version_ceiling(unstable),
47    }
48}
49
50/// The lookup key Kafka's `MetadataVersion.fromVersionString` derives from a
51/// release string: the first two dot-separated segments when there are more
52/// than two, so `4.3.1` and `4.3.x` both look up `4.3`, and the whole string
53/// otherwise. Java's `String.split` drops trailing empty segments, so `4.3.1.`
54/// looks up `4.3` but `4.3.` is looked up as it stands.
55fn kafka_version_key(s: &str) -> &str {
56    let significant = s.trim_end_matches('.');
57    significant
58        .match_indices('.')
59        .nth(1)
60        .and_then(|(end, _)| significant.get(..end))
61        .unwrap_or(s)
62}
63
64/// Map a release string to a supported `metadata.version` feature level, as
65/// Kafka's `MetadataVersion.fromVersionString(versionString,
66/// unstableFeatureVersionsEnabled)` does: an unknown release, or one past the
67/// latest production level while unstable feature versions are off, is
68/// refused with Kafka's message, which lists the releases that are accepted.
69///
70/// Kafka looks the string up by its first two segments and does not trim it.
71/// The `krabka_metadata` lookup does trim, so a key with surrounding
72/// whitespace is refused here, as an unknown release, before that lookup runs.
73fn resolve_release_level(s: &str, unstable: UnstableFeatureVersions) -> Result<i16, String> {
74    let ceiling = metadata_version_ceiling(unstable);
75    Some(kafka_version_key(s))
76        .filter(|key| key.trim() == *key)
77        .and_then(from_version_string)
78        .map(metadata_version::MetadataVersion::feature_level)
79        .filter(|level| is_supported_level(*level) && *level <= ceiling)
80        .ok_or_else(|| {
81            let supported: Vec<&str> = (metadata_version::METADATA_VERSION_MIN..=ceiling)
82                .filter_map(metadata_version::from_feature_level)
83                .map(metadata_version::MetadataVersion::ivn)
84                .collect();
85            format!(
86                "Unknown metadata.version '{s}'. Supported metadata.version are: {}",
87                supported.join(", ")
88            )
89        })
90}
91
92/// Parse one `--feature NAME=LEVEL` spec into the feature and its level.
93pub(super) fn parse_feature_spec(s: &str) -> Result<FeatureLevelRecord, String> {
94    let (name, level) = s
95        .split_once('=')
96        .ok_or("--feature must be NAME=LEVEL, e.g. transaction.version=2")?;
97    let name = name.trim();
98    if name.is_empty() {
99        return Err("feature name must not be empty".into());
100    }
101    let level: i16 = level
102        .trim()
103        .parse()
104        .map_err(|e| format!("feature level: {e}"))?;
105    Ok(FeatureLevelRecord {
106        name: name.to_owned(),
107        level,
108    })
109}
110
111/// The `metadata.version` a cluster formats at when neither `--release-version`
112/// nor `--feature metadata.version` names one: Kafka 4.3's
113/// `MetadataVersion.LATEST_PRODUCTION`, `4.3-IV0`.
114///
115/// The levels above it that the feature table carries, `4.4-IV0` to
116/// `4.5-IV0`, are Kafka trunk's unstable versions. A stock 4.3 node or tool
117/// does not know them, so a format names one only under
118/// `--unstable-feature-versions-enable`, Kafka's
119/// `unstable.feature.versions.enable`.
120pub const LATEST_PRODUCTION_METADATA_VERSION: i16 = krabka_raft::LATEST_PRODUCTION_METADATA_VERSION;
121
122/// Kafka's message for a `--feature` name its formatter does not support. The
123/// list is every registered feature but `metadata.version`, by name, sorted:
124/// Kafka's `Feature.PRODUCTION_FEATURES` and krabka's `krabka.version`.
125fn unsupported_feature(name: &str) -> String {
126    let mut supported: Vec<&str> = krabka_metadata::feature_registry()
127        .iter()
128        .map(|feat| feat.name())
129        .filter(|feature| *feature != metadata_version::METADATA_VERSION_FEATURE)
130        .collect();
131    supported.sort_unstable();
132    format!(
133        "Unsupported feature: {name}. Supported features are: {}",
134        supported.join(", ")
135    )
136}
137
138/// Kafka's `MetadataVersion.fromFeatureLevel` and then
139/// `Formatter.verifyReleaseVersion`, for `--feature metadata.version=LEVEL`:
140/// a level outside the emulated table does not exist, and a level that exists
141/// but is past the latest production one is `not yet stable` unless `unstable`
142/// allows it.
143fn verify_metadata_version_level(
144    level: i16,
145    unstable: UnstableFeatureVersions,
146) -> Result<(), String> {
147    use krabka_metadata::metadata_version::{METADATA_VERSION_MIN, from_feature_level};
148
149    let latest_testing = metadata_version_latest_testing(unstable);
150    let version = from_feature_level(level)
151        .filter(|_| (METADATA_VERSION_MIN..=latest_testing).contains(&level))
152        .ok_or_else(|| {
153            format!(
154                "No MetadataVersion with feature level {level}. \
155                 Valid feature levels are from {METADATA_VERSION_MIN} to {latest_testing}."
156            )
157        })?;
158    if level > metadata_version_ceiling(unstable) {
159        return Err(format!(
160            "metadata.version {} is not yet stable.",
161            version.ivn()
162        ));
163    }
164    Ok(())
165}
166
167/// Resolve `krabka format`'s KIP-1022 feature flags into the bootstrap
168/// `metadata.version` level and the per-feature override map, applying the
169/// validation `kafka-storage format` performs:
170///
171/// - every `--feature` names a feature Kafka's formatter supports, which is
172///   every registered one except `metadata.version`, and a level that feature
173///   defines (else reject);
174/// - `--feature metadata.version=X` conflicts with `--release-version`, must
175///   name a level in the emulated Kafka's table, and a level past the latest
176///   production one is refused as `not yet stable` unless `unstable` allows it
177///   (`Formatter.verifyReleaseVersion`);
178/// - `bootstrap_mv` = `--feature metadata.version` if set, else
179///   `--release-version`, else [`LATEST_PRODUCTION_METADATA_VERSION`], or the
180///   highest level krabka knows under `unstable` (`latestTesting`);
181/// - the fully-resolved feature set satisfies every KIP-1022 dependency.
182///
183/// The checks run in Kafka's order, and the error texts are Kafka's:
184/// `kafka-storage format` parses `--release-version` first, resolves the
185/// release (`calculateEffectiveReleaseVersion`), and only then checks every
186/// feature name and level (`calculateEffectiveFeatureLevels`).
187pub(super) fn resolve_format_features(
188    release_version: Option<&str>,
189    features: &[FeatureLevelRecord],
190    unstable: UnstableFeatureVersions,
191) -> Result<ResolvedFeatures, String> {
192    use krabka_metadata::metadata_version::METADATA_VERSION_FEATURE;
193
194    let release_level = release_version
195        .map(|rv| resolve_release_level(rv, unstable))
196        .transpose()?;
197    let mut feature_mv: Option<i16> = None;
198    for FeatureLevelRecord { level, .. } in features
199        .iter()
200        .filter(|feature| feature.name == METADATA_VERSION_FEATURE)
201    {
202        if release_version.is_some() {
203            return Err(
204                "Use --release-version instead of --feature metadata.version=X to avoid ambiguity."
205                    .into(),
206            );
207        }
208        verify_metadata_version_level(*level, unstable)?;
209        feature_mv = Some(*level);
210    }
211    let bootstrap_mv = feature_mv
212        .or(release_level)
213        .unwrap_or_else(|| metadata_version_ceiling(unstable));
214
215    // Every name is checked before any level, as `calculateEffectiveFeatureLevels`
216    // does. `metadata.version` is not one of Kafka's `supportedFeatures`; the
217    // release checks above own it.
218    if let Some(FeatureLevelRecord { name, .. }) = features.iter().find(|feature| {
219        feature.name != METADATA_VERSION_FEATURE
220            && krabka_metadata::feature(&feature.name).is_none()
221    }) {
222        return Err(unsupported_feature(name));
223    }
224
225    let mut overrides: BTreeMap<String, i16> = BTreeMap::new();
226    for FeatureLevelRecord { name, level } in features {
227        // KIP-853 persists kraft.version as a raft control record, never as a
228        // FeatureLevelRecord. Its mode-specific validation happens separately.
229        if name == KRAFT_VERSION_FEATURE {
230            continue;
231        }
232        // A level that Kafka has not released is a level of trunk, so the
233        // range is the one this node supports under `unstable`. The name was
234        // checked above; `metadata.version` is checked by the release checks.
235        if let Some((min, max)) = krabka_metadata::feature(name)
236            .filter(|_| name != METADATA_VERSION_FEATURE)
237            .map(|feat| krabka_raft::supported_feature_range(feat, unstable))
238            && !(min..=max).contains(level)
239        {
240            return Err(format!("No feature:{name} with feature level {level}"));
241        }
242        if overrides.insert(name.clone(), *level).is_some() {
243            return Err(format!("feature {name} specified more than once"));
244        }
245    }
246
247    // KIP-1022 dependency validation over the fully-resolved feature set
248    // (every registered feature at its override-or-default level).
249    let resolved: BTreeMap<String, i16> = krabka_metadata::feature_registry()
250        .iter()
251        .map(|f| {
252            let level = overrides
253                .get(f.name())
254                .copied()
255                .unwrap_or_else(|| f.default_level(bootstrap_mv));
256            (f.name().to_owned(), level)
257        })
258        .collect();
259    krabka_metadata::validate_feature_dependencies(&resolved)?;
260
261    Ok(ResolvedFeatures {
262        metadata_version: bootstrap_mv,
263        overrides,
264    })
265}
266
267#[cfg(test)]
268mod tests {
269
270    use {
271        super::*,
272        assert2::check,
273        krabka_metadata::{MetadataRecord, metadata_version},
274    };
275
276    /// `unstable.feature.versions.enable` at Kafka's default.
277    const STRICT: UnstableFeatureVersions = UnstableFeatureVersions::Disabled;
278
279    /// A `--feature` spec.
280    fn spec(name: &str, level: i16) -> FeatureLevelRecord {
281        FeatureLevelRecord {
282            name: name.to_owned(),
283            level,
284        }
285    }
286
287    /// A level equal to the supported minimum is in range. Every other case
288    /// sits strictly inside the range or well outside it, so relaxing the
289    /// guard from `<` to `<=` -- which rejects the minimum itself -- changed
290    /// nothing any test looked at.
291    #[test]
292    fn resolve_features_accepts_a_level_at_the_supported_minimum() {
293        // group.version supports 0..=1; metadata.version 7..=25.
294        check!(resolve_format_features(None, &[spec("group.version", 0)], STRICT).is_ok());
295        check!(resolve_format_features(None, &[spec("metadata.version", 7)], STRICT).is_ok());
296    }
297
298    #[yare::parameterized(
299        kafka_4_0 = { "4.0", Some(25) },
300        kafka_3_7_iv4 = { "3.7-IV4", Some(19) },
301        below_supported_minimum = { "2.8", None },
302        unknown_release = { "9.9-IV0", None },
303    )]
304    fn release_version_maps_to_feature_level(input: &str, want: Option<i16>) {
305        assert2::assert!(resolve_release_level(input, STRICT).ok() == want);
306    }
307
308    #[test]
309    fn bootstrap_seeds_every_nonzero_feature_at_release_default() {
310        let bootstrap_mv = metadata_version::from_version_string("4.0")
311            .unwrap()
312            .feature_level();
313        // Exercises the exact helper `run()` uses, so it tracks the registry
314        // as features are added in later tasks. Features whose release default
315        // is 0 are omitted (KIP-1022: level 0 = absent = disabled), matching
316        // `kafka-storage format`.
317        let records = krabka_metadata::bootstrap_feature_records(bootstrap_mv);
318        for feat in krabka_metadata::feature_registry() {
319            let found = records.iter().find_map(|r| match r {
320                MetadataRecord::V1FeatureLevel(f) if f.name == feat.name() => Some(f.level),
321                _ => None,
322            });
323            let expected = feat.default_level(bootstrap_mv);
324            if expected > 0 {
325                assert2::assert!(found == Some(expected));
326            } else {
327                assert2::assert!(found.is_none());
328            }
329        }
330    }
331
332    // The no-flag default path (`LATEST_PRODUCTION_METADATA_VERSION`) is
333    // covered end-to-end by `format_smoke.rs`, which formats without
334    // `--release-version` and asserts the FeatureLevel record is present.
335    #[test]
336    fn short_release_string_resolves_to_its_last_iv() {
337        assert2::assert!(resolve_release_level("4.0", STRICT).unwrap() == 25);
338    }
339
340    #[test]
341    fn parse_feature_spec_happy_path() {
342        assert2::assert!(
343            parse_feature_spec("group.version=1").unwrap() == spec("group.version", 1)
344        );
345        assert2::assert!(
346            parse_feature_spec("metadata.version=20").unwrap() == spec("metadata.version", 20)
347        );
348    }
349
350    #[yare::parameterized(
351        missing_separator = { "noequals" },
352        non_integer_level = { "group.version=abc" },
353        empty_level = { "group.version=" },
354        empty_name = { "=1" },
355    )]
356    fn parse_feature_spec_error_branches(bad: &str) {
357        assert2::assert!(parse_feature_spec(bad).is_err());
358    }
359
360    #[test]
361    fn resolve_features_defaults_bootstrap_mv_to_latest_production() {
362        // No --release-version, no metadata.version override → bootstrap at
363        // 4.3-IV0, not at trunk's unstable levels; an explicit non-metadata
364        // feature becomes an override.
365        let ResolvedFeatures {
366            metadata_version: mv,
367            overrides: ov,
368        } = resolve_format_features(None, &[spec("group.version", 1)], STRICT).expect("resolve");
369        assert2::assert!(mv == LATEST_PRODUCTION_METADATA_VERSION);
370        assert2::assert!(ov.get("group.version") == Some(&1));
371    }
372
373    #[test]
374    fn resolve_features_metadata_version_feature_sets_bootstrap_mv() {
375        let ResolvedFeatures {
376            metadata_version: mv,
377            overrides: ov,
378        } = resolve_format_features(None, &[spec("metadata.version", 20)], STRICT)
379            .expect("resolve");
380        assert2::assert!(mv == 20);
381        assert2::assert!(ov.get("metadata.version") == Some(&20));
382    }
383
384    #[test]
385    fn resolve_features_release_version_sets_bootstrap_mv() {
386        let ResolvedFeatures {
387            metadata_version: mv,
388            overrides: ov,
389        } = resolve_format_features(Some("4.0-IV0"), &[], STRICT).expect("resolve");
390        assert2::assert!(mv == 22);
391        assert2::assert!(ov.is_empty());
392    }
393
394    #[test]
395    fn resolve_features_release_and_feature_combine() {
396        // --release-version sets the base; a non-metadata --feature overrides it.
397        let ResolvedFeatures {
398            metadata_version: mv,
399            overrides: ov,
400        } = resolve_format_features(Some("4.0-IV0"), &[spec("transaction.version", 2)], STRICT)
401            .expect("resolve");
402        assert2::assert!(mv == 22);
403        assert2::assert!(ov.get("transaction.version") == Some(&2));
404    }
405
406    /// Kafka's `Feature.PRODUCTION_FEATURES` names and krabka's own
407    /// `krabka.version`, sorted: what a formatter that meets an unknown
408    /// `--feature` name lists. It has no `metadata.version`.
409    const SUPPORTED_FEATURES: &str = "eligible.leader.replicas.version, group.version, \
410                                      krabka.version, kraft.version, share.version, \
411                                      streams.version, transaction.version";
412
413    /// The releases `MetadataVersion.metadataVersionsToString` lists while
414    /// unstable feature versions are off.
415    const STRICT_RELEASES: &str = "3.3-IV3, 3.4-IV0, 3.5-IV0, 3.5-IV1, 3.5-IV2, 3.6-IV0, \
416                                   3.6-IV1, 3.6-IV2, 3.7-IV0, 3.7-IV1, 3.7-IV2, 3.7-IV3, \
417                                   3.7-IV4, 3.8-IV0, 3.9-IV0, 4.0-IV0, 4.0-IV1, 4.0-IV2, \
418                                   4.0-IV3, 4.1-IV0, 4.1-IV1, 4.2-IV0, 4.2-IV1, 4.3-IV0";
419
420    fn unknown_release(release: &str) -> String {
421        format!(
422            "Unknown metadata.version '{release}'. Supported metadata.version are: \
423             {STRICT_RELEASES}"
424        )
425    }
426
427    /// `kafka-storage format` maps `--release-version` through
428    /// `MetadataVersion.fromVersionString`, which looks the string up by its
429    /// first two dot-separated segments and does not trim it.
430    #[yare::parameterized(
431        production_release = { "4.3", STRICT, Some(30) },
432        three_segment_release = { "4.3.1", STRICT, Some(30) },
433        nonnumeric_suffix = { "4.3.x", STRICT, Some(30) },
434        extra_segments = { "4.3.1.7", STRICT, Some(30) },
435        old_release = { "4.0.0", STRICT, Some(25) },
436        kafka_example = { "3.8.1", STRICT, Some(20) },
437        iv_suffix_and_extra = { "4.3-IV0.9", STRICT, Some(30) },
438        old_iv_suffix_and_extra = { "3.7-IV4.1", STRICT, Some(19) },
439        trunk_release_enabled = { "4.4.1", UnstableFeatureVersions::Enabled, Some(33) },
440        trunk_release_disabled = { "4.4.1", STRICT, None },
441        trunk_iv_disabled = { "4.4-IV0.1", STRICT, None },
442        pre_minimum_release = { "2.8.1", STRICT, None },
443        trailing_extra_segment = { "4.3.1.", STRICT, Some(30) },
444        trailing_empty_segment = { "4.3.", STRICT, None },
445        empty_middle_segment = { "4..3", STRICT, None },
446        missing_minor = { "4", STRICT, None },
447        empty = { "", STRICT, None },
448        whitespace_after_extra_segment = { "4.3.1 ", STRICT, Some(30) },
449        whitespace_before_release = { " 4.3", STRICT, None },
450        whitespace_before_extra_segment = { " 4.3.1", STRICT, None },
451        whitespace_after_release = { "4.3 ", STRICT, None },
452        tab_after_release = { "4.3\t", STRICT, None },
453        whitespace_after_iv = { "4.3-IV0 ", STRICT, None },
454    )]
455    fn release_versions_are_looked_up_by_their_first_two_segments(
456        release: &str,
457        unstable: UnstableFeatureVersions,
458        want: Option<i16>,
459    ) {
460        let want = want.ok_or_else(|| unknown_release(release));
461        assert2::check!(
462            resolve_release_level(release, unstable) == want,
463            "{release:?} {unstable:?}"
464        );
465    }
466
467    /// Kafka 4.3.1's `kafka-storage format` messages, word for word: the
468    /// unknown-feature list is its production features and `krabka.version`
469    /// in name order, and a level a feature does not define is
470    /// `Feature.fromFeatureLevel`'s refusal.
471    #[yare::parameterized(
472        unknown_feature = {
473            "bogus.version", 1,
474            "Unsupported feature: bogus.version. Supported features are: \
475             eligible.leader.replicas.version, group.version, krabka.version, \
476             kraft.version, share.version, streams.version, transaction.version"
477        },
478        transaction_level = {
479            "transaction.version",
480            9,
481            "No feature:transaction.version with feature level 9",
482        },
483        group_level = { "group.version", 5, "No feature:group.version with feature level 5" },
484    )]
485    fn feature_refusals_use_kafkas_messages(feature: &str, level: i16, expected: &str) {
486        let err = resolve_format_features(None, &[spec(feature, level)], STRICT).unwrap_err();
487        assert2::assert!(err == expected, "{feature:?}");
488    }
489
490    #[yare::parameterized(
491        group_version_above_maximum = { "group.version", 5 },
492        metadata_version_above_maximum = { "metadata.version", 99 },
493        metadata_version_below_minimum = { "metadata.version", 1 },
494    )]
495    fn resolve_features_rejects_out_of_range_level(name: &str, level: i16) {
496        assert2::assert!(resolve_format_features(None, &[spec(name, level)], STRICT).is_err());
497    }
498
499    /// `--release-version 4.3.1` reaches the bootstrap `metadata.version`
500    /// through the whole feature resolution, not only the release lookup.
501    #[test]
502    fn a_three_segment_release_version_formats_at_its_release() {
503        check!(
504            resolve_format_features(Some("4.3.1"), &[], STRICT)
505                == Ok(ResolvedFeatures {
506                    metadata_version: 30,
507                    overrides: BTreeMap::new(),
508                })
509        );
510    }
511
512    /// Every rejection `kafka-storage format` makes about `--feature` and
513    /// `--release-version`, with Kafka 4.3.1's text (`Formatter`,
514    /// `Feature.fromFeatureLevel`, `MetadataVersion.fromFeatureLevel`), and
515    /// the order it makes them in: the release first, then every name, then
516    /// every level.
517    #[yare::parameterized(
518        unknown_feature = {
519            None,
520            vec![spec("bogus.version", 1)],
521            STRICT,
522            format!("Unsupported feature: bogus.version. Supported features are: {SUPPORTED_FEATURES}"),
523        },
524        unknown_feature_unstable = {
525            None,
526            vec![spec("bogus.version", 1)],
527            UnstableFeatureVersions::Enabled,
528            format!("Unsupported feature: bogus.version. Supported features are: {SUPPORTED_FEATURES}"),
529        },
530        group_level = {
531            None,
532            vec![spec("group.version", 5)],
533            STRICT,
534            "No feature:group.version with feature level 5".to_owned(),
535        },
536        transaction_level = {
537            None,
538            vec![spec("transaction.version", 3)],
539            STRICT,
540            "No feature:transaction.version with feature level 3".to_owned(),
541        },
542        negative_level = {
543            None,
544            vec![spec("group.version", -1)],
545            STRICT,
546            "No feature:group.version with feature level -1".to_owned(),
547        },
548        metadata_below_table = {
549            None,
550            vec![spec("metadata.version", 6)],
551            STRICT,
552            "No MetadataVersion with feature level 6. Valid feature levels are from 7 to 31.".to_owned(),
553        },
554        metadata_one_below_table = {
555            None,
556            vec![spec("metadata.version", 1)],
557            STRICT,
558            "No MetadataVersion with feature level 1. Valid feature levels are from 7 to 31.".to_owned(),
559        },
560        unstable_level_not_stable = {
561            None,
562            vec![spec("metadata.version", 31)],
563            STRICT,
564            "metadata.version 4.4-IV0 is not yet stable.".to_owned(),
565        },
566        metadata_32_above_table = {
567            None,
568            vec![spec("metadata.version", 32)],
569            STRICT,
570            "No MetadataVersion with feature level 32. Valid feature levels are from 7 to 31.".to_owned(),
571        },
572        metadata_33_above_table = {
573            None,
574            vec![spec("metadata.version", 33)],
575            STRICT,
576            "No MetadataVersion with feature level 33. Valid feature levels are from 7 to 31.".to_owned(),
577        },
578        metadata_34_above_table = {
579            None,
580            vec![spec("metadata.version", 34)],
581            STRICT,
582            "No MetadataVersion with feature level 34. Valid feature levels are from 7 to 31.".to_owned(),
583        },
584        metadata_99_above_table = {
585            None,
586            vec![spec("metadata.version", 99)],
587            STRICT,
588            "No MetadataVersion with feature level 99. Valid feature levels are from 7 to 31.".to_owned(),
589        },
590        unstable_below_table = {
591            None,
592            vec![spec("metadata.version", 6)],
593            UnstableFeatureVersions::Enabled,
594            "No MetadataVersion with feature level 6. Valid feature levels are from 7 to 34.".to_owned(),
595        },
596        unstable_above_table = {
597            None,
598            vec![spec("metadata.version", 35)],
599            UnstableFeatureVersions::Enabled,
600            "No MetadataVersion with feature level 35. Valid feature levels are from 7 to 34.".to_owned(),
601        },
602        unknown_release_precedes_feature = {
603            Some("9.9"),
604            vec![spec("bogus.version", 1)],
605            STRICT,
606            unknown_release("9.9"),
607        },
608        release_and_metadata_are_ambiguous = {
609            Some("4.0-IV0"),
610            vec![spec("metadata.version", 24)],
611            STRICT,
612            "Use --release-version instead of --feature metadata.version=X to avoid ambiguity.".to_owned(),
613        },
614        ambiguity_precedes_bad_level = {
615            Some("4.0-IV0"),
616            vec![spec("metadata.version", 99)],
617            STRICT,
618            "Use --release-version instead of --feature metadata.version=X to avoid ambiguity.".to_owned(),
619        },
620        metadata_error_precedes_unknown_name = {
621            None,
622            vec![spec("bogus.version", 1), spec("metadata.version", 99)],
623            STRICT,
624            "No MetadataVersion with feature level 99. Valid feature levels are from 7 to 31.".to_owned(),
625        },
626        unknown_name_precedes_bad_level = {
627            None,
628            vec![spec("group.version", 5), spec("bogus.version", 1)],
629            STRICT,
630            format!("Unsupported feature: bogus.version. Supported features are: {SUPPORTED_FEATURES}"),
631        },
632    )]
633    fn feature_errors_are_kafkas_formatter_errors(
634        release: Option<&str>,
635        features: Vec<FeatureLevelRecord>,
636        unstable: UnstableFeatureVersions,
637        want: String,
638    ) {
639        assert2::check!(
640            resolve_format_features(release, &features, unstable)
641                .map(|resolved| resolved.metadata_version)
642                == Err(want),
643            "{release:?} {features:?} {unstable:?}"
644        );
645    }
646
647    /// #784: a Kafka trunk `metadata.version` (4.4-IV0 to 4.5-IV0) is
648    /// refused with Kafka 4.3.1's `kafka-storage format` messages unless
649    /// unstable feature versions are enabled, and the default release is
650    /// `latestTesting` when they are.
651    #[yare::parameterized(
652        default_stable = { None, None, STRICT, Ok(30) },
653        default_trunk = { None, None, UnstableFeatureVersions::Enabled, Ok(34) },
654        unknown_iv_stable = { Some("4.4"), None, STRICT, Err(unknown_release("4.4")) },
655        first_trunk_iv = { Some("4.4-IV0"), None, UnstableFeatureVersions::Enabled, Ok(31) },
656        trunk_release = { Some("4.4"), None, UnstableFeatureVersions::Enabled, Ok(33) },
657        next_trunk_iv = { Some("4.5-IV0"), None, UnstableFeatureVersions::Enabled, Ok(34) },
658        next_trunk_release = { Some("4.5"), None, UnstableFeatureVersions::Enabled, Ok(34) },
659        trunk_iv_stable_refused = {
660            Some("4.5-IV0"),
661            None,
662            STRICT,
663            Err(unknown_release("4.5-IV0")),
664        },
665        explicit_unstable_level_refused = {
666            None,
667            Some(31),
668            STRICT,
669            Err("metadata.version 4.4-IV0 is not yet stable.".to_owned()),
670        },
671        unknown_level_33_stable = {
672            None,
673            Some(33),
674            STRICT,
675            Err("No MetadataVersion with feature level 33. Valid feature levels are from 7 to 31.".to_owned()),
676        },
677        unknown_level_34_stable = {
678            None,
679            Some(34),
680            STRICT,
681            Err("No MetadataVersion with feature level 34. Valid feature levels are from 7 to 31.".to_owned()),
682        },
683        explicit_stable_level = { None, Some(30), STRICT, Ok(30) },
684        explicit_trunk_level_31 = { None, Some(31), UnstableFeatureVersions::Enabled, Ok(31) },
685        explicit_trunk_level_32 = { None, Some(32), UnstableFeatureVersions::Enabled, Ok(32) },
686        explicit_trunk_level_33 = { None, Some(33), UnstableFeatureVersions::Enabled, Ok(33) },
687        explicit_trunk_level_34 = { None, Some(34), UnstableFeatureVersions::Enabled, Ok(34) },
688    )]
689    fn unstable_metadata_versions_follow_unstable_feature_versions_enable(
690        release: Option<&str>,
691        level: Option<i16>,
692        unstable: UnstableFeatureVersions,
693        want: Result<i16, String>,
694    ) {
695        let features = level.map_or_else(Vec::new, |level| vec![spec("metadata.version", level)]);
696        assert2::check!(
697            resolve_format_features(release, &features, unstable)
698                .map(|resolved| resolved.metadata_version)
699                == want,
700            "{release:?} {features:?} {unstable:?}"
701        );
702    }
703
704    /// KIP-1191: `share.version` 2 is Kafka trunk's `SV_2`. Kafka 4.3.1 stops
705    /// at 1, so a default format refuses 2 and seeds 1; with unstable feature
706    /// versions the level is accepted, and a format at `4.4-IV0` or later
707    /// seeds it as trunk's default, from the registry's `default_level`.
708    #[yare::parameterized(
709        default_stable = { None, None, STRICT, Ok(Some(1)) },
710        explicit_one_stable = { None, Some(1), STRICT, Ok(Some(1)) },
711        explicit_two_stable_refused = {
712            None,
713            Some(2),
714            STRICT,
715            Err("No feature:share.version with feature level 2".to_owned()),
716        },
717        explicit_two_trunk = { None, Some(2), UnstableFeatureVersions::Enabled, Ok(Some(2)) },
718        explicit_one_trunk = { None, Some(1), UnstableFeatureVersions::Enabled, Ok(Some(1)) },
719        default_trunk = { None, None, UnstableFeatureVersions::Enabled, Ok(Some(2)) },
720        release_4_4_iv0_trunk = {
721            Some("4.4-IV0"),
722            None,
723            UnstableFeatureVersions::Enabled,
724            Ok(Some(2)),
725        },
726        release_4_3_trunk = { Some("4.3"), None, UnstableFeatureVersions::Enabled, Ok(Some(1)) },
727        release_4_1_trunk = { Some("4.1"), None, UnstableFeatureVersions::Enabled, Ok(None) },
728    )]
729    fn share_version_two_is_trunks_level(
730        release: Option<&str>,
731        level: Option<i16>,
732        unstable: UnstableFeatureVersions,
733        want: Result<Option<i16>, String>,
734    ) {
735        let features = level.map_or_else(Vec::new, |level| vec![spec("share.version", level)]);
736        let seeded = resolve_format_features(release, &features, unstable).map(
737            |ResolvedFeatures {
738                 metadata_version: bootstrap_mv,
739                 overrides,
740             }| {
741                krabka_metadata::bootstrap_feature_records_with_overrides(bootstrap_mv, &overrides)
742                    .into_iter()
743                    .find_map(|record| match record {
744                        MetadataRecord::V1FeatureLevel(f) if f.name == "share.version" => {
745                            Some(f.level)
746                        }
747                        _ => None,
748                    })
749            },
750        );
751        assert2::check!(seeded == want, "{release:?} {features:?} {unstable:?}");
752    }
753
754    /// `krabka.version` bootstraps at level 0, which seeds no record, as Kafka
755    /// omits every level-0 feature from `bootstrap.checkpoint`. A JVM node in
756    /// a mixed cluster supports only level 0, so the level is left for an
757    /// operator to finalize once every node is krabka. `--feature
758    /// krabka.version=N` overrides it, as `kafka-storage format --feature`
759    /// overrides a Kafka feature, and a level past the supported range is
760    /// `Feature.fromFeatureLevel`'s refusal.
761    #[yare::parameterized(
762        no_override = { "no override" },
763        override_one = { "override to 1" },
764        override_zero = { "override to 0" },
765        override_above_range = { "override past the range" },
766    )]
767    fn krabka_version_bootstraps_at_level_zero_unless_overridden(scenario: &str) {
768        /// What it is, the `--feature` flags, and the seeded records or the error.
769        struct Case<'a> {
770            what: &'a str,
771            features: Vec<FeatureLevelRecord>,
772            want: Result<Vec<MetadataRecord>, String>,
773        }
774        let feature = |name: &str, level| {
775            MetadataRecord::V1FeatureLevel(krabka_metadata::FeatureLevelRecord {
776                name: name.into(),
777                level,
778            })
779        };
780        // The latest production release's defaults, Kafka 4.3's.
781        let release_defaults = vec![
782            feature("metadata.version", LATEST_PRODUCTION_METADATA_VERSION),
783            feature("group.version", 1),
784            feature("transaction.version", 2),
785            feature("share.version", 1),
786            feature("streams.version", 1),
787            feature("eligible.leader.replicas.version", 1),
788        ];
789        let with_krabka = |level| {
790            let mut records = release_defaults.clone();
791            records.push(feature("krabka.version", level));
792            records
793        };
794        let krabka = |level| vec![spec("krabka.version", level)];
795        let cases = [
796            Case {
797                what: "no override",
798                features: vec![],
799                want: Ok(release_defaults.clone()),
800            },
801            Case {
802                what: "override to 1",
803                features: krabka(1),
804                want: Ok(with_krabka(1)),
805            },
806            Case {
807                what: "override to 0",
808                features: krabka(0),
809                want: Ok(release_defaults.clone()),
810            },
811            Case {
812                what: "override past the range",
813                features: krabka(3),
814                want: Err("No feature:krabka.version with feature level 3".to_owned()),
815            },
816        ];
817        let Case {
818            what,
819            features,
820            want,
821        } = cases
822            .into_iter()
823            .find(|case| case.what == scenario)
824            .expect("named yare case exists");
825        let seeded = resolve_format_features(None, &features, STRICT).map(
826            |ResolvedFeatures {
827                 metadata_version: bootstrap_mv,
828                 overrides,
829             }| {
830                krabka_metadata::bootstrap_feature_records_with_overrides(bootstrap_mv, &overrides)
831            },
832        );
833        assert2::check!(seeded == want, "{what}");
834        assert2::check!(parse_feature_spec("krabka.version=0") == Ok(spec("krabka.version", 0)));
835    }
836
837    #[test]
838    fn resolve_features_rejects_bad_release_string() {
839        assert2::assert!(resolve_format_features(Some("2.8"), &[], STRICT).is_err());
840    }
841}
krabka_format/format/features.rs

krabka_format/format/
features.rs
