use std::collections::BTreeMap;
use crate::MetadataImage;
pub trait Feature: Sync {
fn name(&self) -> &'static str;
fn supported_range(&self) -> (i16, i16);
fn default_level(&self, bootstrap_mv: i16) -> i16;
fn min_required_floor(&self, _image: &MetadataImage) -> i16 {
self.supported_range().0
}
fn dependencies(&self, _level: i16) -> &'static [(&'static str, i16)] {
&[]
}
fn level_name(&self, _level: i16) -> Option<&'static str> {
None
}
}
pub struct MetadataVersionFeature;
impl Feature for MetadataVersionFeature {
fn name(&self) -> &'static str {
crate::metadata_version::METADATA_VERSION_FEATURE
}
fn supported_range(&self) -> (i16, i16) {
(
crate::metadata_version::METADATA_VERSION_MIN,
crate::metadata_version::METADATA_VERSION_MAX,
)
}
fn default_level(&self, bootstrap_mv: i16) -> i16 {
bootstrap_mv.clamp(
crate::metadata_version::METADATA_VERSION_MIN,
crate::metadata_version::METADATA_VERSION_MAX,
)
}
fn min_required_floor(&self, image: &MetadataImage) -> i16 {
image.min_required_metadata_version()
}
fn level_name(&self, level: i16) -> Option<&'static str> {
crate::metadata_version::from_feature_level(level)
.map(crate::metadata_version::MetadataVersion::ivn)
}
}
pub struct GroupVersionFeature;
impl Feature for GroupVersionFeature {
fn name(&self) -> &'static str {
crate::group_version::GROUP_VERSION_FEATURE
}
fn supported_range(&self) -> (i16, i16) {
(
crate::group_version::GROUP_VERSION_MIN,
crate::group_version::GROUP_VERSION_MAX,
)
}
fn default_level(&self, bootstrap_mv: i16) -> i16 {
if bootstrap_mv >= crate::group_version::GROUP_VERSION_GA_METADATA_LEVEL {
crate::group_version::GROUP_VERSION_MAX
} else {
crate::group_version::GROUP_VERSION_MIN
}
}
}
pub struct TransactionVersionFeature;
impl Feature for TransactionVersionFeature {
fn name(&self) -> &'static str {
crate::transaction_version::TRANSACTION_VERSION_FEATURE
}
fn supported_range(&self) -> (i16, i16) {
(
crate::transaction_version::TRANSACTION_VERSION_MIN,
crate::transaction_version::TRANSACTION_VERSION_MAX,
)
}
fn default_level(&self, bootstrap_mv: i16) -> i16 {
use crate::transaction_version::{TV1_METADATA_LEVEL, TV2_METADATA_LEVEL};
match bootstrap_mv {
level if level >= TV2_METADATA_LEVEL => 2,
level if level >= TV1_METADATA_LEVEL => 1,
_ => 0,
}
}
}
pub struct ShareVersionFeature;
impl Feature for ShareVersionFeature {
fn name(&self) -> &'static str {
crate::metadata_version::SHARE_VERSION_FEATURE
}
fn supported_range(&self) -> (i16, i16) {
(
crate::metadata_version::SHARE_VERSION_MIN,
crate::metadata_version::SHARE_VERSION_MAX,
)
}
fn default_level(&self, bootstrap_mv: i16) -> i16 {
use crate::metadata_version::{
SHARE_VERSION_2_DEFAULT_METADATA_LEVEL, SHARE_VERSION_DEFAULT_METADATA_LEVEL,
};
match bootstrap_mv {
level if level >= SHARE_VERSION_2_DEFAULT_METADATA_LEVEL => 2,
level if level >= SHARE_VERSION_DEFAULT_METADATA_LEVEL => 1,
_ => crate::metadata_version::SHARE_VERSION_MIN,
}
}
}
pub struct StreamsVersionFeature;
impl Feature for StreamsVersionFeature {
fn name(&self) -> &'static str {
crate::metadata_version::STREAMS_VERSION_FEATURE
}
fn supported_range(&self) -> (i16, i16) {
(
crate::metadata_version::STREAMS_VERSION_MIN,
crate::metadata_version::STREAMS_VERSION_MAX,
)
}
fn default_level(&self, bootstrap_mv: i16) -> i16 {
if bootstrap_mv >= crate::metadata_version::STREAMS_VERSION_DEFAULT_METADATA_LEVEL {
crate::metadata_version::STREAMS_VERSION_MAX
} else {
crate::metadata_version::STREAMS_VERSION_MIN
}
}
}
pub struct ElrVersionFeature;
impl Feature for ElrVersionFeature {
fn name(&self) -> &'static str {
crate::metadata_version::ELR_VERSION_FEATURE
}
fn supported_range(&self) -> (i16, i16) {
(
crate::metadata_version::ELR_VERSION_MIN,
crate::metadata_version::ELR_VERSION_MAX,
)
}
fn default_level(&self, bootstrap_mv: i16) -> i16 {
if bootstrap_mv >= crate::metadata_version::ELR_DEFAULT_METADATA_LEVEL {
crate::metadata_version::ELR_VERSION_MAX
} else {
crate::metadata_version::ELR_VERSION_MIN
}
}
fn dependencies(&self, level: i16) -> &'static [(&'static str, i16)] {
if level >= 1 {
const ELRV_1_DEPENDENCIES: &[(&str, i16)] = &[(
crate::metadata_version::METADATA_VERSION_FEATURE,
crate::metadata_version::ELR_MIN_LEVEL,
)];
ELRV_1_DEPENDENCIES
} else {
&[]
}
}
}
pub struct KRaftVersionFeature;
impl Feature for KRaftVersionFeature {
fn name(&self) -> &'static str {
crate::metadata_version::KRAFT_VERSION_FEATURE
}
fn supported_range(&self) -> (i16, i16) {
(0, 1)
}
fn default_level(&self, _bootstrap_mv: i16) -> i16 {
0
}
}
pub struct KrabkaVersionFeature;
impl Feature for KrabkaVersionFeature {
fn name(&self) -> &'static str {
crate::krabka_version::KRABKA_VERSION_FEATURE
}
fn supported_range(&self) -> (i16, i16) {
(
crate::krabka_version::KRABKA_VERSION_MIN,
crate::krabka_version::KRABKA_VERSION_MAX,
)
}
fn default_level(&self, _bootstrap_mv: i16) -> i16 {
crate::krabka_version::KRABKA_VERSION_MIN
}
}
#[must_use]
pub fn feature_registry() -> &'static [&'static dyn Feature] {
const REGISTRY: &[&dyn Feature] = &[
&MetadataVersionFeature,
&GroupVersionFeature,
&TransactionVersionFeature,
&ShareVersionFeature,
&StreamsVersionFeature,
&ElrVersionFeature,
&KRaftVersionFeature,
&KrabkaVersionFeature,
];
REGISTRY
}
#[must_use]
pub fn supported_feature_ranges() -> BTreeMap<String, (i16, i16)> {
let mut ranges: BTreeMap<_, _> = feature_registry()
.iter()
.map(|feature| (feature.name().to_string(), feature.supported_range()))
.collect();
ranges.insert(
crate::metadata_version::METADATA_DOWNGRADE_CAPABILITY_FEATURE.into(),
(
crate::metadata_version::METADATA_DOWNGRADE_CAPABILITY_LEVEL,
crate::metadata_version::METADATA_DOWNGRADE_CAPABILITY_LEVEL,
),
);
ranges
}
#[must_use]
pub fn feature(name: &str) -> Option<&'static dyn Feature> {
feature_registry()
.iter()
.copied()
.find(|f| f.name() == name)
}
#[must_use]
pub fn bootstrap_feature_records(bootstrap_mv: i16) -> Vec<crate::MetadataRecord> {
bootstrap_feature_records_with_overrides(bootstrap_mv, &BTreeMap::new())
}
#[must_use]
pub fn bootstrap_feature_records_with_overrides(
bootstrap_mv: i16,
overrides: &BTreeMap<String, i16>,
) -> Vec<crate::MetadataRecord> {
feature_registry()
.iter()
.filter(|f| f.name() != crate::metadata_version::KRAFT_VERSION_FEATURE)
.filter_map(|f| {
let level = overrides
.get(f.name())
.copied()
.unwrap_or_else(|| f.default_level(bootstrap_mv));
(level > 0).then(|| {
crate::MetadataRecord::V1FeatureLevel(crate::FeatureLevelRecord {
name: f.name().to_string(),
level,
})
})
})
.collect()
}
pub fn validate_feature_dependencies(resolved: &BTreeMap<String, i16>) -> Result<(), String> {
check_deps(resolved, |name, level| {
feature(name).map_or(&[][..], |f| f.dependencies(level))
})
}
fn check_deps(
resolved: &BTreeMap<String, i16>,
deps_of: impl Fn(&str, i16) -> &'static [(&'static str, i16)],
) -> Result<(), String> {
for (name, &level) in resolved {
for &(dep, min) in deps_of(name, level) {
let have = resolved.get(dep).copied().unwrap_or(0);
if have < min {
return Err(format!(
"feature {name}={level} requires {dep}>={min}, but {dep} is finalized at {have}"
));
}
}
}
Ok(())
}
#[must_use]
pub fn is_supported_level(name: &str, level: i16) -> bool {
feature(name).is_some_and(|f| {
let (min, max) = f.supported_range();
(min..=max).contains(&level)
})
}
#[cfg(test)]
mod tests {
use std::collections::BTreeMap;
use assert2::check;
use super::*;
#[test]
fn registry_contains_metadata_version() {
let f = feature("metadata.version").expect("registered");
assert2::assert!(f.supported_range() == (7, 34));
assert2::assert!(feature("not.a.feature").is_none());
}
#[test]
fn metadata_version_default_is_the_bootstrap_level_clamped() {
let f = feature("metadata.version").unwrap();
for (_case, bootstrap, want) in [
("maximum bootstrap", 34, 34),
("minimum bootstrap", 7, 7),
("above maximum", 99, 34),
("below minimum", 1, 7),
] {
assert2::assert!(f.default_level(bootstrap) == want);
}
}
#[test]
fn is_supported_level_checks_range() {
for (name, level, want) in [
("metadata.version", 7, true),
("metadata.version", 34, true),
("metadata.version", 6, false),
("metadata.version", 35, false),
("transaction.version", 2, true),
("transaction.version", 3, false),
("krabka.version", 0, true),
("krabka.version", 1, true),
("krabka.version", 2, false),
("not.a.feature", 1, false),
] {
assert2::assert!(is_supported_level(name, level) == want);
}
}
#[test]
fn supported_feature_ranges_are_pinned() {
let want: BTreeMap<String, (i16, i16)> = [
("eligible.leader.replicas.version", (0, 1)),
("group.version", (0, 1)),
("krabka.metadata.downgrade", (1, 1)),
("krabka.version", (0, 1)),
("kraft.version", (0, 1)),
("metadata.version", (7, 34)),
("share.version", (0, 2)),
("streams.version", (0, 1)),
("transaction.version", (0, 2)),
]
.into_iter()
.map(|(name, range)| (name.to_string(), range))
.collect();
check!(supported_feature_ranges() == want);
}
#[test]
fn krabka_version_bootstraps_at_zero() {
use crate::metadata_version::{METADATA_VERSION_MAX, METADATA_VERSION_MIN};
for (case, bootstrap, overrides, want) in [
("minimum release", METADATA_VERSION_MIN, vec![], None),
("latest release", METADATA_VERSION_MAX, vec![], None),
(
"override to 1 earns a record",
METADATA_VERSION_MAX,
vec![("krabka.version".to_string(), 1)],
Some(1),
),
(
"override to 0 omits the record",
METADATA_VERSION_MAX,
vec![("krabka.version".to_string(), 0)],
None,
),
] {
let overrides: BTreeMap<String, i16> = overrides.into_iter().collect();
let levels = levels_of(&bootstrap_feature_records_with_overrides(
bootstrap, &overrides,
));
check!(levels.get("krabka.version").copied() == want, "{case}");
}
}
#[test]
fn krabka_version_declares_no_dependencies_and_no_level_name() {
let f = feature("krabka.version").expect("registered");
check!((f.dependencies(0), f.dependencies(1), f.level_name(1)) == (&[][..], &[][..], None));
let resolved = BTreeMap::from([("krabka.version".to_string(), 1)]);
check!(validate_feature_dependencies(&resolved) == Ok(()));
}
#[test]
fn downgrade_capability_is_registration_only() {
let ranges = supported_feature_ranges();
check!(
ranges.get(crate::metadata_version::METADATA_DOWNGRADE_CAPABILITY_FEATURE)
== Some(&(
crate::metadata_version::METADATA_DOWNGRADE_CAPABILITY_LEVEL,
crate::metadata_version::METADATA_DOWNGRADE_CAPABILITY_LEVEL,
))
);
check!(feature(crate::metadata_version::METADATA_DOWNGRADE_CAPABILITY_FEATURE).is_none());
}
#[test]
fn metadata_version_level_name() {
let f = feature("metadata.version").unwrap();
for (_case, level, want) in [
("latest level", 34, Some("4.5-IV0")),
("controller unregistration level", 33, Some("4.4-IV2")),
("CIDR ACL level", 32, Some("4.4-IV1")),
("cordoned log dirs level", 30, Some("4.3-IV0")),
("earliest level", 7, Some("3.3-IV3")),
("unknown level", 99, None),
] {
assert2::assert!(f.level_name(level) == want);
}
}
#[test]
fn group_version_registered_with_range() {
let f = feature("group.version").expect("registered");
assert2::assert!(f.supported_range() == (0, 1));
}
#[test]
fn registry_feature_contracts_are_pinned() {
let image = MetadataImage::new(uuid::Uuid::nil());
let expected = [
("metadata.version", (7, 34), 25, 7),
("group.version", (0, 1), 1, 0),
("transaction.version", (0, 2), 2, 0),
("share.version", (0, 2), 0, 0),
("streams.version", (0, 1), 0, 0),
("eligible.leader.replicas.version", (0, 1), 0, 0),
("kraft.version", (0, 1), 0, 0),
("krabka.version", (0, 1), 0, 0),
];
for (name, range, default_at_25, floor) in expected {
let f = feature(name).expect("registered");
check!(
(
f.supported_range(),
f.default_level(25),
f.min_required_floor(&image),
) == (range, default_at_25, floor),
"feature {name}"
);
}
}
#[test]
fn group_version_default_follows_release() {
let f = feature("group.version").unwrap();
for (_case, bootstrap, want) in [
(
"before GA threshold",
crate::group_version::GROUP_VERSION_GA_METADATA_LEVEL - 1,
0,
),
(
"at GA threshold",
crate::group_version::GROUP_VERSION_GA_METADATA_LEVEL,
1,
),
("after GA threshold", 25, 1),
] {
assert2::assert!(f.default_level(bootstrap) == want);
}
}
#[test]
fn group_version_declares_no_hard_dependencies() {
let f = feature("group.version").unwrap();
assert2::assert!(
(f.dependencies(0).is_empty(), f.dependencies(1).is_empty()) == (true, true)
);
}
#[test]
fn transaction_version_registered() {
let f = feature("transaction.version").expect("registered");
assert2::assert!(f.supported_range() == (0, 2));
}
#[test]
fn streams_version_registered_with_no_dependencies() {
let f = feature("streams.version").expect("registered");
check!((f.supported_range(), f.dependencies(1).is_empty()) == ((0, 1), true));
}
#[test]
fn opt_in_feature_defaults_follow_kafka_bootstrap_levels() {
use crate::metadata_version::{
ELR_DEFAULT_METADATA_LEVEL, METADATA_VERSION_MAX,
SHARE_VERSION_2_DEFAULT_METADATA_LEVEL, SHARE_VERSION_DEFAULT_METADATA_LEVEL,
STREAMS_VERSION_DEFAULT_METADATA_LEVEL,
};
for (name, bootstrap, want) in [
("eligible.leader.replicas.version", 25, 0),
(
"eligible.leader.replicas.version",
ELR_DEFAULT_METADATA_LEVEL - 1,
0,
),
(
"eligible.leader.replicas.version",
ELR_DEFAULT_METADATA_LEVEL,
1,
),
("eligible.leader.replicas.version", METADATA_VERSION_MAX, 1),
("share.version", SHARE_VERSION_DEFAULT_METADATA_LEVEL - 1, 0),
("share.version", SHARE_VERSION_DEFAULT_METADATA_LEVEL, 1),
(
"share.version",
SHARE_VERSION_2_DEFAULT_METADATA_LEVEL - 1,
1,
),
("share.version", SHARE_VERSION_2_DEFAULT_METADATA_LEVEL, 2),
("share.version", METADATA_VERSION_MAX, 2),
(
"streams.version",
STREAMS_VERSION_DEFAULT_METADATA_LEVEL - 1,
0,
),
("streams.version", STREAMS_VERSION_DEFAULT_METADATA_LEVEL, 1),
("streams.version", METADATA_VERSION_MAX, 1),
] {
check!(
feature(name).unwrap().default_level(bootstrap) == want,
"feature {name} at bootstrap level {bootstrap}"
);
}
}
#[test]
fn elr_version_is_opt_in_and_depends_on_the_elr_metadata_level() {
let f = feature("eligible.leader.replicas.version").expect("registered");
check!(
(
f.supported_range(),
f.default_level(crate::metadata_version::METADATA_VERSION_MAX),
f.dependencies(0),
f.dependencies(1),
) == (
(0, 1),
1,
&[][..],
&[(
crate::metadata_version::METADATA_VERSION_FEATURE,
crate::metadata_version::ELR_MIN_LEVEL,
)][..],
)
);
}
#[test]
fn elr_version_level_one_requires_the_elr_metadata_version() {
for (_case, metadata_level, want_err) in [
(
"below 4.0-IV1",
crate::metadata_version::ELR_MIN_LEVEL - 1,
true,
),
("at 4.0-IV1", crate::metadata_version::ELR_MIN_LEVEL, false),
(
"above 4.0-IV1",
crate::metadata_version::METADATA_VERSION_MAX,
false,
),
] {
let resolved = BTreeMap::from([
(
crate::metadata_version::METADATA_VERSION_FEATURE.to_string(),
metadata_level,
),
(crate::metadata_version::ELR_VERSION_FEATURE.to_string(), 1),
]);
check!(validate_feature_dependencies(&resolved).is_err() == want_err);
}
}
#[test]
fn transaction_version_default_jumps_to_two_at_4_0_iv2() {
let f = feature("transaction.version").unwrap();
for (_case, bootstrap, want) in [
("below activation", 23, 0),
("at activation", 24, 2),
("after activation", 25, 2),
] {
assert2::assert!(f.default_level(bootstrap) == want);
}
}
#[test]
fn transaction_version_declares_no_hard_dependencies() {
let f = feature("transaction.version").unwrap();
for level in [0, 1, 2] {
assert2::assert!(f.dependencies(level).is_empty());
}
}
#[test]
fn transaction_version_ga_threshold_is_4_0_iv2() {
assert2::assert!(
crate::metadata_version::from_feature_level(
crate::transaction_version::TV2_METADATA_LEVEL
)
.unwrap()
.ivn()
== "4.0-IV2"
);
}
fn levels_of(recs: &[crate::MetadataRecord]) -> BTreeMap<String, i16> {
recs.iter()
.filter_map(|r| match r {
crate::MetadataRecord::V1FeatureLevel(f) => Some((f.name.clone(), f.level)),
_ => None,
})
.collect()
}
#[test]
fn bootstrap_omits_level_zero_features() {
let levels = levels_of(&bootstrap_feature_records(25));
assert2::assert!(
levels
== BTreeMap::from([
("metadata.version".to_string(), 25),
("group.version".to_string(), 1),
("transaction.version".to_string(), 2),
])
);
}
#[test]
fn bootstrap_at_the_latest_level_enables_every_released_feature() {
let want: Vec<crate::MetadataRecord> = [
("metadata.version", 34),
("group.version", 1),
("transaction.version", 2),
("share.version", 2),
("streams.version", 1),
("eligible.leader.replicas.version", 1),
]
.into_iter()
.map(|(name, level)| {
crate::MetadataRecord::V1FeatureLevel(crate::FeatureLevelRecord {
name: name.to_string(),
level,
})
})
.collect();
check!(bootstrap_feature_records(crate::metadata_version::METADATA_VERSION_MAX) == want);
}
#[test]
fn bootstrap_with_overrides_applies_explicit_levels() {
let mut ov = BTreeMap::new();
ov.insert("group.version".to_string(), 0i16);
let levels = levels_of(&bootstrap_feature_records_with_overrides(25, &ov));
assert2::assert!(
levels
== BTreeMap::from([
("metadata.version".to_string(), 25),
("transaction.version".to_string(), 2),
])
);
}
#[test]
fn bootstrap_override_can_enable_an_opt_in_feature() {
let mut ov = BTreeMap::new();
ov.insert("streams.version".to_string(), 1i16);
let levels = levels_of(&bootstrap_feature_records_with_overrides(25, &ov));
assert2::assert!(levels.get("streams.version") == Some(&1));
}
#[test]
fn bootstrap_unlisted_feature_follows_bootstrap_mv() {
let levels = levels_of(&bootstrap_feature_records_with_overrides(
23,
&BTreeMap::new(),
));
assert2::assert!(
levels
== BTreeMap::from([
("metadata.version".to_string(), 23),
("group.version".to_string(), 1),
])
);
}
#[test]
fn validate_dependencies_ok_for_real_registry() {
let mut resolved = BTreeMap::new();
resolved.insert("metadata.version".to_string(), 25i16);
resolved.insert("group.version".to_string(), 1i16);
resolved.insert("transaction.version".to_string(), 2i16);
assert2::assert!(validate_feature_dependencies(&resolved).is_ok());
}
#[test]
fn check_deps_enforces_minimum_dependency_levels() {
fn deps_of(name: &str, level: i16) -> &'static [(&'static str, i16)] {
match (name, level) {
("b", 1) => &[("a", 2)],
_ => &[],
}
}
let mut ok = BTreeMap::new();
ok.insert("a".to_string(), 2i16);
ok.insert("b".to_string(), 1i16);
assert2::assert!(check_deps(&ok, deps_of).is_ok());
let mut too_low = BTreeMap::new();
too_low.insert("a".to_string(), 1i16);
too_low.insert("b".to_string(), 1i16);
assert2::assert!(check_deps(&too_low, deps_of).is_err());
let mut missing = BTreeMap::new();
missing.insert("b".to_string(), 1i16);
assert2::assert!(check_deps(&missing, deps_of).is_err());
}
#[test]
fn group_version_ga_threshold_is_4_0_iv0() {
assert2::assert!(
crate::metadata_version::from_feature_level(
crate::group_version::GROUP_VERSION_GA_METADATA_LEVEL
)
.unwrap()
.ivn()
== "4.0-IV0"
);
}
#[test]
fn default_min_required_floor_is_supported_min() {
let img = MetadataImage::new(uuid::Uuid::nil());
assert2::assert!(feature("group.version").unwrap().min_required_floor(&img) == 0);
}
#[test]
fn metadata_version_min_required_floor_tracks_image() {
let img = MetadataImage::new(uuid::Uuid::nil());
let mv = feature("metadata.version").unwrap();
assert2::assert!(mv.min_required_floor(&img) == 7);
assert2::assert!(img.min_required_metadata_version() == 7);
}
#[test]
fn plain_features_have_no_level_name() {
assert2::assert!(feature("group.version").unwrap().level_name(1).is_none());
assert2::assert!(
feature("transaction.version")
.unwrap()
.level_name(2)
.is_none()
);
}
#[test]
fn share_version_accessors() {
let f = feature("share.version").expect("registered");
check!(
(f.name(), f.supported_range(), f.default_level(25)) == ("share.version", (0, 2), 0)
);
}
}