//! Translate Krabka's internal [`MetadataRecord`] currency to and from the
//! KIP-631 [`KraftMetadataRecord`] wire shape, so the controller log and
//! snapshots are genuinely KIP-631-framed.
//!
//! The conversion is *not* a pure field rename. Several Krabka records
//! carry state that the KIP-631 records do not, and the reverse also holds.
//! This module bridges the gap and defaults the KIP-631 extras that Krabka
//! does not model. The round-trip
//! `rec -> to_kraft(&image) -> from_kraft(&image) == rec` is the
//! correctness bar for every modeled variant.
//!
//! ## Variants with no faithful KIP-631 counterpart
//!
//! - `V1Voters` and `V1KRaftVersion` are KIP-853 raft *control* records and
//! not metadata records. They have no entry in the
//! [`KraftMetadataRecord`] dispatch, and nothing submits them through the
//! translation boundary, so
//! [`to_kraft`] returns [`TranslateError::NoCounterpart`] for them.
//!
//! ## Fan-out / image-keyed mappings
//!
//! - `V1TopicConfig` is whole-map authoritative, and KIP-631 has only per-key
//! `ConfigRecord`s. [`to_kraft_records`] diffs the new map against the
//! current overrides of the image. It emits a *set* per present key plus a
//! *tombstone* (`value: None`) per dropped key. [`from_kraft`] merges the
//! delta of one key back onto the image map and re-emits the whole map, so
//! the replace semantics of `MetadataImage::apply` stay unchanged. Records
//! must apply in log order against a live image.
//! - `V1DeleteAccessControlEntry(AclEntryFilter)`: KIP-631's
//! `RemoveAccessControlEntryRecord` deletes by `id: Uuid` only. Krabka does
//! not model ACL ids, so [`to_kraft_records`] resolves the filter against
//! the image to the matching entries and emits one `RemoveAccessControlEntry`
//! per match, keyed by a deterministic content-hash id. That is the same id
//! that the add path stamps. [`from_kraft`] rescans the image by that hash to
//! recover the entry and emit a pinned filter. This is self-consistent within
//! Krabka.
//!
//! ## Lossy / awkward mappings (documented)
//!
//! - `RegisterBroker`: Krabka stores a top-level `(host, port)` plus a
//! per-listener `endpoints` list; KIP-631 has only `end_points` (no
//! top-level host/port). When listeners exist, their first entry supplies
//! the legacy top-level pair on decode. A legacy Krabka record with no
//! listeners uses one empty-name compatibility entry. `broker_epoch`,
//! `incarnation_id`, `features`, `log_dirs`, `fenced`,
//! `in_controlled_shutdown` and `cordoned_log_dirs` are carried; the ZK
//! migration flag is dropped.
//! - `DelegationToken`: `KafkaPrincipal`s map through their `Type:Name`
//! string form.
use bytes::Bytes;
use krabka_protocol::{
owned::{
access_control_entry_record::AccessControlEntryRecord,
broker_registration_change_record::BrokerRegistrationChangeRecord,
client_quota_record::{ClientQuotaRecord as KClientQuotaRecord, EntityData},
config_record::ConfigRecord,
delegation_token_record::DelegationTokenRecord as KDelegationTokenRecord,
feature_level_record::FeatureLevelRecord as KFeatureLevelRecord,
no_op_record::NoOpRecord,
partition_change_record::PartitionChangeRecord as KPartitionChangeRecord,
partition_record::PartitionRecord as KPartitionRecord,
producer_ids_record::ProducerIdsRecord as KProducerIdsRecord,
register_broker_record::{
BrokerEndpoint as KBrokerEndpoint, BrokerFeature as KBrokerFeature,
RegisterBrokerRecord,
},
register_controller_record::{
ControllerEndpoint as KControllerEndpoint, ControllerFeature as KControllerFeature,
RegisterControllerRecord,
},
remove_access_control_entry_record::RemoveAccessControlEntryRecord,
remove_topic_record::RemoveTopicRecord,
remove_user_scram_credential_record::RemoveUserScramCredentialRecord,
topic_record::TopicRecord as KTopicRecord,
unregister_broker_record::UnregisterBrokerRecord as KUnregisterBrokerRecord,
unregister_controller_record::UnregisterControllerRecord as KUnregisterControllerRecord,
user_scram_credential_record::UserScramCredentialRecord,
},
primitives::uuid::Uuid as KUuid,
records::metadata::KraftMetadataRecord,
tagged_fields::{UnknownTaggedField, UnknownTaggedFields},
};
use krabka_security::{KafkaPrincipal, ListenerProtocol, SaslMechanism};
use wincode::{SchemaRead, Serialize as _, config::DefaultConfig};
use crate::{
MetadataImage,
acl::{AclEntry, AclEntryFilter, AclOperation, PatternType, PermissionType, ResourceType},
records::{
BrokerConfigRecord, BrokerEndpoint, BrokerRegistrationChangeRecord as ChangeRecord,
BrokerRegistrationRecord, ClientQuotaRecord, ControllerRegistrationRecord,
DEFAULT_BROKER_CONFIG_NODE_ID, DelegationTokenRecord, DeleteScramCredentialRecord,
DeleteTopicRecord, FeatureLevelRecord, FencingChange, GroupConfigRecord, LeaderEpoch,
LeaderRecoveryState, MetadataRecord, NodeId, PartitionDirAssignmentRecord, PartitionRecord,
ProducerIdsRecord, QuotaEntity, ScramCredentialRecord, TopicConfigRecord, TopicRecord,
UnregisterBrokerRecord, UnregisterControllerRecord,
},
};
#[derive(Debug, thiserror::Error, PartialEq, Eq)]
pub enum TranslateError {
#[error("record has no KIP-631 metadata counterpart: {0}")]
NoCounterpart(&'static str),
#[error("unknown topic id {0} on decode")]
UnknownTopicId(uuid::Uuid),
#[error("unknown topic name {0} on encode")]
UnknownTopicName(String),
/// A value that is well-formed on its own but names something the
/// metadata image does not hold in the shape the record needs: an unknown
/// partition, a broker that is not a replica, a directory list whose
/// length does not match the replicas the image holds, or an epoch that
/// the image already holds at its maximum. Two racing committed writes can legitimately
/// produce one, and every replica sees the same image in log order, so
/// every replica skips the record in the same way.
#[error("invalid reference in {field}: {detail}")]
InvalidReference { field: &'static str, detail: String },
/// A value that no build accepts, whatever the image holds: an unknown
/// enum discriminant, a negative node id or iteration count, an integer
/// that does not fit the wire width, a string that does not parse, or a
/// directory list whose length does not match the replicas that the same
/// record sets. The bytes alone decide it, and Kafka treats the same content as a
/// fatal decode failure.
#[error("invalid {field}: {detail}")]
InvalidValue { field: &'static str, detail: String },
#[error("KIP-631 value encode failed: {0}")]
Encode(String),
#[error("KIP-631 value decode failed: {0}")]
Decode(String),
/// A `RemoveAccessControlEntry{{id}}` referenced an ACL id absent from the
/// image. On the Krabka-only path the matching add always precedes the
/// remove in log order, so this means a corrupt or out-of-order log.
#[error("unknown ACL id {0} on decode")]
UnknownAclId(uuid::Uuid),
/// A `NoOpRecord` carries one tagged field at a tag in the krabka-private
/// range (1001 and up) that this build does not assign. 1002 is one: it
/// is burned and no build writes it.
#[error("unknown krabka-private NoOpRecord tag {0}")]
UnknownPrivateTag(u32),
/// The body of a krabka-private `NoOpRecord` tag starts with a version
/// other than [`PRIVATE_RECORD_VERSION`]. A newer broker wrote it, or the
/// metadata log predates krabka 1.0, whose bodies carry no version.
#[error(
"krabka-private NoOpRecord tag {tag} has record version {version}, and this build reads \
only version {PRIVATE_RECORD_VERSION}; a record without a version predates krabka 1.0, \
and its node must be reformatted"
)]
UnknownPrivateRecordVersion { tag: u32, version: i16 },
/// A krabka-private `NoOpRecord` tag carries a record that the tag table
/// assigns to another tag (`expected_tag`), or to no tag (`None`).
#[error(
"krabka-private NoOpRecord tag {tag} carries a record that belongs to tag {expected_tag:?}"
)]
PrivateTagMismatch { tag: u32, expected_tag: Option<u32> },
/// The body of a krabka-private `NoOpRecord` tag holds `trailing` bytes
/// after its record. A re-encode would drop them, so the body is refused.
#[error("krabka-private NoOpRecord tag {tag} has {trailing} bytes after its record")]
TrailingPrivateRecordBytes { tag: u32, trailing: usize },
}
// ----- uuid bridging -----
fn to_kuuid(u: uuid::Uuid) -> KUuid {
KUuid(*u.as_bytes())
}
fn from_kuuid(u: KUuid) -> uuid::Uuid {
uuid::Uuid::from_bytes(u.0)
}
// ----- SCRAM mechanism (Kafka ScramMechanism: SCRAM_SHA_256=1, SCRAM_SHA_512=2) -----
fn scram_mechanism_to_wire(m: SaslMechanism) -> Result<i8, TranslateError> {
match m {
SaslMechanism::ScramSha256 => Ok(1),
SaslMechanism::ScramSha512 => Ok(2),
other => Err(TranslateError::InvalidValue {
field: "scram mechanism",
detail: format!("{other:?} is not a SCRAM mechanism"),
}),
}
}
fn scram_mechanism_from_wire(b: i8) -> Result<SaslMechanism, TranslateError> {
match b {
1 => Ok(SaslMechanism::ScramSha256),
2 => Ok(SaslMechanism::ScramSha512),
other => Err(TranslateError::InvalidValue {
field: "scram mechanism",
detail: format!("unknown SCRAM mechanism wire byte {other}"),
}),
}
}
// ----- ACL enum <-> Kafka i8 wire discriminants -----
//
// These mirror the canonical mappings in krabka-broker's `acl_wire`
// (Kafka serializes ACL enums as `i8`). Replicated here so the metadata
// crate does not depend on the broker.
fn resource_type_to_wire(rt: ResourceType) -> i8 {
match rt {
ResourceType::Topic => 2,
ResourceType::Group => 3,
ResourceType::Cluster => 4,
ResourceType::TransactionalId => 5,
ResourceType::DelegationToken => 6,
ResourceType::User => 7,
}
}
fn resource_type_from_wire(b: i8) -> Result<ResourceType, TranslateError> {
match b {
2 => Ok(ResourceType::Topic),
3 => Ok(ResourceType::Group),
4 => Ok(ResourceType::Cluster),
5 => Ok(ResourceType::TransactionalId),
6 => Ok(ResourceType::DelegationToken),
7 => Ok(ResourceType::User),
other => Err(TranslateError::InvalidValue {
field: "acl resource_type",
detail: format!("unknown wire byte {other}"),
}),
}
}
fn pattern_type_to_wire(pt: PatternType) -> i8 {
match pt {
PatternType::Literal => 3,
PatternType::Prefixed => 4,
}
}
fn pattern_type_from_wire(b: i8) -> Result<PatternType, TranslateError> {
match b {
3 => Ok(PatternType::Literal),
4 => Ok(PatternType::Prefixed),
other => Err(TranslateError::InvalidValue {
field: "acl pattern_type",
detail: format!("unknown wire byte {other}"),
}),
}
}
fn operation_to_wire(op: AclOperation) -> i8 {
match op {
AclOperation::All => 2,
AclOperation::Read => 3,
AclOperation::Write => 4,
AclOperation::Create => 5,
AclOperation::Delete => 6,
AclOperation::Alter => 7,
AclOperation::Describe => 8,
AclOperation::ClusterAction => 9,
AclOperation::DescribeConfigs => 10,
AclOperation::AlterConfigs => 11,
AclOperation::IdempotentWrite => 12,
AclOperation::CreateTokens => 13,
AclOperation::DescribeTokens => 14,
AclOperation::TwoPhaseCommit => 15,
}
}
fn operation_from_wire(b: i8) -> Result<AclOperation, TranslateError> {
match b {
2 => Ok(AclOperation::All),
3 => Ok(AclOperation::Read),
4 => Ok(AclOperation::Write),
5 => Ok(AclOperation::Create),
6 => Ok(AclOperation::Delete),
7 => Ok(AclOperation::Alter),
8 => Ok(AclOperation::Describe),
9 => Ok(AclOperation::ClusterAction),
10 => Ok(AclOperation::DescribeConfigs),
11 => Ok(AclOperation::AlterConfigs),
12 => Ok(AclOperation::IdempotentWrite),
13 => Ok(AclOperation::CreateTokens),
14 => Ok(AclOperation::DescribeTokens),
15 => Ok(AclOperation::TwoPhaseCommit),
other => Err(TranslateError::InvalidValue {
field: "acl operation",
detail: format!("unknown wire byte {other}"),
}),
}
}
fn permission_to_wire(pt: PermissionType) -> i8 {
match pt {
PermissionType::Deny => 2,
PermissionType::Allow => 3,
}
}
fn permission_from_wire(b: i8) -> Result<PermissionType, TranslateError> {
match b {
2 => Ok(PermissionType::Deny),
3 => Ok(PermissionType::Allow),
other => Err(TranslateError::InvalidValue {
field: "acl permission_type",
detail: format!("unknown wire byte {other}"),
}),
}
}
// ----- ACL content-hash id (KIP-631 keys ACLs by Uuid; Krabka keys by content) -----
//
// KIP-631's `AccessControlEntryRecord` carries a controller-assigned `id`, and
// `RemoveAccessControlEntryRecord` deletes by that id alone. Krabka's image does
// not model ACL ids (it keys entries by their full tuple), so to keep the log
// genuinely id-keyed and round-trippable we synthesize a *deterministic*
// 128-bit id from the entry's canonical content. The add path stamps this id;
// the delete path resolves a filter to matching entries and emits one
// `RemoveAccessControlEntry{id}` each; decode rescans the image by the same hash
// to recover the entry. The derivation is self-consistent within Krabka — Slice
// 6 replaces it with Kafka's real id scheme for cross-impl fidelity.
fn acl_id(e: &AclEntry) -> KUuid {
// Canonical byte encoding: the four enum axes (i8) then the three string
// axes, each length-prefixed and separated, so distinct entries never alias.
let mut buf: Vec<u8> = vec![
resource_type_to_wire(e.resource_type).cast_unsigned(),
pattern_type_to_wire(e.pattern_type).cast_unsigned(),
operation_to_wire(e.operation).cast_unsigned(),
permission_to_wire(e.permission_type).cast_unsigned(),
];
for s in [&e.resource_name, &e.principal, &e.host] {
buf.extend_from_slice(&(s.len() as u64).to_le_bytes());
buf.extend_from_slice(s.as_bytes());
}
KUuid(fnv1a_128(&buf))
}
/// FNV-1a over 128 bits: two independent 64-bit FNV-1a lanes with distinct
/// offset bases, concatenated. It is deterministic and dependency-free, unlike
/// `DefaultHasher`, whose output is not guaranteed stable across toolchains.
fn fnv1a_128(bytes: &[u8]) -> [u8; 16] {
const PRIME: u64 = 0x0000_0100_0000_01b3;
let lane = |mut h: u64| -> u64 {
for b in bytes {
h ^= u64::from(*b);
h = h.wrapping_mul(PRIME);
}
h
};
let hi = lane(0xcbf2_9ce4_8422_2325);
let lo = lane(0x1099_5663_8a1b_c7e9);
let mut out = [0u8; 16];
out[..8].copy_from_slice(&hi.to_be_bytes());
out[8..].copy_from_slice(&lo.to_be_bytes());
out
}
/// A filter pinned to exactly one entry, with every axis set. The decode of a
/// `RemoveAccessControlEntry` uses it, so `apply` removes precisely the
/// resolved entry.
fn exact_filter(e: &AclEntry) -> AclEntryFilter {
AclEntryFilter {
resource_type: Some(e.resource_type),
resource_name: Some(e.resource_name.clone()),
pattern_type: Some(e.pattern_type),
principal: Some(e.principal.clone()),
host: Some(e.host.clone()),
operation: Some(e.operation),
permission_type: Some(e.permission_type),
}
}
// ----- security_protocol (Kafka SecurityProtocol == ListenerProtocol ordinal) -----
fn protocol_to_wire(p: ListenerProtocol) -> i16 {
match p {
ListenerProtocol::Plaintext => 0,
ListenerProtocol::Ssl => 1,
ListenerProtocol::SaslPlaintext => 2,
ListenerProtocol::SaslSsl => 3,
}
}
fn protocol_from_wire(b: i16) -> Result<ListenerProtocol, TranslateError> {
match b {
0 => Ok(ListenerProtocol::Plaintext),
1 => Ok(ListenerProtocol::Ssl),
2 => Ok(ListenerProtocol::SaslPlaintext),
3 => Ok(ListenerProtocol::SaslSsl),
other => Err(TranslateError::InvalidValue {
field: "security_protocol",
detail: format!("unknown wire value {other}"),
}),
}
}
/// Convert a single [`MetadataRecord`] to its KIP-631 counterpart.
///
/// Whole-map config variants may fan out to N `Config` records; callers that
/// must preserve every key must use [`to_kraft_records`]. For a fanned record,
/// this function returns only the first translated key.
///
/// # Errors
/// [`TranslateError::NoCounterpart`] for raft control records and delete-ACL
/// filters, [`TranslateError::UnknownTopicName`] when a topic/partition
/// references a topic absent from `image`, [`TranslateError::InvalidValue`]
/// for out-of-range enum values and node ids that exceed `i32`, and
/// [`TranslateError::InvalidReference`] for a partition absent from `image` or
/// a broker that is not one of its replicas.
pub fn to_kraft(
rec: &MetadataRecord,
image: &MetadataImage,
) -> Result<KraftMetadataRecord, TranslateError> {
to_kraft_iter(rec, image)?
.next()
.ok_or(TranslateError::NoCounterpart("empty translation"))
}
/// Convert a single [`MetadataRecord`] to one or more KIP-631 records.
///
/// Whole-map topic, client-metrics, and group configs fan out to one `Config`
/// record per set key plus one tombstone per removed key. Every other modeled
/// variant yields exactly one record.
///
/// # Errors
/// Same as [`to_kraft`].
pub fn to_kraft_records(
rec: &MetadataRecord,
image: &MetadataImage,
) -> Result<Vec<KraftMetadataRecord>, TranslateError> {
Ok(to_kraft_iter(rec, image)?.collect())
}
/// The engine/snapshot framing entrypoint: translate one [`MetadataRecord`] to
/// the KIP-631 value byte blobs to put on the log — one per fanned-out record
/// (most records yield one; whole-map configs yield one per key plus
/// tombstones, and `V1DeleteAccessControlEntry` yields one per matched ACL).
/// Multi-version metadata records are framed at the version selected by the
/// image's finalized `metadata.version`; every other modeled record frames at
/// apiVersion 0. Yields an empty `Vec` when a record
/// fans out to nothing (an empty config clear with no prior keys, or a
/// delete-ACL filter matching nothing) — callers treat an all-empty batch as a
/// committed no-op.
///
/// A missing finalized metadata.version is treated as the latest supported
/// level, matching standalone bootstrap. `Partition` and `PartitionChange` use
/// v0, v1, or v2 at the Kafka KIP-858/KIP-966 boundaries. `RegisterBroker` uses v1,
/// v2, v3, or v4 at the migration/KIP-858/KIP-1066 boundaries, as Kafka's
/// `MetadataVersion.registerBrokerRecordVersion` does. `BrokerRegistrationChange`
/// uses the lowest of v0 to v3 that carries the fields it sets, as Kafka's
/// `ReplicationControlManager` writes it, after dropping log dirs below
/// `3.7-IV2` and cordoned log dirs below `4.3-IV0`.
///
/// # Errors
/// Propagates [`to_kraft_records`] errors, plus [`TranslateError::Encode`] if
/// the KIP-631 value encoder fails.
pub fn to_kraft_values(
rec: &MetadataRecord,
image: &MetadataImage,
) -> Result<Vec<Bytes>, TranslateError> {
let metadata_version = image
.finalized_metadata_version()
.unwrap_or(crate::metadata_version::METADATA_VERSION_MAX);
to_kraft_records(rec, image)?
.into_iter()
.map(|mut kr| {
// A registration change carries log dirs only from 3.7-IV2 and
// cordoned log dirs only from 4.3-IV0, like RegisterBroker below.
if let KraftMetadataRecord::BrokerRegistrationChange(change) = &mut kr {
if metadata_version < crate::metadata_version::DIRECTORY_ASSIGNMENT_MIN_LEVEL {
change.log_dirs.clear();
}
if metadata_version < crate::metadata_version::CORDONED_LOG_DIRS_MIN_LEVEL {
change.cordoned_log_dirs = None;
}
}
let version: i16 = match &kr {
KraftMetadataRecord::BrokerRegistrationChange(change) => {
broker_registration_change_version(change)
}
KraftMetadataRecord::Partition(_) | KraftMetadataRecord::PartitionChange(_) => {
match metadata_version {
version if version >= crate::metadata_version::ELR_MIN_LEVEL => 2,
version
if version
>= crate::metadata_version::DIRECTORY_ASSIGNMENT_MIN_LEVEL =>
{
1
}
_ => 0,
}
}
KraftMetadataRecord::RegisterBroker(_) => {
if metadata_version >= crate::metadata_version::CORDONED_LOG_DIRS_MIN_LEVEL {
4
} else if metadata_version
>= crate::metadata_version::DIRECTORY_ASSIGNMENT_MIN_LEVEL
{
3
} else if metadata_version >= 8 {
2
} else {
1
}
}
// All other modeled record types frame at the defaulted v0.
_ => 0,
};
// Tagged fields remain syntactically valid on older flexible
// record versions, so the generated encoder cannot by itself
// express the semantic version gate. Strip fields introduced by
// later RegisterBroker versions before encoding a downgraded
// image; otherwise a future Krabka reader would resurrect data
// that the target Kafka version was required to discard.
if let KraftMetadataRecord::RegisterBroker(register) = &mut kr {
if version < 3 {
register.log_dirs.clear();
}
if version < 4 {
register.cordoned_log_dirs = None;
}
}
kr.encode_value(version)
.map_err(|e| TranslateError::Encode(e.to_string()))
})
.collect()
}
/// The inverse framing entrypoint.
///
/// This function decodes one KIP-631 value blob back to a [`MetadataRecord`]
/// and resolves the image-keyed fields against `image`. Those fields are the
/// topic ids, the whole-map config merge, and the ACL ids. The caller must
/// apply decoded records to `image` in log order, so each later decode sees the
/// up-to-date state.
///
/// # Errors
/// [`TranslateError::Decode`] if the envelope or the body is malformed, plus
/// any [`from_kraft`] error, such as an unknown topic id, an unknown ACL id, or
/// an unmodeled record.
pub fn from_kraft_value(
value: &[u8],
image: &MetadataImage,
) -> Result<MetadataRecord, TranslateError> {
let (rec, _version) = KraftMetadataRecord::decode_value(value)
.map_err(|e| TranslateError::Decode(e.to_string()))?;
from_kraft(&rec, image)
}
fn topic_config_to_kraft(
config: &crate::records::TopicConfigRecord,
image: &MetadataImage,
) -> Vec<KraftMetadataRecord> {
let empty = std::collections::BTreeMap::new();
let old = image.topic_config(&config.topic).unwrap_or(&empty);
let mut output = Vec::with_capacity(config.overrides.len());
for (key, value) in &config.overrides {
output.push(KraftMetadataRecord::Config(ConfigRecord {
resource_type: 2,
resource_name: config.topic.clone(),
name: key.clone(),
value: Some(value.clone()),
..Default::default()
}));
}
for key in old.keys() {
if !config.overrides.contains_key(key) {
output.push(KraftMetadataRecord::Config(ConfigRecord {
resource_type: 2,
resource_name: config.topic.clone(),
name: key.clone(),
value: None,
..Default::default()
}));
}
}
output
}
fn group_config_to_kraft(
config: &GroupConfigRecord,
image: &MetadataImage,
) -> Vec<KraftMetadataRecord> {
let empty = std::collections::BTreeMap::new();
let old = image.group_config(&config.group_id).unwrap_or(&empty);
let mut output = Vec::with_capacity(config.configs.len());
for (key, value) in &config.configs {
output.push(KraftMetadataRecord::Config(ConfigRecord {
resource_type: 32,
resource_name: config.group_id.clone(),
name: key.clone(),
value: Some(value.clone()),
..Default::default()
}));
}
for key in old.keys() {
if !config.configs.contains_key(key) {
output.push(KraftMetadataRecord::Config(ConfigRecord {
resource_type: 32,
resource_name: config.group_id.clone(),
name: key.clone(),
value: None,
..Default::default()
}));
}
}
output
}
fn client_metrics_config_to_kraft(
config: &crate::records::ClientMetricsConfigRecord,
image: &MetadataImage,
) -> Vec<KraftMetadataRecord> {
let empty = std::collections::BTreeMap::new();
let old = image.client_metrics_config(&config.name).unwrap_or(&empty);
let mut output = Vec::with_capacity(config.configs.len());
for (key, value) in &config.configs {
output.push(KraftMetadataRecord::Config(ConfigRecord {
resource_type: 16,
resource_name: config.name.clone(),
name: key.clone(),
value: Some(value.clone()),
..Default::default()
}));
}
for key in old.keys() {
if !config.configs.contains_key(key) {
output.push(KraftMetadataRecord::Config(ConfigRecord {
resource_type: 16,
resource_name: config.name.clone(),
name: key.clone(),
value: None,
..Default::default()
}));
}
}
output
}
fn delete_acl_to_kraft(
filter: &crate::AclEntryFilter,
image: &MetadataImage,
) -> Vec<KraftMetadataRecord> {
image
.all_acls()
.filter(|entry| filter.matches(entry))
.map(|entry| {
KraftMetadataRecord::RemoveAccessControlEntry(RemoveAccessControlEntryRecord {
id: acl_id(entry),
..Default::default()
})
})
.collect()
}
fn acl_to_kraft(entry: &crate::AclEntry) -> KraftMetadataRecord {
KraftMetadataRecord::AccessControlEntry(AccessControlEntryRecord {
id: acl_id(entry),
resource_type: resource_type_to_wire(entry.resource_type),
resource_name: entry.resource_name.clone(),
pattern_type: pattern_type_to_wire(entry.pattern_type),
principal: entry.principal.clone(),
host: entry.host.clone(),
operation: operation_to_wire(entry.operation),
permission_type: permission_to_wire(entry.permission_type),
..Default::default()
})
}
fn scram_to_kraft(record: &MetadataRecord) -> Result<Vec<KraftMetadataRecord>, TranslateError> {
match record {
MetadataRecord::V1ScramCredential(scram) => {
let iterations =
i32::try_from(scram.iterations).map_err(|_| TranslateError::InvalidValue {
field: "scram iterations",
detail: format!("{} exceeds i32", scram.iterations),
})?;
Ok(vec![KraftMetadataRecord::UserScramCredential(
UserScramCredentialRecord {
name: scram.user.clone(),
mechanism: scram_mechanism_to_wire(scram.mechanism)?,
salt: bytes::Bytes::copy_from_slice(&scram.salt),
stored_key: bytes::Bytes::copy_from_slice(&scram.stored_key),
server_key: bytes::Bytes::copy_from_slice(&scram.server_key),
iterations,
..Default::default()
},
)])
}
MetadataRecord::V1DeleteScramCredential(scram) => {
Ok(vec![KraftMetadataRecord::RemoveUserScramCredential(
RemoveUserScramCredentialRecord {
name: scram.user.clone(),
mechanism: scram_mechanism_to_wire(scram.mechanism)?,
..Default::default()
},
)])
}
_ => unreachable!("non-SCRAM record routed to scram_to_kraft"),
}
}
// exhaustive match over MetadataRecord
fn to_kraft_iter(
rec: &MetadataRecord,
image: &MetadataImage,
) -> Result<std::vec::IntoIter<KraftMetadataRecord>, TranslateError> {
let recs: Vec<KraftMetadataRecord> = match rec {
MetadataRecord::V1FeatureLevel(f) => {
vec![KraftMetadataRecord::FeatureLevel(KFeatureLevelRecord {
name: f.name.clone(),
feature_level: f.level,
..Default::default()
})]
}
MetadataRecord::V1BrokerRegistration(b) => {
vec![KraftMetadataRecord::RegisterBroker(
register_broker_to_kraft(b)?,
)]
}
MetadataRecord::V1ControllerRegistration(c) => {
vec![KraftMetadataRecord::RegisterController(
register_controller_to_kraft(c)?,
)]
}
MetadataRecord::V1UnregisterBroker(u) => {
vec![KraftMetadataRecord::UnregisterBroker(
KUnregisterBrokerRecord {
broker_id: node_id_to_wire(u.node_id, "broker_id")?,
broker_epoch: u.broker_epoch,
..Default::default()
},
)]
}
MetadataRecord::V1BrokerRegistrationChange(change) => {
vec![KraftMetadataRecord::BrokerRegistrationChange(
broker_registration_change_to_kraft(change)?,
)]
}
MetadataRecord::V1UnregisterController(u) => {
vec![KraftMetadataRecord::UnregisterController(
KUnregisterControllerRecord {
controller_id: node_id_to_wire(u.node_id, "controller_id")?,
..Default::default()
},
)]
}
MetadataRecord::V1ScramCredential(_) | MetadataRecord::V1DeleteScramCredential(_) => {
scram_to_kraft(rec)?
}
MetadataRecord::V1AccessControlEntry(entry) => vec![acl_to_kraft(entry)],
MetadataRecord::V1ClientQuota(q) => {
vec![KraftMetadataRecord::ClientQuota(client_quota_to_kraft(q))]
}
MetadataRecord::V1DelegationToken(t) => {
vec![KraftMetadataRecord::DelegationToken(
delegation_token_to_kraft(t),
)]
}
MetadataRecord::V1DeleteDelegationToken(t) => {
vec![KraftMetadataRecord::RemoveDelegationToken(
krabka_protocol::owned::remove_delegation_token_record::RemoveDelegationTokenRecord {
token_id: t.token_id.clone(),
..Default::default()
},
)]
}
MetadataRecord::V1BrokerConfig(c) => {
vec![KraftMetadataRecord::Config(ConfigRecord {
resource_type: 4, // ConfigResource.Type.BROKER
resource_name: if c.node_id == DEFAULT_BROKER_CONFIG_NODE_ID {
String::new()
} else {
c.node_id.to_string()
},
name: c.config_name.clone(),
value: c.config_value.clone(),
..Default::default()
})]
}
// ----- topic / partition / remove-topic / topic-config -----
MetadataRecord::V1Topic(t) => {
vec![KraftMetadataRecord::Topic(KTopicRecord {
name: t.name.clone(),
topic_id: to_kuuid(t.topic_id),
..Default::default()
})]
}
MetadataRecord::V1Partition(p) => {
let topic = image
.topic(&p.topic)
.ok_or_else(|| TranslateError::UnknownTopicName(p.topic.clone()))?;
vec![KraftMetadataRecord::Partition(partition_to_kraft(
p,
topic.topic_id,
image,
)?)]
}
MetadataRecord::V1DeleteTopic(d) => {
let topic = image
.topic(&d.name)
.ok_or_else(|| TranslateError::UnknownTopicName(d.name.clone()))?;
vec![KraftMetadataRecord::RemoveTopic(RemoveTopicRecord {
topic_id: to_kuuid(topic.topic_id),
..Default::default()
})]
}
MetadataRecord::V1TopicConfig(c) => topic_config_to_kraft(c, image),
MetadataRecord::V1GroupConfig(c) => group_config_to_kraft(c, image),
MetadataRecord::V1ProducerIds(record) => {
vec![KraftMetadataRecord::ProducerIds(KProducerIdsRecord {
broker_id: i32::try_from(record.broker_id.0).map_err(|_| {
TranslateError::InvalidValue {
field: "producer IDs broker id",
detail: format!("node_id {} exceeds i32", record.broker_id),
}
})?,
broker_epoch: record.broker_epoch,
next_producer_id: record.next_producer_id,
..KProducerIdsRecord::default()
})]
}
MetadataRecord::V1DeleteAccessControlEntry(filter) => delete_acl_to_kraft(filter, image),
// KIP-714 CLIENT_METRICS is a standard ConfigRecord resource (type 16).
MetadataRecord::V1ClientMetricsConfig(config) => {
client_metrics_config_to_kraft(config, image)
}
// KIP-858 directory-assignment delta. Kafka's standard
// PartitionChangeRecord carries the complete directories vector while
// leaving every unrelated partition field unset.
MetadataRecord::V1PartitionDirAssignment(r) => {
vec![KraftMetadataRecord::PartitionChange(
partition_dir_assignment_to_kraft(r, image)?,
)]
}
MetadataRecord::V1PartitionElr(r) => vec![KraftMetadataRecord::PartitionChange(
partition_state_to_kraft(
image,
&r.topic,
r.partition,
-1,
Some(&r.eligible_leader_replicas),
Some(&r.last_known_elr),
)?,
)],
MetadataRecord::V1PartitionRecovery(r) => vec![KraftMetadataRecord::PartitionChange(
partition_state_to_kraft(image, &r.topic, r.partition, r.state as i8, None, None)?,
)],
MetadataRecord::V1PartitionUpdate(r) => vec![KraftMetadataRecord::PartitionChange(
partition_update_to_kraft(r, image)?,
)],
// Krabka-only records ride the private carrier, at the tag that
// `private_tag` assigns.
// - `V1FeaturesEpoch` is a snapshot-only state pin. Kafka derives this
// epoch from FeatureLevel record offsets, but a materialized snapshot
// no longer contains those offsets, so the derived value is kept.
// - `V1PartitionOffsetAdvance` is a diskless offset delta.
// - KFC-9 freeze registry and break-glass proposals. KIP-631 has no
// counterpart for either, and `NoCounterpart` is not an option: the
// controller submits all three through `submit_change`, which
// encodes every record it appends.
MetadataRecord::V1FeaturesEpoch(_)
| MetadataRecord::V1PartitionOffsetAdvance(_)
| MetadataRecord::V1TopicFreeze(_)
| MetadataRecord::V1BreakGlassProposal(_)
| MetadataRecord::V1DeleteBreakGlassProposal(_) => vec![private_carrier(rec)?],
// ----- no KIP-631 metadata counterpart -----
MetadataRecord::V1Voters(_) => {
return Err(TranslateError::NoCounterpart("V1Voters"));
}
MetadataRecord::V1KRaftVersion(_) => {
return Err(TranslateError::NoCounterpart("V1KRaftVersion"));
}
};
Ok(recs.into_iter())
}
/// Krabka-private tags for records that KIP-631 has no schema for.
///
/// Each one rides as the only tagged field of a `NoOpRecord` (apiKey 20). Kafka
/// reads the record as a `NoOpRecord`: its generated reader keeps an undeclared
/// tag in `_unknownTaggedFields` (KIP-482), and `MetadataDelta` and
/// `QuorumController` skip a `NoOpRecord` on replay. So
/// `kafka-dump-log --cluster-metadata-decoder` and `kafka-metadata-shell`
/// decode every krabka log segment and checkpoint.
///
/// `NoOpRecord` declares no tagged fields. The tags start at
/// [`PRIVATE_TAG_FLOOR`], far above any tag a later Kafka schema is likely to
/// add.
///
/// # The 1.x tag table
///
/// This table is part of the krabka 1.x on-disk contract. A tag carries exactly
/// one [`MetadataRecord`] variant, and [`from_private_carrier`] refuses any
/// other variant under it. Never reassign a tag. A new private record takes a
/// new tag above 1006.
///
/// | Tag | Variant |
/// | :--- | :-------------------------------------------- |
/// | 1001 | [`MetadataRecord::V1FeaturesEpoch`] |
/// | 1002 | None. Burned: never written, never reused. |
/// | 1003 | [`MetadataRecord::V1PartitionOffsetAdvance`] |
/// | 1004 | [`MetadataRecord::V1TopicFreeze`] |
/// | 1005 | [`MetadataRecord::V1BreakGlassProposal`] |
/// | 1006 | [`MetadataRecord::V1DeleteBreakGlassProposal`]|
///
/// # The tag body
///
/// A big-endian `i16` [`PRIVATE_RECORD_VERSION`], then the wincode-serialized
/// [`MetadataRecord`].
const PRIVATE_FEATURES_EPOCH_TAG: u32 = 1001;
/// Diskless offset-advance delta carried verbatim so it stays a per-partition
/// increment on apply (never a full-record replace).
const PRIVATE_PARTITION_OFFSET_ADVANCE_TAG: u32 = 1003;
// 1002 is burned. No build writes it, and the reader refuses it as an unknown
// private tag. Do not reuse it.
/// KFC-9 write-freeze registry entry.
const PRIVATE_TOPIC_FREEZE_TAG: u32 = 1004;
/// KFC-9 break-glass proposal, carried verbatim so the approval list, the
/// consumption stamp, and the signatures all survive a snapshot.
const PRIVATE_BREAK_GLASS_PROPOSAL_TAG: u32 = 1005;
/// KFC-9 break-glass proposal tombstone, which the expiry sweep emits.
const PRIVATE_DELETE_BREAK_GLASS_PROPOSAL_TAG: u32 = 1006;
/// The lowest tag of the krabka-private range. A lone `NoOpRecord` tag below
/// it is not krabka's, so it has no counterpart; one at or above it that the
/// table does not assign is [`TranslateError::UnknownPrivateTag`].
const PRIVATE_TAG_FLOOR: u32 = 1001;
/// The tags that [`private_carrier`] writes and [`from_private_carrier`] reads.
const PRIVATE_TAGS: [u32; 5] = [
PRIVATE_FEATURES_EPOCH_TAG,
PRIVATE_PARTITION_OFFSET_ADVANCE_TAG,
PRIVATE_TOPIC_FREEZE_TAG,
PRIVATE_BREAK_GLASS_PROPOSAL_TAG,
PRIVATE_DELETE_BREAK_GLASS_PROPOSAL_TAG,
];
/// The version of a krabka-private `NoOpRecord` tag body, the big-endian `i16`
/// in front of the wincode [`MetadataRecord`]. It is part of the krabka 1.x
/// on-disk contract.
pub const PRIVATE_RECORD_VERSION: i16 = 0;
/// The tag that carries `rec`, per the table on
/// [`PRIVATE_FEATURES_EPOCH_TAG`], or `None` for a variant that does not ride
/// a private tag.
fn private_tag(rec: &MetadataRecord) -> Option<u32> {
match rec {
MetadataRecord::V1FeaturesEpoch(_) => Some(PRIVATE_FEATURES_EPOCH_TAG),
MetadataRecord::V1PartitionOffsetAdvance(_) => Some(PRIVATE_PARTITION_OFFSET_ADVANCE_TAG),
MetadataRecord::V1TopicFreeze(_) => Some(PRIVATE_TOPIC_FREEZE_TAG),
MetadataRecord::V1BreakGlassProposal(_) => Some(PRIVATE_BREAK_GLASS_PROPOSAL_TAG),
MetadataRecord::V1DeleteBreakGlassProposal(_) => {
Some(PRIVATE_DELETE_BREAK_GLASS_PROPOSAL_TAG)
}
_ => None,
}
}
/// Wraps `rec` in a `NoOpRecord` as the tagged field that [`private_tag`]
/// assigns it. See [`PRIVATE_FEATURES_EPOCH_TAG`] for the body layout and why
/// Kafka's tools can read the result.
///
/// # Errors
/// [`TranslateError::NoCounterpart`] for a variant that has no private tag,
/// and [`TranslateError::Encode`] if wincode fails.
fn private_carrier(rec: &MetadataRecord) -> Result<KraftMetadataRecord, TranslateError> {
let tag = private_tag(rec).ok_or(TranslateError::NoCounterpart("private carrier"))?;
let record = <serde_wincode::SerdeCompat<MetadataRecord>>::serialize(rec)
.map_err(|e| TranslateError::Encode(e.to_string()))?;
let mut body = Vec::with_capacity(2 + record.len());
body.extend_from_slice(&PRIVATE_RECORD_VERSION.to_be_bytes());
body.extend_from_slice(&record);
Ok(KraftMetadataRecord::NoOp(NoOpRecord {
unknown_tagged_fields: UnknownTaggedFields(vec![UnknownTaggedField {
tag,
bytes: Bytes::from(body),
}]),
}))
}
/// Reads the record that [`private_carrier`] wrapped.
///
/// # Errors
/// - [`TranslateError::NoCounterpart`] for a `NoOpRecord` that does not carry
/// exactly one tagged field at a tag of [`PRIVATE_TAG_FLOOR`] or above, such
/// as the empty KIP-835 no-op that Kafka writes.
/// - [`TranslateError::UnknownPrivateTag`] for a lone private-range tag that
/// the table does not assign, 1002 included.
/// - [`TranslateError::UnknownPrivateRecordVersion`] for a body whose version
/// is not [`PRIVATE_RECORD_VERSION`], which includes a 0.x body that has no
/// version.
/// - [`TranslateError::PrivateTagMismatch`] for a record that the table
/// assigns to another tag.
/// - [`TranslateError::TrailingPrivateRecordBytes`] for a body with bytes
/// after its record.
/// - [`TranslateError::Decode`] for a body too short to hold its version, or a
/// malformed record.
fn from_private_carrier(no_op: &NoOpRecord) -> Result<MetadataRecord, TranslateError> {
let [field] = no_op.unknown_tagged_fields.0.as_slice() else {
return Err(TranslateError::NoCounterpart("NoOp"));
};
let tag = field.tag;
if tag < PRIVATE_TAG_FLOOR {
return Err(TranslateError::NoCounterpart("NoOp"));
}
if !PRIVATE_TAGS.contains(&tag) {
return Err(TranslateError::UnknownPrivateTag(tag));
}
let Some((version, record)) = field.bytes.split_first_chunk::<2>() else {
return Err(TranslateError::Decode(format!(
"krabka-private NoOpRecord tag {tag} body is {} bytes, too short for its i16 version",
field.bytes.len()
)));
};
let version = i16::from_be_bytes(*version);
if version != PRIVATE_RECORD_VERSION {
return Err(TranslateError::UnknownPrivateRecordVersion { tag, version });
}
let mut rest: &[u8] = record;
let rec = <serde_wincode::SerdeCompat<MetadataRecord> as SchemaRead<'_, DefaultConfig>>::get(
&mut rest,
)
.map_err(|e| TranslateError::Decode(e.to_string()))?;
if !rest.is_empty() {
return Err(TranslateError::TrailingPrivateRecordBytes {
tag,
trailing: rest.len(),
});
}
let expected_tag = private_tag(&rec);
if expected_tag != Some(tag) {
return Err(TranslateError::PrivateTagMismatch { tag, expected_tag });
}
Ok(rec)
}
fn register_broker_to_kraft(
b: &BrokerRegistrationRecord,
) -> Result<RegisterBrokerRecord, TranslateError> {
// A JVM broker expects every endpoint to carry a valid listener name.
// Encode the real listeners verbatim. Only old single-listener Krabka
// records need the empty-name compatibility carrier for their separate
// top-level host/port fields.
let end_points = if b.endpoints.is_empty() {
vec![KBrokerEndpoint {
host: b.host.clone(),
port: b.port,
..Default::default()
}]
} else {
b.endpoints
.iter()
.map(|e| KBrokerEndpoint {
name: e.name.clone(),
host: e.host.clone(),
port: e.port,
security_protocol: protocol_to_wire(e.protocol),
..Default::default()
})
.collect()
};
let features = b
.features
.iter()
.map(
|(name, &(min_supported_version, max_supported_version))| KBrokerFeature {
name: name.clone(),
min_supported_version,
max_supported_version,
..Default::default()
},
)
.collect();
Ok(RegisterBrokerRecord {
broker_id: i32::try_from(b.node_id.0).map_err(|_| TranslateError::InvalidValue {
field: "broker_id",
detail: format!("node_id {} exceeds i32", b.node_id),
})?,
rack: b.rack.clone(),
end_points,
features,
log_dirs: b.log_dirs.iter().copied().map(to_kuuid).collect(),
cordoned_log_dirs: b
.cordoned_log_dirs
.as_ref()
.map(|dirs| dirs.iter().copied().map(to_kuuid).collect()),
broker_epoch: b.broker_epoch,
incarnation_id: to_kuuid(b.incarnation_id),
fenced: b.fenced,
in_controlled_shutdown: b.in_controlled_shutdown,
..Default::default()
})
}
fn register_controller_to_kraft(
controller: &ControllerRegistrationRecord,
) -> Result<RegisterControllerRecord, TranslateError> {
let controller_id =
i32::try_from(controller.node_id.0).map_err(|_| TranslateError::InvalidValue {
field: "controller_id",
detail: format!("node_id {} exceeds i32", controller.node_id),
})?;
let end_points = controller
.endpoints
.iter()
.map(|endpoint| KControllerEndpoint {
name: endpoint.name.clone(),
host: endpoint.host.clone(),
port: endpoint.port,
security_protocol: protocol_to_wire(endpoint.protocol),
..Default::default()
})
.collect();
let features = controller
.features
.iter()
.map(
|(name, &(min_supported_version, max_supported_version))| KControllerFeature {
name: name.clone(),
min_supported_version,
max_supported_version,
..Default::default()
},
)
.collect();
Ok(RegisterControllerRecord {
controller_id,
incarnation_id: to_kuuid(controller.incarnation_id),
zk_migration_ready: controller.zk_migration_ready,
end_points,
features,
..Default::default()
})
}
fn partition_to_kraft(
p: &PartitionRecord,
topic_id: uuid::Uuid,
image: &MetadataImage,
) -> Result<KPartitionRecord, TranslateError> {
let cast = |v: &[NodeId], field: &'static str| -> Result<Vec<i32>, TranslateError> {
v.iter()
.map(|n| {
i32::try_from(n.0).map_err(|_| TranslateError::InvalidValue {
field,
detail: format!("node id {n} exceeds i32"),
})
})
.collect()
};
let (eligible, last_known) = image.partition_elr(&p.topic, p.partition);
let has_elr = !eligible.is_empty() || !last_known.is_empty();
Ok(KPartitionRecord {
partition_id: p.partition,
topic_id: to_kuuid(topic_id),
replicas: cast(&p.replicas, "partition replicas")?,
isr: cast(&p.isr, "partition isr")?,
removing_replicas: cast(&p.removing_replicas, "partition removing_replicas")?,
adding_replicas: cast(&p.adding_replicas, "partition adding_replicas")?,
leader: partition_leader_to_wire(p.leader, "partition leader")?,
// Wire boundary: KRaft's KPartitionRecord carries leader epoch as a raw
// int32, so unwrap the LeaderEpoch newtype here.
leader_epoch: p.leader_epoch.get(),
partition_epoch: p.partition_epoch,
// KIP-858: per-replica log-directory assignment, carried at KRaft v1+.
directories: p.directories.iter().map(|u| to_kuuid(*u)).collect(),
leader_recovery_state: image.leader_recovery_state(&p.topic, p.partition) as i8,
eligible_leader_replicas: has_elr
.then(|| cast(eligible, "eligible leader replica"))
.transpose()?,
last_known_elr: has_elr
.then(|| cast(last_known, "last known eligible leader replica"))
.transpose()?,
..Default::default()
})
}
fn partition_dir_assignment_to_kraft(
assignment: &PartitionDirAssignmentRecord,
image: &MetadataImage,
) -> Result<KPartitionChangeRecord, TranslateError> {
let topic = image
.topic(&assignment.topic)
.ok_or_else(|| TranslateError::UnknownTopicName(assignment.topic.clone()))?;
let partition = image
.partition(&assignment.topic, assignment.partition)
.ok_or_else(|| TranslateError::InvalidReference {
field: "partition directory assignment",
detail: format!(
"unknown partition {}-{}",
assignment.topic, assignment.partition
),
})?;
let replica_slot = partition
.replicas
.iter()
.position(|replica| *replica == assignment.replica)
.ok_or_else(|| TranslateError::InvalidReference {
field: "partition directory assignment",
detail: format!(
"broker {} is not a replica of {}-{}",
assignment.replica, assignment.topic, assignment.partition
),
})?;
// An empty list is every replica `MIGRATING`, as Kafka's
// `PartitionRegistration.defaultToMigrating` reads it.
let mut directories = partition.directories.clone();
directories.resize(partition.replicas.len(), crate::directory_id::MIGRATING);
directories[replica_slot] = assignment.directory;
Ok(KPartitionChangeRecord {
partition_id: assignment.partition,
topic_id: to_kuuid(topic.topic_id),
directories: Some(directories.into_iter().map(to_kuuid).collect()),
..Default::default()
})
}
fn partition_state_to_kraft(
image: &MetadataImage,
topic: &str,
partition: i32,
leader_recovery_state: i8,
eligible: Option<&[NodeId]>,
last_known: Option<&[NodeId]>,
) -> Result<KPartitionChangeRecord, TranslateError> {
let topic_id = image
.topic(topic)
.ok_or_else(|| TranslateError::UnknownTopicName(topic.to_string()))?
.topic_id;
let nodes = |values: &[NodeId], field| {
values
.iter()
.map(|node| {
i32::try_from(node.0).map_err(|_| TranslateError::InvalidValue {
field,
detail: format!("node id {node} exceeds i32"),
})
})
.collect::<Result<Vec<_>, _>>()
};
Ok(KPartitionChangeRecord {
partition_id: partition,
topic_id: to_kuuid(topic_id),
leader_recovery_state,
eligible_leader_replicas: eligible
.map(|values| nodes(values, "eligible leader replica"))
.transpose()?,
last_known_elr: last_known
.map(|values| nodes(values, "last known eligible leader replica"))
.transpose()?,
..Default::default()
})
}
fn partition_update_to_kraft(
update: &crate::PartitionUpdateRecord,
image: &MetadataImage,
) -> Result<KPartitionChangeRecord, TranslateError> {
let current = image
.partition(&update.partition.topic, update.partition.partition)
.ok_or_else(|| TranslateError::InvalidReference {
field: "partition update",
detail: format!(
"unknown partition {}-{}",
update.partition.topic, update.partition.partition
),
})?;
let mut change = partition_state_to_kraft(
image,
&update.partition.topic,
update.partition.partition,
update.recovery_state.map_or(-1, |state| state as i8),
update.eligible_leader_replicas.as_deref(),
update.last_known_elr.as_deref(),
)?;
let cast = |values: &[NodeId], field| {
values
.iter()
.map(|node| {
i32::try_from(node.0).map_err(|_| TranslateError::InvalidValue {
field,
detail: format!("node id {node} exceeds i32"),
})
})
.collect::<Result<Vec<_>, _>>()
};
if update.partition.replicas != current.replicas {
change.replicas = Some(cast(&update.partition.replicas, "partition replicas")?);
}
if update.partition.isr != current.isr {
change.isr = Some(cast(&update.partition.isr, "partition isr")?);
}
if update.partition.removing_replicas != current.removing_replicas {
change.removing_replicas = Some(cast(
&update.partition.removing_replicas,
"partition removing replicas",
)?);
}
if update.partition.adding_replicas != current.adding_replicas {
change.adding_replicas = Some(cast(
&update.partition.adding_replicas,
"partition adding replicas",
)?);
}
if update.partition.leader != current.leader {
change.leader = partition_leader_to_wire(update.partition.leader, "partition leader")?;
}
if update.partition.directories != current.directories {
change.directories = Some(
update
.partition
.directories
.iter()
.copied()
.map(to_kuuid)
.collect(),
);
}
Ok(change)
}
fn client_quota_to_kraft(q: &ClientQuotaRecord) -> KClientQuotaRecord {
let entity = q
.entity
.iter()
.map(|e| EntityData {
entity_type: e.entity_type.clone(),
entity_name: e.entity_name.clone(),
..Default::default()
})
.collect();
// KIP-631 splits Krabka's Option<f64> into (value, remove): None => the
// remove tombstone; Some(v) => value with remove=false.
let (value, remove) = match q.config_value {
Some(v) => (v, false),
None => (0.0, true),
};
KClientQuotaRecord {
entity,
key: q.config_key.clone(),
value,
remove,
..Default::default()
}
}
fn delegation_token_to_kraft(t: &DelegationTokenRecord) -> KDelegationTokenRecord {
KDelegationTokenRecord {
owner: t.owner.to_string(),
requester: t.requester.to_string(),
renewers: t.renewers.iter().map(ToString::to_string).collect(),
issue_timestamp: t.issue_timestamp_ms,
max_timestamp: t.max_timestamp_ms,
expiration_timestamp: t.expiry_timestamp_ms,
token_id: t.token_id.clone(),
..Default::default()
}
}
/// Convert one KIP-631 record back to its [`MetadataRecord`] counterpart.
///
/// Inverse of [`to_kraft`]. A `Config` record routes back to
/// `V1TopicConfig` or `V1BrokerConfig` by `resource_type`. Topic-config
/// records decode to single-key singletons that the image merges.
///
/// # Errors
/// [`TranslateError::UnknownTopicId`] when a topic or partition references a
/// topic id absent from `image`, [`TranslateError::NoCounterpart`] for
/// records this layer does not model, [`TranslateError::InvalidValue`] for
/// out-of-range enum values, negative ids, unparsable strings and a directory
/// list whose length does not match the replicas in the same record, and
/// [`TranslateError::InvalidReference`] for a partition change that the image
/// cannot apply: an unknown partition, a directory list whose length does not
/// match the image's replicas, or an epoch already at its maximum.
// exhaustive match over KraftMetadataRecord
pub fn from_kraft(
rec: &KraftMetadataRecord,
image: &MetadataImage,
) -> Result<MetadataRecord, TranslateError> {
match rec {
KraftMetadataRecord::FeatureLevel(f) => {
Ok(MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
name: f.name.clone(),
level: f.feature_level,
}))
}
KraftMetadataRecord::RegisterBroker(b) => Ok(MetadataRecord::V1BrokerRegistration(
register_broker_from_kraft(b)?,
)),
KraftMetadataRecord::BrokerRegistrationChange(change) => {
Ok(MetadataRecord::V1BrokerRegistrationChange(
broker_registration_change_from_kraft(change)?,
))
}
KraftMetadataRecord::RegisterController(c) => Ok(MetadataRecord::V1ControllerRegistration(
register_controller_from_kraft(c)?,
)),
KraftMetadataRecord::UnregisterController(u) => Ok(MetadataRecord::V1UnregisterController(
UnregisterControllerRecord {
node_id: node_id_from_wire(u.controller_id, "unregister controller id")?,
},
)),
KraftMetadataRecord::UnregisterBroker(u) => {
Ok(MetadataRecord::V1UnregisterBroker(UnregisterBrokerRecord {
node_id: node_id_from_wire(u.broker_id, "unregister broker id")?,
broker_epoch: u.broker_epoch,
}))
}
KraftMetadataRecord::UserScramCredential(s) => {
Ok(MetadataRecord::V1ScramCredential(ScramCredentialRecord {
user: s.name.clone(),
mechanism: scram_mechanism_from_wire(s.mechanism)?,
salt: s.salt.to_vec(),
stored_key: s.stored_key.to_vec(),
server_key: s.server_key.to_vec(),
iterations: u32::try_from(s.iterations).map_err(|_| {
TranslateError::InvalidValue {
field: "scram iterations",
detail: format!("{} is negative", s.iterations),
}
})?,
}))
}
KraftMetadataRecord::RemoveUserScramCredential(s) => Ok(
MetadataRecord::V1DeleteScramCredential(DeleteScramCredentialRecord {
user: s.name.clone(),
mechanism: scram_mechanism_from_wire(s.mechanism)?,
}),
),
KraftMetadataRecord::AccessControlEntry(e) => {
Ok(MetadataRecord::V1AccessControlEntry(AclEntry {
resource_type: resource_type_from_wire(e.resource_type)?,
resource_name: e.resource_name.clone(),
pattern_type: pattern_type_from_wire(e.pattern_type)?,
principal: e.principal.clone(),
host: e.host.clone(),
operation: operation_from_wire(e.operation)?,
permission_type: permission_from_wire(e.permission_type)?,
}))
}
KraftMetadataRecord::ClientQuota(q) => {
Ok(MetadataRecord::V1ClientQuota(client_quota_from_kraft(q)))
}
KraftMetadataRecord::ProducerIds(record) => {
Ok(MetadataRecord::V1ProducerIds(ProducerIdsRecord {
broker_id: node_id_from_wire(record.broker_id, "producer IDs broker id")?,
broker_epoch: record.broker_epoch,
next_producer_id: record.next_producer_id,
}))
}
KraftMetadataRecord::DelegationToken(t) => Ok(MetadataRecord::V1DelegationToken(
delegation_token_from_kraft(t)?,
)),
KraftMetadataRecord::RemoveDelegationToken(t) => Ok(
MetadataRecord::V1DeleteDelegationToken(crate::records::DeleteDelegationTokenRecord {
token_id: t.token_id.clone(),
}),
),
KraftMetadataRecord::Config(c) => config_from_kraft(c, image),
KraftMetadataRecord::Topic(t) => Ok(MetadataRecord::V1Topic(topic_from_kraft(t, image))),
KraftMetadataRecord::Partition(p) => partition_from_kraft(p, image),
KraftMetadataRecord::PartitionChange(p) => partition_change_from_kraft(p, image),
KraftMetadataRecord::RemoveTopic(t) => {
let id = from_kuuid(t.topic_id);
let name = topic_name_for_id(image, id).ok_or(TranslateError::UnknownTopicId(id))?;
Ok(MetadataRecord::V1DeleteTopic(DeleteTopicRecord { name }))
}
KraftMetadataRecord::RemoveAccessControlEntry(r) => {
// Resolve the id back to the concrete entry by rescanning the image
// for the entry whose content-hash matches, then emit a pinned
// (every-axis) filter so `apply` removes exactly that entry. The
// matching add always precedes this remove in log order, so the
// entry is present when this decodes.
let entry = image
.all_acls()
.find(|e| acl_id(e) == r.id)
.ok_or_else(|| TranslateError::UnknownAclId(from_kuuid(r.id)))?;
Ok(MetadataRecord::V1DeleteAccessControlEntry(exact_filter(
entry,
)))
}
// Krabka-private records ride a `NoOpRecord` tagged field (see
// `private_carrier`).
KraftMetadataRecord::NoOp(no_op) => from_private_carrier(no_op),
other => Err(TranslateError::NoCounterpart(kraft_variant_name(other))),
}
}
fn kraft_variant_name(rec: &KraftMetadataRecord) -> &'static str {
match rec {
KraftMetadataRecord::BrokerRegistrationChange(_) => "BrokerRegistrationChange",
KraftMetadataRecord::PartitionChange(_) => "PartitionChange",
KraftMetadataRecord::NoOp(_) => "NoOp",
KraftMetadataRecord::BeginTransaction(_) => "BeginTransaction",
KraftMetadataRecord::EndTransaction(_) => "EndTransaction",
KraftMetadataRecord::RemoveAccessControlEntry(_) => "RemoveAccessControlEntry",
KraftMetadataRecord::Unknown { .. } => "Unknown",
_ => "unmodeled",
}
}
fn register_broker_from_kraft(
b: &RegisterBrokerRecord,
) -> Result<BrokerRegistrationRecord, TranslateError> {
// Older Krabka logs used an empty-name leading endpoint solely to carry
// the legacy top-level pair. New Krabka and JVM records contain only real
// listener endpoints; in that case the first endpoint supplies the
// legacy fields while every endpoint stays in the listener list.
let (host, port, endpoints_on_wire) = match b.end_points.split_first() {
Some((head, rest)) if head.name.is_empty() => (head.host.clone(), head.port, rest),
Some((head, _)) => (head.host.clone(), head.port, b.end_points.as_slice()),
None => (String::new(), 0, &[][..]),
};
let endpoints = endpoints_on_wire
.iter()
.map(|e| {
Ok(BrokerEndpoint {
name: e.name.clone(),
host: e.host.clone(),
port: e.port,
protocol: protocol_from_wire(e.security_protocol)?,
})
})
.collect::<Result<Vec<_>, TranslateError>>()?;
let features = b
.features
.iter()
.map(|feature| {
(
feature.name.clone(),
(feature.min_supported_version, feature.max_supported_version),
)
})
.collect();
let log_dirs = b.log_dirs.iter().copied().map(from_kuuid).collect();
Ok(BrokerRegistrationRecord {
node_id: node_id_from_wire(b.broker_id, "register broker id")?,
broker_epoch: b.broker_epoch,
incarnation_id: from_kuuid(b.incarnation_id),
host,
port,
rack: b.rack.clone(),
endpoints,
log_dirs,
fenced: b.fenced,
in_controlled_shutdown: b.in_controlled_shutdown,
cordoned_log_dirs: b
.cordoned_log_dirs
.as_ref()
.map(|dirs| dirs.iter().copied().map(from_kuuid).collect()),
features,
})
}
/// Reads a `BrokerRegistrationChangeRecord`. `fenced` must be -1 (unfence), 0
/// (no change) or 1 (fence) and `inControlledShutdown` 0 (no change) or 1, as
/// Kafka's `ClusterControlManager.replay` requires. The epoch check against
/// the registration belongs to [`MetadataImage::validate`] and
/// [`MetadataImage::apply`], which see the image in log order.
fn broker_registration_change_from_kraft(
change: &BrokerRegistrationChangeRecord,
) -> Result<ChangeRecord, TranslateError> {
let fenced =
FencingChange::from_wire(change.fenced).ok_or_else(|| TranslateError::InvalidValue {
field: "broker registration change fenced",
detail: format!("unknown value {}", change.fenced),
})?;
let in_controlled_shutdown = match change.in_controlled_shutdown {
0 => false,
1 => true,
other => {
return Err(TranslateError::InvalidValue {
field: "broker registration change in_controlled_shutdown",
detail: format!("unknown value {other}"),
});
}
};
Ok(ChangeRecord {
node_id: node_id_from_wire(change.broker_id, "broker registration change id")?,
broker_epoch: change.broker_epoch,
fenced,
in_controlled_shutdown,
log_dirs: change.log_dirs.iter().copied().map(from_kuuid).collect(),
cordoned_log_dirs: change
.cordoned_log_dirs
.as_ref()
.map(|dirs| dirs.iter().copied().map(from_kuuid).collect()),
})
}
/// Writes a `BrokerRegistrationChangeRecord`. [`to_kraft_values`] frames it at
/// the lowest version that carries every field it sets, as Kafka's
/// `ReplicationControlManager` does.
fn broker_registration_change_to_kraft(
change: &ChangeRecord,
) -> Result<BrokerRegistrationChangeRecord, TranslateError> {
Ok(BrokerRegistrationChangeRecord {
broker_id: node_id_to_wire(change.node_id, "broker_id")?,
broker_epoch: change.broker_epoch,
fenced: change.fenced.wire_value(),
in_controlled_shutdown: i8::from(change.in_controlled_shutdown),
log_dirs: change.log_dirs.iter().copied().map(to_kuuid).collect(),
cordoned_log_dirs: change
.cordoned_log_dirs
.as_ref()
.map(|dirs| dirs.iter().copied().map(to_kuuid).collect()),
..Default::default()
})
}
/// The `BrokerRegistrationChangeRecord` version Kafka's
/// `ReplicationControlManager` frames a change at: v0 for a fencing change,
/// v1 once it sets `InControlledShutdown`, v2 once it sets `LogDirs` and v3
/// once it sets `CordonedLogDirs`.
fn broker_registration_change_version(change: &BrokerRegistrationChangeRecord) -> i16 {
if change.cordoned_log_dirs.is_some() {
3
} else if !change.log_dirs.is_empty() {
2
} else {
i16::from(change.in_controlled_shutdown != 0)
}
}
fn register_controller_from_kraft(
controller: &RegisterControllerRecord,
) -> Result<ControllerRegistrationRecord, TranslateError> {
let endpoints = controller
.end_points
.iter()
.map(|endpoint| {
Ok(BrokerEndpoint {
name: endpoint.name.clone(),
host: endpoint.host.clone(),
port: endpoint.port,
protocol: protocol_from_wire(endpoint.security_protocol)?,
})
})
.collect::<Result<Vec<_>, TranslateError>>()?;
let features = controller
.features
.iter()
.map(|feature| {
(
feature.name.clone(),
(feature.min_supported_version, feature.max_supported_version),
)
})
.collect();
Ok(ControllerRegistrationRecord {
node_id: node_id_from_wire(controller.controller_id, "controller id")?,
incarnation_id: from_kuuid(controller.incarnation_id),
zk_migration_ready: controller.zk_migration_ready,
endpoints,
features,
})
}
fn topic_from_kraft(t: &KTopicRecord, image: &MetadataImage) -> TopicRecord {
let name = t.name.clone();
// Reconstruct partitions/RF from the image: partition count is derived
// from the partitions map; RF from the first partition's replica count
// (1 if none yet present). These fields are not part of the KIP-631
// record — the image is authoritative.
let partitions = image.topic_partition_count(&name);
let replication_factor = image
.partitions_of(&name)
.next()
.map_or(1, |p| i16::try_from(p.replicas.len()).unwrap_or(1));
TopicRecord {
name,
topic_id: from_kuuid(t.topic_id),
partitions,
replication_factor,
}
}
fn partition_from_kraft(
p: &KPartitionRecord,
image: &MetadataImage,
) -> Result<MetadataRecord, TranslateError> {
let id = from_kuuid(p.topic_id);
let topic = topic_name_for_id(image, id).ok_or(TranslateError::UnknownTopicId(id))?;
let nodes = |values: &[i32], field| {
values
.iter()
.map(|value| node_id_from_wire(*value, field))
.collect::<Result<Vec<_>, _>>()
};
let partition = PartitionRecord {
topic,
partition: p.partition_id,
leader: partition_leader_from_wire(p.leader, "partition leader")?,
replicas: nodes(&p.replicas, "partition replica")?,
isr: nodes(&p.isr, "partition ISR member")?,
// Wire boundary: wrap the raw int32 leader epoch decoded off KRaft.
leader_epoch: LeaderEpoch(p.leader_epoch),
partition_epoch: p.partition_epoch,
adding_replicas: nodes(&p.adding_replicas, "adding replica")?,
removing_replicas: nodes(&p.removing_replicas, "removing replica")?,
// KIP-858: per-replica log-directory assignment, present at KRaft v1+.
directories: p.directories.iter().map(|u| from_kuuid(*u)).collect(),
};
let recovery_state = match p.leader_recovery_state {
0 => LeaderRecoveryState::Recovered,
1 => LeaderRecoveryState::Recovering,
value => {
return Err(TranslateError::InvalidValue {
field: "partition leader recovery state",
detail: format!("unknown state {value}"),
});
}
};
if p.eligible_leader_replicas.is_some()
|| p.last_known_elr.is_some()
|| recovery_state == LeaderRecoveryState::Recovering
{
Ok(MetadataRecord::V1PartitionUpdate(
crate::PartitionUpdateRecord {
partition,
eligible_leader_replicas: p
.eligible_leader_replicas
.as_deref()
.map(|values| nodes(values, "eligible leader replica"))
.transpose()?,
last_known_elr: p
.last_known_elr
.as_deref()
.map(|values| nodes(values, "last known eligible leader replica"))
.transpose()?,
recovery_state: Some(recovery_state),
},
))
} else {
Ok(MetadataRecord::V1Partition(partition))
}
}
fn partition_change_from_kraft(
change: &KPartitionChangeRecord,
image: &MetadataImage,
) -> Result<MetadataRecord, TranslateError> {
let topic_id = from_kuuid(change.topic_id);
let topic =
topic_name_for_id(image, topic_id).ok_or(TranslateError::UnknownTopicId(topic_id))?;
let has_recovery = change.leader_recovery_state != -1;
let has_elr = change.eligible_leader_replicas.is_some() || change.last_known_elr.is_some();
let nodes = |values: &[i32], field| {
values
.iter()
.map(|value| node_id_from_wire(*value, field))
.collect::<Result<Vec<_>, _>>()
};
let recovery_state = if has_recovery {
Some(match change.leader_recovery_state {
0 => LeaderRecoveryState::Recovered,
1 => LeaderRecoveryState::Recovering,
value => {
return Err(TranslateError::InvalidValue {
field: "partition change leader recovery state",
detail: format!("unknown state {value}"),
});
}
})
} else {
None
};
let mut partition = image
.partition(&topic, change.partition_id)
.cloned()
.ok_or_else(|| TranslateError::InvalidReference {
field: "partition change",
detail: format!("unknown partition {topic}-{}", change.partition_id),
})?;
if let Some(replicas) = &change.replicas {
partition.replicas = nodes(replicas, "partition change replica")?;
}
if let Some(isr) = &change.isr {
partition.isr = nodes(isr, "partition change ISR member")?;
}
if let Some(removing) = &change.removing_replicas {
partition.removing_replicas = nodes(removing, "partition change removing replica")?;
}
if let Some(adding) = &change.adding_replicas {
partition.adding_replicas = nodes(adding, "partition change adding replica")?;
}
if change.leader != -2 {
partition.leader = partition_leader_from_wire(change.leader, "partition change leader")?;
// A reference error: the record carries no epoch, so the overflow
// comes from the epoch the image holds, not from the bytes.
partition.leader_epoch =
LeaderEpoch(partition.leader_epoch.get().checked_add(1).ok_or_else(|| {
TranslateError::InvalidReference {
field: "partition change leader epoch",
detail: "leader epoch overflow".into(),
}
})?);
}
if let Some(directories) = &change.directories {
if directories.len() != partition.replicas.len() {
let field = "partition change directories";
let detail = format!(
"{} directories for {} replicas",
directories.len(),
partition.replicas.len()
);
// When the same record replaces the replicas, the replicas were
// set from its bytes above, so the bytes alone decide the
// mismatch. Otherwise the length to match is the image's replica
// count, and the mismatch depends on the image.
return Err(if change.replicas.is_some() {
TranslateError::InvalidValue { field, detail }
} else {
TranslateError::InvalidReference { field, detail }
});
}
partition.directories = directories.iter().copied().map(from_kuuid).collect();
}
// A reference error, like the leader epoch: the overflow comes from the
// partition epoch the image holds.
partition.partition_epoch = partition.partition_epoch.checked_add(1).ok_or_else(|| {
TranslateError::InvalidReference {
field: "partition change partition epoch",
detail: "partition epoch overflow".into(),
}
})?;
if has_recovery || has_elr {
Ok(MetadataRecord::V1PartitionUpdate(
crate::PartitionUpdateRecord {
partition,
eligible_leader_replicas: change
.eligible_leader_replicas
.as_deref()
.map(|values| nodes(values, "eligible leader replica"))
.transpose()?,
last_known_elr: change
.last_known_elr
.as_deref()
.map(|values| nodes(values, "last known eligible leader replica"))
.transpose()?,
recovery_state,
},
))
} else {
Ok(MetadataRecord::V1Partition(partition))
}
}
fn partition_leader_to_wire(leader: NodeId, field: &'static str) -> Result<i32, TranslateError> {
if leader == NodeId(0) {
return Ok(-1);
}
i32::try_from(leader.0).map_err(|_| TranslateError::InvalidValue {
field,
detail: format!("leader {leader} exceeds i32"),
})
}
fn partition_leader_from_wire(value: i32, field: &'static str) -> Result<NodeId, TranslateError> {
if value == -1 {
return Ok(NodeId(0));
}
node_id_from_wire(value, field)
}
fn node_id_from_wire(value: i32, field: &'static str) -> Result<NodeId, TranslateError> {
u64::try_from(value)
.map(NodeId)
.map_err(|_| TranslateError::InvalidValue {
field,
detail: format!("{value} is negative"),
})
}
fn node_id_to_wire(node_id: NodeId, field: &'static str) -> Result<i32, TranslateError> {
i32::try_from(node_id.0).map_err(|_| TranslateError::InvalidValue {
field,
detail: format!("node_id {node_id} exceeds i32"),
})
}
fn client_quota_from_kraft(q: &KClientQuotaRecord) -> ClientQuotaRecord {
let entity = q
.entity
.iter()
.map(|e| QuotaEntity {
entity_type: e.entity_type.clone(),
entity_name: e.entity_name.clone(),
})
.collect();
let config_value = if q.remove { None } else { Some(q.value) };
ClientQuotaRecord {
entity,
config_key: q.key.clone(),
config_value,
}
}
fn delegation_token_from_kraft(
t: &KDelegationTokenRecord,
) -> Result<DelegationTokenRecord, TranslateError> {
let owner = t
.owner
.parse::<KafkaPrincipal>()
.map_err(|e| TranslateError::InvalidValue {
field: "delegation token owner",
detail: e,
})?;
let requester =
t.requester
.parse::<KafkaPrincipal>()
.map_err(|e| TranslateError::InvalidValue {
field: "delegation token requester",
detail: e,
})?;
let renewers = t
.renewers
.iter()
.map(|r| {
r.parse::<KafkaPrincipal>()
.map_err(|e| TranslateError::InvalidValue {
field: "delegation token renewer",
detail: e,
})
})
.collect::<Result<Vec<_>, _>>()?;
Ok(DelegationTokenRecord {
token_id: t.token_id.clone(),
owner,
requester,
issue_timestamp_ms: t.issue_timestamp,
expiry_timestamp_ms: t.expiration_timestamp,
max_timestamp_ms: t.max_timestamp,
renewers,
})
}
fn config_from_kraft(
c: &ConfigRecord,
image: &MetadataImage,
) -> Result<MetadataRecord, TranslateError> {
match c.resource_type {
4 => {
// BROKER — `V1BrokerConfig` is already per-key (Some=set, None=delete),
// so a single `ConfigRecord` maps 1:1 (the image merges per-key).
let node_id = if c.resource_name.is_empty() {
DEFAULT_BROKER_CONFIG_NODE_ID
} else {
NodeId(c.resource_name.parse::<u64>().map_err(|e| {
TranslateError::InvalidValue {
field: "broker config resource_name",
detail: e.to_string(),
}
})?)
};
Ok(MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
node_id,
config_name: c.name.clone(),
config_value: c.value.clone(),
}))
}
2 => {
// TOPIC — `V1TopicConfig` is whole-map (replace) authoritative, so
// merge this one key's delta onto the image's current overrides and
// emit the resulting whole map. `Some(v)` sets the key; `None` (a
// tombstone) removes it. Applying the emitted record (replace) lands
// the merged map. Decoding successive Config records against the
// live image (engine apply / snapshot read) reproduces the original
// whole map exactly.
let mut overrides = image
.topic_config(&c.resource_name)
.cloned()
.unwrap_or_default();
match &c.value {
Some(v) => {
overrides.insert(c.name.clone(), v.clone());
}
None => {
overrides.remove(&c.name);
}
}
Ok(MetadataRecord::V1TopicConfig(TopicConfigRecord {
topic: c.resource_name.clone(),
overrides,
}))
}
16 => {
let mut configs = image
.client_metrics_config(&c.resource_name)
.cloned()
.unwrap_or_default();
match &c.value {
Some(v) => {
configs.insert(c.name.clone(), v.clone());
}
None => {
configs.remove(&c.name);
}
}
Ok(MetadataRecord::V1ClientMetricsConfig(
crate::records::ClientMetricsConfigRecord {
name: c.resource_name.clone(),
configs,
},
))
}
32 => {
let mut configs = image
.group_config(&c.resource_name)
.cloned()
.unwrap_or_default();
match &c.value {
Some(v) => {
configs.insert(c.name.clone(), v.clone());
}
None => {
configs.remove(&c.name);
}
}
Ok(MetadataRecord::V1GroupConfig(GroupConfigRecord {
group_id: c.resource_name.clone(),
configs,
}))
}
other => Err(TranslateError::InvalidValue {
field: "config resource_type",
detail: format!("unsupported resource_type {other}"),
}),
}
}
/// Resolve a topic id to its name with a scan of the topics of the image.
fn topic_name_for_id(image: &MetadataImage, id: uuid::Uuid) -> Option<String> {
image
.topics()
.find(|t| t.topic_id == id)
.map(|t| t.name.clone())
}
#[cfg(test)]
mod tests {
use assert2::check;
use super::*;
use crate::{
acl::AclEntryFilter,
records::{
ClientMetricsConfigRecord, DeleteDelegationTokenRecord, FeaturesEpochRecord,
KRaftVersionRecord, PartitionDirAssignmentRecord, VotersRecord,
},
};
fn img() -> MetadataImage {
MetadataImage::new(uuid::Uuid::nil())
}
fn round_trip(rec: &MetadataRecord, image: &MetadataImage) {
let k = to_kraft(rec, image).unwrap();
assert2::assert!(from_kraft(&k, image).unwrap() == *rec);
}
// ---------- aligned variants ----------
#[test]
fn feature_level_round_trips() {
let rec = MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
name: "metadata.version".into(),
level: 25,
});
round_trip(&rec, &img());
}
#[test]
fn register_broker_no_endpoints_round_trips() {
// Non-zero `broker_epoch` (KIP-903): `round_trip` asserts full equality,
// so the field is dropped iff this fails (it would decode back as 0).
round_trip(
&MetadataRecord::V1BrokerRegistration(BrokerRegistrationRecord {
node_id: NodeId(7),
broker_epoch: 42,
incarnation_id: uuid::Uuid::from_u128(0xdeadbeef_cafe_babe_0123_456789abcdef),
host: "192.168.1.10".into(),
port: 9092,
rack: Some("us-east-1a".into()),
log_dirs: vec![],
fenced: false,
in_controlled_shutdown: false,
cordoned_log_dirs: None,
endpoints: vec![],
features: std::collections::BTreeMap::new(),
}),
&img(),
);
}
#[test]
fn register_broker_with_endpoints_round_trips() {
// Non-zero `broker_epoch` (KIP-903) carried alongside the endpoints;
// `round_trip`'s full-equality assert covers the field.
round_trip(
&MetadataRecord::V1BrokerRegistration(BrokerRegistrationRecord {
node_id: NodeId(1),
broker_epoch: 7,
incarnation_id: uuid::Uuid::from_u128(0xfeedface_0000_0000_0000_000000000001),
host: "ext.example.com".into(),
port: 9093,
rack: None,
log_dirs: vec![],
fenced: false,
in_controlled_shutdown: false,
cordoned_log_dirs: None,
endpoints: vec![
BrokerEndpoint {
name: "EXTERNAL".into(),
host: "ext.example.com".into(),
port: 9093,
protocol: ListenerProtocol::SaslSsl,
},
BrokerEndpoint {
name: "INTERNAL".into(),
host: "int".into(),
port: 9094,
protocol: ListenerProtocol::Plaintext,
},
],
features: std::collections::BTreeMap::new(),
}),
&img(),
);
}
#[test]
fn register_broker_to_kraft_preserves_endpoint_details_and_unfenced_state() {
let rec = MetadataRecord::V1BrokerRegistration(BrokerRegistrationRecord {
node_id: NodeId(1),
broker_epoch: 7,
incarnation_id: uuid::Uuid::from_u128(0xfeedface_0000_0000_0000_000000000001),
host: "broker.local".into(),
port: 9092,
rack: None,
log_dirs: vec![],
fenced: false,
in_controlled_shutdown: false,
cordoned_log_dirs: None,
endpoints: vec![BrokerEndpoint {
name: "EXTERNAL".into(),
host: "ext.example.com".into(),
port: 9093,
protocol: ListenerProtocol::SaslSsl,
}],
features: std::collections::BTreeMap::new(),
});
let KraftMetadataRecord::RegisterBroker(k) = to_kraft(&rec, &img()).unwrap() else {
panic!("expected RegisterBroker");
};
check!(
(
k.fenced,
k.end_points
.iter()
.map(|endpoint| (endpoint.name.as_str(), endpoint.security_protocol))
.collect::<Vec<_>>(),
) == (
false,
vec![("EXTERNAL", protocol_to_wire(ListenerProtocol::SaslSsl),)],
)
);
}
#[test]
fn controller_registration_round_trips() {
round_trip(
&MetadataRecord::V1ControllerRegistration(ControllerRegistrationRecord {
node_id: NodeId(3),
incarnation_id: uuid::Uuid::from_u128(3),
zk_migration_ready: false,
endpoints: vec![BrokerEndpoint {
name: "CONTROLLER".into(),
host: "controller-3".into(),
port: 9093,
protocol: ListenerProtocol::Plaintext,
}],
features: std::collections::BTreeMap::from([("metadata.version".into(), (7, 25))]),
}),
&img(),
);
}
#[test]
fn register_broker_incarnation_id_survives_round_trip() {
let id = uuid::Uuid::from_u128(0x0102_0304_0506_0708_090a_0b0c_0d0e_0f10);
round_trip(
&MetadataRecord::V1BrokerRegistration(BrokerRegistrationRecord {
node_id: NodeId(3),
broker_epoch: 5,
incarnation_id: id,
host: "127.0.0.1".into(),
port: 9092,
rack: None,
log_dirs: vec![],
fenced: false,
in_controlled_shutdown: false,
cordoned_log_dirs: None,
endpoints: vec![],
features: std::collections::BTreeMap::new(),
}),
&img(),
);
}
#[test]
fn register_broker_log_dirs_follow_metadata_version_record_gate() {
let directory = uuid::Uuid::from_u128(0xD1);
let record = MetadataRecord::V1BrokerRegistration(BrokerRegistrationRecord {
node_id: NodeId(3),
broker_epoch: 5,
incarnation_id: uuid::Uuid::from_u128(3),
host: "127.0.0.1".into(),
port: 9092,
rack: None,
endpoints: vec![],
log_dirs: vec![directory],
fenced: false,
in_controlled_shutdown: false,
cordoned_log_dirs: None,
features: std::collections::BTreeMap::new(),
});
let mut kip858_image = img();
kip858_image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
name: crate::metadata_version::METADATA_VERSION_FEATURE.into(),
level: crate::metadata_version::DIRECTORY_ASSIGNMENT_MIN_LEVEL,
}));
let value = to_kraft_values(&record, &kip858_image).unwrap().remove(0);
let (wire, version) = KraftMetadataRecord::decode_value(&value).unwrap();
assert2::assert!(version == 3);
assert2::assert!(from_kraft(&wire, &kip858_image).unwrap() == record);
let mut pre_kip858_image = img();
pre_kip858_image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
name: crate::metadata_version::METADATA_VERSION_FEATURE.into(),
level: crate::metadata_version::DIRECTORY_ASSIGNMENT_MIN_LEVEL - 1,
}));
let value = to_kraft_values(&record, &pre_kip858_image)
.unwrap()
.remove(0);
let (wire, version) = KraftMetadataRecord::decode_value(&value).unwrap();
assert2::assert!(version == 2);
let MetadataRecord::V1BrokerRegistration(decoded) =
from_kraft(&wire, &pre_kip858_image).unwrap()
else {
panic!("expected broker registration")
};
assert2::assert!(decoded.log_dirs.is_empty());
}
#[test]
fn partition_epoch_survives_round_trip() {
let mut image = img();
let tid = uuid::Uuid::from_u128(99);
image.apply(&MetadataRecord::V1Topic(TopicRecord {
name: "epoch-test".into(),
topic_id: tid,
partitions: 1,
replication_factor: 1,
}));
let rec = MetadataRecord::V1Partition(PartitionRecord {
topic: "epoch-test".into(),
partition: 0,
leader: NodeId(1),
replicas: vec![NodeId(1)],
isr: vec![NodeId(1)],
leader_epoch: LeaderEpoch(7),
adding_replicas: vec![],
removing_replicas: vec![],
directories: vec![],
partition_epoch: 42,
});
image.apply(&rec);
round_trip(&rec, &image);
}
/// `RegisterBroker` frames at the version Kafka's
/// `MetadataVersion.registerBrokerRecordVersion` picks for each level: v4
/// from `4.3-IV0` (KIP-1066 cordoned log dirs), v3 from `3.7-IV2`, v2 from
/// `3.4-IV0`, v1 below.
#[test]
fn register_broker_frames_at_kafkas_record_version() {
use crate::metadata_version::{
CORDONED_LOG_DIRS_MIN_LEVEL, DIRECTORY_ASSIGNMENT_MIN_LEVEL, METADATA_VERSION_MAX,
METADATA_VERSION_MIN,
};
// Every field a registration carries, so each rung shows which of them
// survive: log dirs from v3, cordoned log dirs from v4.
let directory = uuid::Uuid::from_u128(0xD1);
let full = BrokerRegistrationRecord {
node_id: NodeId(3),
broker_epoch: 5,
incarnation_id: uuid::Uuid::from_u128(3),
host: "127.0.0.1".into(),
port: 9092,
rack: None,
endpoints: vec![],
log_dirs: vec![directory],
fenced: true,
in_controlled_shutdown: true,
cordoned_log_dirs: Some(vec![directory]),
features: std::collections::BTreeMap::new(),
};
let without_cordoned = BrokerRegistrationRecord {
cordoned_log_dirs: None,
..full.clone()
};
let without_dirs = BrokerRegistrationRecord {
log_dirs: vec![],
..without_cordoned.clone()
};
for (level, want_version, want) in [
(METADATA_VERSION_MAX, 4_i16, &full),
(CORDONED_LOG_DIRS_MIN_LEVEL, 4, &full),
(CORDONED_LOG_DIRS_MIN_LEVEL - 1, 3, &without_cordoned),
(DIRECTORY_ASSIGNMENT_MIN_LEVEL, 3, &without_cordoned),
(DIRECTORY_ASSIGNMENT_MIN_LEVEL - 1, 2, &without_dirs),
(8, 2, &without_dirs),
(METADATA_VERSION_MIN, 1, &without_dirs),
] {
let mut image = img();
image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
name: crate::metadata_version::METADATA_VERSION_FEATURE.into(),
level,
}));
let record = MetadataRecord::V1BrokerRegistration(full.clone());
let value = to_kraft_values(&record, &image).unwrap().remove(0);
let (wire, version) = KraftMetadataRecord::decode_value(&value).unwrap();
check!(
(version, from_kraft(&wire, &image).unwrap())
== (
want_version,
MetadataRecord::V1BrokerRegistration(want.clone())
),
"metadata.version {level}"
);
}
}
/// A `BrokerRegistrationChangeRecord` decodes to a
/// `V1BrokerRegistrationChange`, and Kafka's undefined `Fenced` and
/// `InControlledShutdown` values are refused, as
/// `ClusterControlManager.replay` refuses them.
#[test]
fn broker_registration_change_decodes_to_the_change_record() {
let (d1, d2) = (uuid::Uuid::from_u128(0xD1), uuid::Uuid::from_u128(0xD2));
let wire = |fenced: i8,
in_controlled_shutdown: i8,
log_dirs: Vec<uuid::Uuid>,
cordoned: Option<Vec<uuid::Uuid>>| {
BrokerRegistrationChangeRecord {
broker_id: 3,
broker_epoch: 5,
fenced,
in_controlled_shutdown,
log_dirs: log_dirs.into_iter().map(to_kuuid).collect(),
cordoned_log_dirs: cordoned.map(|dirs| dirs.into_iter().map(to_kuuid).collect()),
..Default::default()
}
};
let decoded = |edit: fn(&mut ChangeRecord)| {
let mut change = ChangeRecord::no_change(NodeId(3), 5);
edit(&mut change);
Ok(MetadataRecord::V1BrokerRegistrationChange(change))
};
let invalid = |field, detail: &str| {
Err(TranslateError::InvalidValue {
field,
detail: detail.into(),
})
};
for (case, record, want) in [
("no change", wire(0, 0, vec![], None), decoded(|_| {})),
(
"unfence",
wire(-1, 0, vec![], None),
decoded(|c| c.fenced = FencingChange::Unfence),
),
(
"fence",
wire(1, 0, vec![], None),
decoded(|c| c.fenced = FencingChange::Fence),
),
(
"controlled shutdown",
wire(0, 1, vec![], None),
decoded(|c| c.in_controlled_shutdown = true),
),
(
"new log dirs",
wire(0, 0, vec![d1, d2], None),
decoded(|c| {
c.log_dirs = vec![uuid::Uuid::from_u128(0xD1), uuid::Uuid::from_u128(0xD2)];
}),
),
(
"uncordon every dir",
wire(0, 0, vec![], Some(vec![])),
decoded(|c| c.cordoned_log_dirs = Some(vec![])),
),
(
"unknown fenced value",
wire(2, 0, vec![], None),
invalid("broker registration change fenced", "unknown value 2"),
),
(
"unknown controlled shutdown value",
wire(0, -1, vec![], None),
invalid(
"broker registration change in_controlled_shutdown",
"unknown value -1",
),
),
(
"negative broker id",
BrokerRegistrationChangeRecord {
broker_id: -1,
..wire(0, 0, vec![], None)
},
invalid("broker registration change id", "-1 is negative"),
),
] {
// Through the v3 value bytes, the version that carries every field.
let value = KraftMetadataRecord::BrokerRegistrationChange(record)
.encode_value(3)
.unwrap();
check!(from_kraft_value(&value, &img()) == want, "case {case}");
}
}
/// A registration change is written as Kafka's `BrokerRegistrationChangeRecord`
/// (apiKey 17), at the version `ReplicationControlManager` picks for the
/// fields it sets, and reads back unchanged.
#[test]
fn broker_registration_change_encodes_at_kafkas_record_version() {
let dir = uuid::Uuid::from_u128(0xD1);
let change = |edit: fn(&mut ChangeRecord)| {
let mut change = ChangeRecord::no_change(NodeId(3), 5);
edit(&mut change);
change
};
// Envelope: frame version 1, apiKey 17 (0x11), then the apiVersion.
// Body: BrokerId int32 3, BrokerEpoch int64 5, then the tagged fields.
let frame = |version: u8, tagged: &[u8]| {
let mut bytes = vec![0x01, 0x11, version, 0, 0, 0, 3, 0, 0, 0, 0, 0, 0, 0, 5];
bytes.extend_from_slice(tagged);
bytes
};
// One tagged field: tag, size 17 (a compact array length of one, then
// one 16-byte uuid), the length and the uuid.
let dir_bytes = dir.into_bytes();
let mut log_dirs_tagged = vec![0x01, 0x02, 0x11, 0x02];
log_dirs_tagged.extend_from_slice(&dir_bytes);
let mut cordoned_tagged = vec![0x01, 0x03, 0x11, 0x02];
cordoned_tagged.extend_from_slice(&dir_bytes);
for (case, record, wire) in [
(
"fence at v0",
change(|c| c.fenced = FencingChange::Fence),
frame(0, &[0x01, 0x00, 0x01, 0x01]),
),
(
"unfence at v0",
change(|c| c.fenced = FencingChange::Unfence),
frame(0, &[0x01, 0x00, 0x01, 0xff]),
),
(
"controlled shutdown at v1",
change(|c| c.in_controlled_shutdown = true),
frame(1, &[0x01, 0x01, 0x01, 0x01]),
),
(
"log dirs at v2",
change(|c| c.log_dirs = vec![uuid::Uuid::from_u128(0xD1)]),
frame(2, &log_dirs_tagged),
),
(
"cordoned log dirs at v3",
change(|c| c.cordoned_log_dirs = Some(vec![uuid::Uuid::from_u128(0xD1)])),
frame(3, &cordoned_tagged),
),
] {
let record = MetadataRecord::V1BrokerRegistrationChange(record);
let values = to_kraft_values(&record, &img()).unwrap();
check!(
values.iter().map(|v| v.to_vec()).collect::<Vec<_>>() == vec![wire],
"{case}"
);
check!(from_kraft_value(&values[0], &img()) == Ok(record), "{case}");
}
}
/// Below the level that introduced them, a registration change drops its
/// log dirs (`3.7-IV2`) and cordoned log dirs (`4.3-IV0`), as a
/// `RegisterBrokerRecord` does, and frames at the version left.
#[test]
fn broker_registration_change_drops_fields_the_metadata_version_lacks() {
use crate::metadata_version::{
CORDONED_LOG_DIRS_MIN_LEVEL, DIRECTORY_ASSIGNMENT_MIN_LEVEL,
};
let dir = uuid::Uuid::from_u128(0xD1);
let full = ChangeRecord {
fenced: FencingChange::Unfence,
log_dirs: vec![dir],
cordoned_log_dirs: Some(vec![dir]),
..ChangeRecord::no_change(NodeId(3), 5)
};
for (level, want_version, want) in [
(CORDONED_LOG_DIRS_MIN_LEVEL, 3, full.clone()),
(
CORDONED_LOG_DIRS_MIN_LEVEL - 1,
2,
ChangeRecord {
cordoned_log_dirs: None,
..full.clone()
},
),
(
DIRECTORY_ASSIGNMENT_MIN_LEVEL - 1,
0,
ChangeRecord {
log_dirs: vec![],
cordoned_log_dirs: None,
..full.clone()
},
),
] {
let mut image = img();
image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
name: crate::metadata_version::METADATA_VERSION_FEATURE.into(),
level,
}));
let values = to_kraft_values(
&MetadataRecord::V1BrokerRegistrationChange(full.clone()),
&image,
)
.unwrap();
let (_, version) = KraftMetadataRecord::decode_value(&values[0]).unwrap();
check!(
(version, from_kraft_value(&values[0], &image))
== (
want_version,
Ok(MetadataRecord::V1BrokerRegistrationChange(want))
),
"metadata.version {level}"
);
}
}
/// KIP-1312: controller unregistration is Kafka trunk's
/// `UnregisterControllerRecord` (apiKey 29, v0) in both directions.
#[test]
fn unregister_controller_round_trips_as_api_key_29() {
let record = MetadataRecord::V1UnregisterController(UnregisterControllerRecord {
node_id: NodeId(3000),
});
let values = to_kraft_values(&record, &img()).unwrap();
check!(
values.iter().map(|v| v.to_vec()).collect::<Vec<_>>()
== vec![vec![0x01, 0x1d, 0x00, 0x00, 0x00, 0x0b, 0xb8, 0x00]]
);
check!(from_kraft_value(&values[0], &img()) == Ok(record));
let negative = KraftMetadataRecord::UnregisterController(KUnregisterControllerRecord {
controller_id: -1,
..Default::default()
});
check!(
from_kraft(&negative, &img())
== Err(TranslateError::InvalidValue {
field: "unregister controller id",
detail: "-1 is negative".into(),
})
);
}
/// Partition records frame at v2 once ELR is available, v1 once directory
/// assignment is, and v0 below that. Nothing pinned that ladder: with the
/// first guard forced true every image emits v2, and with the second forced
/// either way the middle rung merges into one of its neighbours -- each of
/// which writes a partition record a Kafka at the target version is
/// required not to understand.
#[test]
fn partition_records_frame_at_the_version_the_image_allows() {
use crate::metadata_version::{DIRECTORY_ASSIGNMENT_MIN_LEVEL, ELR_MIN_LEVEL};
let rec = MetadataRecord::V1Partition(PartitionRecord {
topic: "orders".into(),
partition: 0,
leader: NodeId(1),
replicas: vec![NodeId(1)],
isr: vec![NodeId(1)],
leader_epoch: LeaderEpoch(7),
adding_replicas: vec![],
removing_replicas: vec![],
directories: vec![],
partition_epoch: 42,
});
for (level, want) in [
(ELR_MIN_LEVEL, 2_i16),
(ELR_MIN_LEVEL - 1, 1),
(DIRECTORY_ASSIGNMENT_MIN_LEVEL, 1),
(DIRECTORY_ASSIGNMENT_MIN_LEVEL - 1, 0),
] {
let mut image = img();
image.apply(&MetadataRecord::V1Topic(TopicRecord {
name: "orders".into(),
topic_id: uuid::Uuid::from_u128(1),
partitions: 1,
replication_factor: 1,
}));
image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
name: crate::metadata_version::METADATA_VERSION_FEATURE.into(),
level,
}));
let value = to_kraft_values(&rec, &image).unwrap().remove(0);
let (_, version) = KraftMetadataRecord::decode_value(&value).unwrap();
check!(version == want, "metadata.version {level}");
}
}
/// Broker and controller registrations carry a supported-version map and,
/// for controllers, endpoints and the ZK-migration flag. Every round trip
/// so far left those empty or defaulted, so dropping a feature's name or
/// either version bound, an endpoint's security protocol, or the migration
/// flag changed nothing any test looked at -- and each is what a JVM
/// controller reads to decide what the peer can speak.
#[test]
fn registration_features_and_endpoints_survive_translation() {
let image = img();
let broker = MetadataRecord::V1BrokerRegistration(BrokerRegistrationRecord {
node_id: NodeId(1),
broker_epoch: 5,
incarnation_id: uuid::Uuid::from_u128(3),
host: "127.0.0.1".into(),
port: 9092,
rack: None,
endpoints: vec![],
log_dirs: vec![],
fenced: false,
in_controlled_shutdown: false,
cordoned_log_dirs: None,
features: std::collections::BTreeMap::from([
("metadata.version".to_string(), (7_i16, 25_i16)),
("kraft.version".to_string(), (0, 1)),
]),
});
round_trip(&broker, &image);
let controller = MetadataRecord::V1ControllerRegistration(ControllerRegistrationRecord {
node_id: NodeId(3),
incarnation_id: uuid::Uuid::from_u128(9),
zk_migration_ready: true,
endpoints: vec![crate::records::BrokerEndpoint {
name: "CONTROLLER".into(),
host: "controller-3".into(),
port: 9093,
protocol: krabka_security::ListenerProtocol::SaslSsl,
}],
features: std::collections::BTreeMap::from([(
"kraft.version".to_string(),
(0_i16, 1_i16),
)]),
});
round_trip(&controller, &image);
}
/// `UnregisterBrokerRecord` carries the epoch of the registration it
/// removes, as Kafka's `ReplicationControlManager.handleBrokerUnregistered`
/// writes it.
#[test]
fn unregister_broker_carries_the_broker_epoch() {
let rec = MetadataRecord::V1UnregisterBroker(UnregisterBrokerRecord {
node_id: NodeId(42),
broker_epoch: 0x0102,
});
let values = to_kraft_values(&rec, &img()).unwrap();
// Frame version 1, apiKey 1, v0; BrokerId int32 42, BrokerEpoch int64
// 0x0102, then an empty tagged-field section.
check!(
values.iter().map(|v| v.to_vec()).collect::<Vec<_>>()
== vec![vec![
0x01, 0x01, 0x00, 0, 0, 0, 42, 0, 0, 0, 0, 0, 0, 0x01, 0x02, 0x00
]]
);
check!(from_kraft_value(&values[0], &img()) == Ok(rec));
}
#[test]
fn negative_wire_broker_id_is_rejected() {
let record = KraftMetadataRecord::UnregisterBroker(KUnregisterBrokerRecord {
broker_id: -1,
..Default::default()
});
let error = from_kraft(&record, &img()).unwrap_err();
assert!(matches!(error, TranslateError::InvalidValue { .. }));
}
#[test]
fn wire_partition_no_leader_sentinel_round_trips() {
let mut image = img();
let topic_id = uuid::Uuid::from_u128(100);
image.apply(&MetadataRecord::V1Topic(TopicRecord {
name: "leaderless-test".into(),
topic_id,
partitions: 1,
replication_factor: 1,
}));
let record = KraftMetadataRecord::Partition(KPartitionRecord {
topic_id: to_kuuid(topic_id),
partition_id: 0,
leader: -1,
..Default::default()
});
let decoded = from_kraft(&record, &image).unwrap();
let MetadataRecord::V1Partition(decoded) = decoded else {
panic!("expected partition")
};
assert2::assert!(decoded.leader == NodeId(0));
let encoded = to_kraft(&MetadataRecord::V1Partition(decoded), &image).unwrap();
let KraftMetadataRecord::Partition(encoded) = encoded else {
panic!("expected partition")
};
assert2::assert!(encoded.leader == -1);
}
#[test]
fn wire_partition_change_no_leader_updates_epochs() {
let mut image = img();
let topic_id = uuid::Uuid::from_u128(101);
image.apply(&MetadataRecord::V1Topic(TopicRecord {
name: "leaderless-change-test".into(),
topic_id,
partitions: 1,
replication_factor: 1,
}));
image.apply(&MetadataRecord::V1Partition(PartitionRecord {
topic: "leaderless-change-test".into(),
partition: 0,
leader: NodeId(1),
replicas: vec![NodeId(1)],
isr: vec![NodeId(1)],
leader_epoch: LeaderEpoch(7),
partition_epoch: 11,
adding_replicas: vec![],
removing_replicas: vec![],
directories: vec![],
}));
let change = KraftMetadataRecord::PartitionChange(KPartitionChangeRecord {
topic_id: to_kuuid(topic_id),
partition_id: 0,
leader: -1,
..Default::default()
});
let decoded = from_kraft(&change, &image).unwrap();
let MetadataRecord::V1Partition(decoded) = decoded else {
panic!("expected partition")
};
assert2::assert!(decoded.leader == NodeId(0));
assert2::assert!(decoded.leader_epoch == LeaderEpoch(8));
assert2::assert!(decoded.partition_epoch == 12);
}
#[test]
fn negative_wire_partition_replica_is_rejected() {
let mut image = img();
let topic_id = uuid::Uuid::from_u128(102);
image.apply(&MetadataRecord::V1Topic(TopicRecord {
name: "negative-node-test".into(),
topic_id,
partitions: 1,
replication_factor: 1,
}));
let record = KraftMetadataRecord::Partition(KPartitionRecord {
topic_id: to_kuuid(topic_id),
partition_id: 0,
leader: -1,
replicas: vec![-1],
..Default::default()
});
let error = from_kraft(&record, &image).unwrap_err();
assert!(matches!(error, TranslateError::InvalidValue { .. }));
}
#[test]
fn scram_credential_round_trips() {
let rec = MetadataRecord::V1ScramCredential(ScramCredentialRecord {
user: "alice".into(),
mechanism: SaslMechanism::ScramSha512,
salt: vec![1u8; 16],
stored_key: vec![2u8; 64],
server_key: vec![3u8; 64],
iterations: 4096,
});
round_trip(&rec, &img());
}
#[test]
fn delete_scram_credential_round_trips() {
let rec = MetadataRecord::V1DeleteScramCredential(DeleteScramCredentialRecord {
user: "alice".into(),
mechanism: SaslMechanism::ScramSha256,
});
round_trip(&rec, &img());
}
/// KIP-966 eligible-leader-replica state is refused, and either field
/// alone is enough to refuse it. Read as `&&`, a change carrying only one
/// Kafka may set either ELR field independently in a partition delta.
#[test]
fn either_elr_field_alone_is_preserved() {
let topic_id = uuid::Uuid::from_u128(1);
let mut image = img();
image.apply(&MetadataRecord::V1Topic(TopicRecord {
name: "orders".into(),
topic_id,
partitions: 1,
replication_factor: 1,
}));
image.apply(&MetadataRecord::V1Partition(PartitionRecord {
topic: "orders".into(),
partition: 0,
leader: NodeId(1),
replicas: vec![NodeId(1)],
isr: vec![NodeId(1)],
leader_epoch: LeaderEpoch(1),
adding_replicas: vec![],
removing_replicas: vec![],
directories: vec![],
partition_epoch: 1,
}));
let change = |elr: Option<Vec<i32>>, last_known: Option<Vec<i32>>| {
KraftMetadataRecord::PartitionChange(KPartitionChangeRecord {
topic_id: to_kuuid(topic_id),
partition_id: 0,
leader_recovery_state: -1,
eligible_leader_replicas: elr,
last_known_elr: last_known,
..Default::default()
})
};
check!(from_kraft(&change(None, None), &image).is_ok());
check!(matches!(
from_kraft(&change(Some(vec![1]), None), &image),
Ok(MetadataRecord::V1PartitionUpdate(crate::PartitionUpdateRecord {
eligible_leader_replicas: Some(eligible_leader_replicas),
last_known_elr: None,
..
})) if eligible_leader_replicas == vec![NodeId(1)]
));
check!(matches!(
from_kraft(&change(None, Some(vec![1])), &image),
Ok(MetadataRecord::V1PartitionUpdate(crate::PartitionUpdateRecord {
eligible_leader_replicas: None,
last_known_elr: Some(last_known_elr),
..
})) if last_known_elr == vec![NodeId(1)]
));
let combined = KraftMetadataRecord::PartitionChange(KPartitionChangeRecord {
topic_id: to_kuuid(topic_id),
partition_id: 0,
leader: 2,
isr: Some(vec![2]),
leader_recovery_state: LeaderRecoveryState::Recovering as i8,
eligible_leader_replicas: Some(vec![2]),
last_known_elr: Some(vec![1]),
..Default::default()
});
let decoded = from_kraft(&combined, &image).expect("combined partition delta");
let MetadataRecord::V1PartitionUpdate(update) = decoded else {
panic!("combined delta must stay atomic");
};
check!(update.partition.leader == NodeId(2));
check!(update.partition.isr == vec![NodeId(2)]);
check!(update.recovery_state == Some(LeaderRecoveryState::Recovering));
check!(update.eligible_leader_replicas == Some(vec![NodeId(2)]));
check!(update.last_known_elr == Some(vec![NodeId(1)]));
}
/// The ACL id is a content hash: it is what a later `RemoveAccessControlEntry`
/// matches against to find the entry it deletes. Dropped, every entry
/// translates with the same defaulted id, so distinct ACLs become
/// indistinguishable -- which the round-trip above does not notice, because
/// it reconstructs the entry from the other fields.
#[test]
fn distinct_acls_translate_to_distinct_ids() {
let entry = |operation, resource_name: &str| crate::AclEntry {
resource_type: ResourceType::Topic,
resource_name: resource_name.into(),
pattern_type: PatternType::Literal,
principal: "User:alice".into(),
host: "*".into(),
operation,
permission_type: PermissionType::Allow,
};
let id_of = |e: &crate::AclEntry| {
let KraftMetadataRecord::AccessControlEntry(r) = acl_to_kraft(e) else {
panic!("expected an access control entry");
};
r.id
};
let read_foo = id_of(&entry(AclOperation::Read, "foo"));
check!(read_foo == id_of(&entry(AclOperation::Read, "foo")));
check!(read_foo != id_of(&entry(AclOperation::Write, "foo")));
check!(read_foo != id_of(&entry(AclOperation::Read, "bar")));
}
#[test]
fn access_control_entry_round_trips() {
let rec = MetadataRecord::V1AccessControlEntry(AclEntry {
resource_type: ResourceType::Topic,
resource_name: "foo".into(),
pattern_type: PatternType::Literal,
principal: "User:alice".into(),
host: "*".into(),
operation: AclOperation::Read,
permission_type: PermissionType::Allow,
});
round_trip(&rec, &img());
}
#[test]
fn access_control_entry_two_phase_commit_round_trips() {
// KIP-939: TWO_PHASE_COMMIT (wire byte 15) must survive the
// record↔wire round-trip on a TransactionalId resource.
let rec = MetadataRecord::V1AccessControlEntry(AclEntry {
resource_type: ResourceType::TransactionalId,
resource_name: "my-txn".into(),
pattern_type: PatternType::Literal,
principal: "User:flink".into(),
host: "*".into(),
operation: AclOperation::TwoPhaseCommit,
permission_type: PermissionType::Allow,
});
round_trip(&rec, &img());
}
/// Every ACL resource type and operation maps to Kafka's `i8`
/// discriminant and back through a full `AccessControlEntryRecord`
/// round trip, KIP-373's `USER` (7), `CREATE_TOKENS` (13) and
/// `DESCRIBE_TOKENS` (14) included.
#[test]
fn acl_enums_map_to_kafka_wire_codes() {
for (resource_type, wire) in [
(ResourceType::Topic, 2_i8),
(ResourceType::Group, 3),
(ResourceType::Cluster, 4),
(ResourceType::TransactionalId, 5),
(ResourceType::DelegationToken, 6),
(ResourceType::User, 7),
] {
check!(
(
resource_type_to_wire(resource_type),
resource_type_from_wire(wire).unwrap(),
) == (wire, resource_type)
);
}
for (operation, wire) in [
(AclOperation::All, 2_i8),
(AclOperation::Read, 3),
(AclOperation::Write, 4),
(AclOperation::Create, 5),
(AclOperation::Delete, 6),
(AclOperation::Alter, 7),
(AclOperation::Describe, 8),
(AclOperation::ClusterAction, 9),
(AclOperation::DescribeConfigs, 10),
(AclOperation::AlterConfigs, 11),
(AclOperation::IdempotentWrite, 12),
(AclOperation::CreateTokens, 13),
(AclOperation::DescribeTokens, 14),
(AclOperation::TwoPhaseCommit, 15),
] {
check!(
(
operation_to_wire(operation),
operation_from_wire(wire).unwrap()
) == (wire, operation)
);
}
check!(resource_type_from_wire(8).is_err());
check!(operation_from_wire(16).is_err());
}
/// KIP-373: `CREATE_TOKENS` and `DESCRIBE_TOKENS` granted on a `USER`
/// resource survive the record-to-wire round trip.
#[test]
fn access_control_entry_user_token_operations_round_trip() {
for operation in [AclOperation::CreateTokens, AclOperation::DescribeTokens] {
let rec = MetadataRecord::V1AccessControlEntry(AclEntry {
resource_type: ResourceType::User,
resource_name: "alice".into(),
pattern_type: PatternType::Literal,
principal: "User:token-requester".into(),
host: "*".into(),
operation,
permission_type: PermissionType::Allow,
});
let value = to_kraft_values(&rec, &img()).unwrap().remove(0);
check!(from_kraft_value(&value, &img()).unwrap() == rec);
let KraftMetadataRecord::AccessControlEntry(wire) = to_kraft(&rec, &img()).unwrap()
else {
panic!("expected an access control entry");
};
check!((wire.resource_type, wire.operation) == (7, operation_to_wire(operation)));
}
}
#[test]
fn access_control_entry_prefixed_deny_round_trips() {
let rec = MetadataRecord::V1AccessControlEntry(AclEntry {
resource_type: ResourceType::DelegationToken,
resource_name: "User:bob".into(),
pattern_type: PatternType::Prefixed,
principal: "User:eve".into(),
host: "10.0.0.1".into(),
operation: AclOperation::Describe,
permission_type: PermissionType::Deny,
});
round_trip(&rec, &img());
}
fn acl(rt: ResourceType, name: &str, principal: &str) -> AclEntry {
AclEntry {
resource_type: rt,
resource_name: name.into(),
pattern_type: PatternType::Literal,
principal: principal.into(),
host: "*".into(),
operation: AclOperation::Read,
permission_type: PermissionType::Allow,
}
}
#[test]
fn distinct_acls_get_distinct_ids() {
let a = acl(ResourceType::Topic, "foo", "User:alice");
let b = acl(ResourceType::Topic, "foo", "User:bob");
let c = acl(ResourceType::Group, "foo", "User:alice");
check!(acl_id(&a) != acl_id(&b));
check!(acl_id(&a) != acl_id(&c));
// Deterministic: same entry hashes identically.
check!(acl_id(&a) == acl_id(&a.clone()));
}
#[test]
fn delete_acl_filter_resolves_to_remove_records_and_round_trips() {
// Image holds three ACLs; a Topic-only filter deletes the two Topic
// entries and leaves the Group entry.
let mut image = img();
let e1 = acl(ResourceType::Topic, "foo", "User:alice");
let e2 = acl(ResourceType::Topic, "bar", "User:bob");
let e3 = acl(ResourceType::Group, "cg", "User:alice");
for e in [&e1, &e2, &e3] {
image.apply(&MetadataRecord::V1AccessControlEntry(e.clone()));
}
let del = MetadataRecord::V1DeleteAccessControlEntry(AclEntryFilter {
resource_type: Some(ResourceType::Topic),
..AclEntryFilter::default()
});
let records = to_kraft_records(&del, &image).unwrap();
assert2::assert!(records.len() == 2); // e1, e2 — not e3
// Each RemoveAccessControlEntry decodes back to a pinned filter that
// removes exactly its entry. Applying both clears e1+e2, keeps e3.
for k in &records {
let back = from_kraft(k, &image).unwrap();
image.apply(&back);
}
assert2::assert!(image.all_acls().count() == 1);
let survivor = image.all_acls().next().unwrap();
assert2::assert!(*survivor == e3);
}
#[test]
fn delete_acl_filter_matching_nothing_yields_no_records() {
let mut image = img();
image.apply(&MetadataRecord::V1AccessControlEntry(acl(
ResourceType::Topic,
"foo",
"User:alice",
)));
let del = MetadataRecord::V1DeleteAccessControlEntry(AclEntryFilter {
resource_type: Some(ResourceType::Group),
..AclEntryFilter::default()
});
assert2::assert!(to_kraft_records(&del, &image).unwrap().is_empty());
}
#[test]
fn client_quota_round_trips() {
let rec = MetadataRecord::V1ClientQuota(ClientQuotaRecord {
entity: vec![
QuotaEntity {
entity_type: "client-id".into(),
entity_name: Some("app1".into()),
},
QuotaEntity {
entity_type: "user".into(),
entity_name: Some("alice".into()),
},
],
config_key: "producer_byte_rate".into(),
config_value: Some(1024.0),
});
round_trip(&rec, &img());
}
#[test]
fn client_quota_remove_round_trips() {
let rec = MetadataRecord::V1ClientQuota(ClientQuotaRecord {
entity: vec![QuotaEntity {
entity_type: "user".into(),
entity_name: None,
}],
config_key: "consumer_byte_rate".into(),
config_value: None,
});
round_trip(&rec, &img());
}
#[test]
fn delegation_token_round_trips() {
let rec = MetadataRecord::V1DelegationToken(DelegationTokenRecord {
token_id: "tok-abc".into(),
owner: KafkaPrincipal {
principal_type: "User".into(),
name: "alice".into(),
},
requester: KafkaPrincipal {
principal_type: "User".into(),
name: "admin".into(),
},
issue_timestamp_ms: 1_700_000_000_000,
expiry_timestamp_ms: 1_700_000_600_000,
max_timestamp_ms: 1_700_604_800_000,
renewers: vec![KafkaPrincipal {
principal_type: "User".into(),
name: "bob".into(),
}],
});
round_trip(&rec, &img());
}
#[test]
fn delete_delegation_token_round_trips() {
let rec = MetadataRecord::V1DeleteDelegationToken(DeleteDelegationTokenRecord {
token_id: "tok-abc".into(),
});
round_trip(&rec, &img());
}
#[test]
fn broker_config_round_trips() {
let rec = MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
node_id: NodeId(7),
config_name: "leader.replication.throttled.rate".into(),
config_value: Some("2048".into()),
});
let encoded = to_kraft_records(&rec, &img()).expect("encode broker config");
assert2::assert!(matches!(
encoded.as_slice(),
[KraftMetadataRecord::Config(config)]
if config.resource_type == 4 && config.resource_name == "7"
));
round_trip(&rec, &img());
}
#[test]
fn default_broker_config_round_trips_with_empty_resource_name() {
let rec = MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
node_id: DEFAULT_BROKER_CONFIG_NODE_ID,
config_name: "follower.replication.throttled.rate".into(),
config_value: Some("4096".into()),
});
let encoded = to_kraft_records(&rec, &img()).expect("encode default broker config");
assert2::assert!(matches!(
encoded.as_slice(),
[KraftMetadataRecord::Config(config)]
if config.resource_type == 4 && config.resource_name.is_empty()
));
round_trip(&rec, &img());
}
#[test]
fn broker_config_delete_round_trips() {
let rec = MetadataRecord::V1BrokerConfig(BrokerConfigRecord {
node_id: NodeId(7),
config_name: "leader.replication.throttled.rate".into(),
config_value: None,
});
round_trip(&rec, &img());
}
#[test]
fn client_metrics_config_uses_standard_kraft_config_records() {
let rec = MetadataRecord::V1ClientMetricsConfig(ClientMetricsConfigRecord {
name: "metrics-sub-1".into(),
configs: [
("interval.ms".to_string(), "60000".to_string()),
("metrics".to_string(), "org.apache.kafka".to_string()),
]
.into_iter()
.collect(),
});
let mut image = img();
let records = to_kraft_records(&rec, &image).unwrap();
assert2::assert!(records.len() == 2);
assert2::assert!(records.iter().all(|record| matches!(
record,
KraftMetadataRecord::Config(config) if config.resource_type == 16
)));
for record in records {
let decoded = from_kraft(&record, &image).unwrap();
image.apply(&decoded);
}
assert2::assert!(
image.client_metrics_config("metrics-sub-1")
== match &rec {
MetadataRecord::V1ClientMetricsConfig(config) => Some(&config.configs),
_ => unreachable!(),
}
);
}
#[test]
fn client_metrics_config_clear_emits_standard_tombstones() {
let mut image = img();
image.apply(&MetadataRecord::V1ClientMetricsConfig(
ClientMetricsConfigRecord {
name: "metrics-sub-1".into(),
configs: std::collections::BTreeMap::from([
("interval.ms".into(), "60000".into()),
("metrics".into(), "org.apache.kafka".into()),
]),
},
));
let clear = MetadataRecord::V1ClientMetricsConfig(ClientMetricsConfigRecord {
name: "metrics-sub-1".into(),
configs: std::collections::BTreeMap::new(),
});
let records = to_kraft_records(&clear, &image).unwrap();
assert2::assert!(records.len() == 2);
assert2::assert!(records.iter().all(|record| matches!(
record,
KraftMetadataRecord::Config(config)
if config.resource_type == 16 && config.value.is_none()
)));
for record in records {
let decoded = from_kraft(&record, &image).unwrap();
image.apply(&decoded);
}
assert2::assert!(image.client_metrics_config("metrics-sub-1").is_none());
}
#[test]
fn features_epoch_round_trips_via_private_carrier() {
// The synthetic `V1FeaturesEpoch` epoch pin is Krabka-internal (no KRaft
// record); it round-trips through the same private `NoOpRecord` carrier
// so a snapshot preserves `finalized_features_epoch`.
let rec = MetadataRecord::V1FeaturesEpoch(FeaturesEpochRecord { epoch: 42 });
let k = to_kraft(&rec, &img()).unwrap();
assert2::assert!(matches!(
k,
KraftMetadataRecord::NoOp(no_op)
if no_op.unknown_tagged_fields.0.len() == 1
&& no_op.unknown_tagged_fields.0[0].tag == PRIVATE_FEATURES_EPOCH_TAG
));
round_trip(&rec, &img());
}
#[test]
fn partition_dir_assignment_uses_standard_partition_change_record() {
let topic_id = uuid::Uuid::from_u128(0x00C0_FFEE);
let mut image = img();
image.apply(&MetadataRecord::V1Topic(TopicRecord {
name: "t".into(),
topic_id,
partitions: 1,
replication_factor: 3,
}));
image.apply(&MetadataRecord::V1Partition(PartitionRecord {
topic: "t".into(),
partition: 0,
leader: NodeId(1),
replicas: vec![NodeId(1), NodeId(2), NodeId(3)],
isr: vec![NodeId(1), NodeId(2)],
leader_epoch: LeaderEpoch(4),
adding_replicas: vec![NodeId(3)],
removing_replicas: vec![],
directories: vec![uuid::Uuid::nil(); 3],
partition_epoch: 9,
}));
let rec = MetadataRecord::V1PartitionDirAssignment(PartitionDirAssignmentRecord {
topic: "t".into(),
partition: 0,
replica: NodeId(2),
directory: uuid::Uuid::from_u128(0xAB),
});
let k = to_kraft(&rec, &image).unwrap();
let KraftMetadataRecord::PartitionChange(change) = &k else {
panic!("expected standard PartitionChangeRecord");
};
assert2::assert!(k.api_key() == 5);
assert2::assert!(change.isr.is_none());
assert2::assert!(change.replicas.is_none());
assert2::assert!(
change.directories.as_ref().unwrap()[1] == to_kuuid(uuid::Uuid::from_u128(0xAB))
);
let values = to_kraft_values(&rec, &image).unwrap();
let (decoded_wire, version) = KraftMetadataRecord::decode_value(&values[0]).unwrap();
assert2::assert!(version == 2);
let decoded = from_kraft(&decoded_wire, &image).unwrap();
image.apply(&decoded);
let partition = image.partition("t", 0).unwrap();
assert2::assert!(partition.directories[1] == uuid::Uuid::from_u128(0xAB));
assert2::assert!(partition.isr == vec![NodeId(1), NodeId(2)]);
assert2::assert!(partition.adding_replicas == vec![NodeId(3)]);
assert2::assert!(partition.leader_epoch == LeaderEpoch(4));
assert2::assert!(partition.partition_epoch == 10);
}
#[test]
fn voters_and_kraft_version_have_no_counterpart() {
let v = MetadataRecord::V1Voters(VotersRecord {
voters: crate::voters::VoterSet::default(),
});
let k = MetadataRecord::V1KRaftVersion(KRaftVersionRecord { kraft_version: 1 });
for (_name, record, expected) in [
("voter record", v, TranslateError::NoCounterpart("V1Voters")),
(
"KRaft version record",
k,
TranslateError::NoCounterpart("V1KRaftVersion"),
),
] {
assert2::assert!(to_kraft(&record, &img()) == Err(expected));
}
}
// ---------- topic / partition / config with image context ----------
#[test]
fn topic_partition_config_round_trip_with_image_context() {
let mut image = img();
let tid = uuid::Uuid::from_u128(7);
let topic = MetadataRecord::V1Topic(TopicRecord {
name: "t".into(),
topic_id: tid,
partitions: 1,
replication_factor: 1,
});
image.apply(&topic);
let part = MetadataRecord::V1Partition(PartitionRecord {
topic: "t".into(),
partition: 0,
leader: NodeId(1),
replicas: vec![NodeId(1)],
isr: vec![NodeId(1)],
leader_epoch: LeaderEpoch(0),
adding_replicas: vec![],
removing_replicas: vec![],
directories: vec![],
partition_epoch: 3,
});
image.apply(&part);
for rec in [&topic, &part] {
let k = to_kraft(rec, &image).unwrap();
assert2::assert!(from_kraft(&k, &image).unwrap() == *rec);
}
let cfg = MetadataRecord::V1TopicConfig(TopicConfigRecord {
topic: "t".into(),
overrides: [("retention.ms".to_string(), "9".to_string())].into(),
});
let k = to_kraft(&cfg, &image).unwrap();
assert2::assert!(from_kraft(&k, &image).unwrap() == cfg);
}
#[test]
fn delete_topic_round_trips_with_image_context() {
let mut image = img();
let tid = uuid::Uuid::from_u128(11);
image.apply(&MetadataRecord::V1Topic(TopicRecord {
name: "doomed".into(),
topic_id: tid,
partitions: 1,
replication_factor: 1,
}));
let rec = MetadataRecord::V1DeleteTopic(DeleteTopicRecord {
name: "doomed".into(),
});
round_trip(&rec, &image);
}
#[test]
fn topic_config_two_keys_yields_two_config_records() {
let mut image = img();
image.apply(&MetadataRecord::V1Topic(TopicRecord {
name: "t".into(),
topic_id: uuid::Uuid::from_u128(7),
partitions: 1,
replication_factor: 1,
}));
let cfg = MetadataRecord::V1TopicConfig(TopicConfigRecord {
topic: "t".into(),
overrides: [
("retention.ms".to_string(), "9".to_string()),
("segment.bytes".to_string(), "1048576".to_string()),
]
.into(),
});
let records = to_kraft_records(&cfg, &image).unwrap();
assert2::assert!(records.len() == 2);
// Each decodes back to a single-key mergeable singleton.
for k in &records {
let back = from_kraft(k, &image).unwrap();
match back {
MetadataRecord::V1TopicConfig(tc) => {
assert2::assert!(tc.topic.as_str() == "t");
assert2::assert!(tc.overrides.len() == 1);
}
other => panic!("expected V1TopicConfig singleton, got {other:?}"),
}
}
}
#[test]
fn topic_config_to_kraft_preserves_set_values() {
let mut image = img();
image.apply(&MetadataRecord::V1Topic(TopicRecord {
name: "t".into(),
topic_id: uuid::Uuid::from_u128(7),
partitions: 1,
replication_factor: 1,
}));
let cfg = MetadataRecord::V1TopicConfig(TopicConfigRecord {
topic: "t".into(),
overrides: [("retention.ms".to_string(), "9".to_string())].into(),
});
let records = to_kraft_records(&cfg, &image).unwrap();
assert2::assert!(records.len() == 1);
let KraftMetadataRecord::Config(c) = &records[0] else {
panic!("expected Config");
};
assert2::assert!(c.name.as_str() == "retention.ms");
assert2::assert!(c.value.as_deref() == Some("9"));
}
/// Seed `image` with topic `t` and the config map `kv`.
fn seed_topic_config(image: &mut MetadataImage, kv: &[(&str, &str)]) {
image.apply(&MetadataRecord::V1Topic(TopicRecord {
name: "t".into(),
topic_id: uuid::Uuid::from_u128(7),
partitions: 1,
replication_factor: 1,
}));
image.apply(&MetadataRecord::V1TopicConfig(TopicConfigRecord {
topic: "t".into(),
overrides: kv
.iter()
.map(|(k, v)| ((*k).to_string(), (*v).to_string()))
.collect(),
}));
}
#[test]
fn topic_config_change_emits_set_and_tombstone_and_merges_back() {
// Current config {a:1, b:2}; new config {a:9} (a changed, b dropped).
let mut image = img();
seed_topic_config(&mut image, &[("a", "1"), ("b", "2")]);
let submit = MetadataRecord::V1TopicConfig(TopicConfigRecord {
topic: "t".into(),
overrides: [("a".to_string(), "9".to_string())].into(),
});
let records = to_kraft_records(&submit, &image).unwrap();
// One set (a=9) + one tombstone (b removed).
assert2::assert!(records.len() == 2);
// Apply the fanned records in order against the live image; the merge
// reproduces the new whole map exactly.
for k in &records {
let back = from_kraft(k, &image).unwrap();
image.apply(&back);
}
let want: std::collections::BTreeMap<String, String> =
[("a".to_string(), "9".to_string())].into();
assert2::assert!(image.topic_config("t") == Some(&want));
}
#[test]
fn topic_config_clear_emits_tombstones_for_every_key() {
// Current {a:1, b:2}; an empty new map clears all → two tombstones.
let mut image = img();
seed_topic_config(&mut image, &[("a", "1"), ("b", "2")]);
let submit = MetadataRecord::V1TopicConfig(TopicConfigRecord {
topic: "t".into(),
overrides: std::collections::BTreeMap::new(),
});
let records = to_kraft_records(&submit, &image).unwrap();
assert2::assert!(records.len() == 2);
for k in &records {
let back = from_kraft(k, &image).unwrap();
image.apply(&back);
}
// All overrides gone (apply of an empty whole-map removes the entry).
assert2::assert!(image.topic_config("t").is_none());
}
#[test]
fn group_config_change_uses_kraft_group_records_and_merges_back() {
let mut image = img();
image.apply(&MetadataRecord::V1GroupConfig(GroupConfigRecord {
group_id: "streams-app".into(),
configs: std::collections::BTreeMap::from([
("streams.num.standby.replicas".into(), "0".into()),
("streams.num.warmup.replicas".into(), "2".into()),
]),
}));
let submitted = MetadataRecord::V1GroupConfig(GroupConfigRecord {
group_id: "streams-app".into(),
configs: std::collections::BTreeMap::from([(
"streams.num.standby.replicas".into(),
"1".into(),
)]),
});
let records = to_kraft_records(&submitted, &image).expect("translate group config");
assert2::assert!(records.len() == 2);
assert2::assert!(records.iter().all(|record| matches!(
record,
KraftMetadataRecord::Config(config) if config.resource_type == 32
)));
for record in records {
let decoded = from_kraft(&record, &image).expect("decode group config");
image.apply(&decoded);
}
assert2::assert!(
image.group_config("streams-app")
== match &submitted {
MetadataRecord::V1GroupConfig(config) => Some(&config.configs),
_ => unreachable!(),
}
);
}
#[test]
fn producer_ids_use_standard_kraft_record() {
let image = img();
let submitted = MetadataRecord::V1ProducerIds(ProducerIdsRecord {
broker_id: NodeId(3),
broker_epoch: 11,
next_producer_id: 5_000,
});
let values = to_kraft_values(&submitted, &image).expect("encode producer IDs record");
assert2::assert!(values.len() == 1);
let (wire, version) = KraftMetadataRecord::decode_value(&values[0]).unwrap();
assert2::assert!(wire.api_key() == 15);
assert2::assert!(version == 0);
assert2::assert!(from_kraft(&wire, &image).unwrap() == submitted);
}
#[test]
fn to_kraft_values_round_trips_through_value_bytes() {
// The crate framing entrypoints: a record encodes to KIP-631 value
// bytes and decodes back identically (single-output record).
let image = img();
let rec = MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
name: "metadata.version".into(),
level: 25,
});
let values = to_kraft_values(&rec, &image).unwrap();
assert2::assert!(values.len() == 1);
let back = from_kraft_value(&values[0], &image).unwrap();
assert2::assert!(back == rec);
}
#[test]
fn partition_record_directories_survive_kraft_round_trip() {
use crate::{MetadataImage, MetadataRecord, PartitionRecord, TopicRecord};
let topic_id = uuid::Uuid::from_u128(0x11);
let mut image = MetadataImage::new(uuid::Uuid::nil());
image.apply(&MetadataRecord::V1Topic(TopicRecord {
name: "t".into(),
topic_id,
partitions: 1,
replication_factor: 2,
}));
let rec = MetadataRecord::V1Partition(PartitionRecord {
topic: "t".into(),
partition: 0,
leader: NodeId(1),
replicas: vec![NodeId(1), NodeId(2)],
isr: vec![NodeId(1), NodeId(2)],
leader_epoch: LeaderEpoch(4),
adding_replicas: vec![],
removing_replicas: vec![],
directories: vec![uuid::Uuid::from_u128(0xAA), uuid::Uuid::from_u128(0xBB)],
partition_epoch: 0,
});
let values = super::to_kraft_values(&rec, &image).expect("encode");
assert2::assert!(values.len() == 1);
let decoded = super::from_kraft_value(&values[0], &image).expect("decode");
assert2::assert!(decoded == rec);
}
#[test]
fn offset_advance_round_trips_through_carrier() {
let rec = MetadataRecord::V1PartitionOffsetAdvance(crate::PartitionOffsetAdvanceRecord {
topic: "t".into(),
partition: 2,
count: 9,
});
let values = to_kraft_values(&rec, &img()).unwrap();
assert_eq!(values.len(), 1);
let back = from_kraft_value(&values[0], &img()).unwrap();
assert_eq!(back, rec);
}
// ---------- KFC-9 private carriers ----------
fn freeze_records() -> [(&'static str, MetadataRecord, u32); 3] {
let id = uuid::Uuid::from_u128(0xB1);
[
(
"topic freeze",
MetadataRecord::V1TopicFreeze(crate::write_freeze::TopicFreezeRecord {
scope: "tenant-a.".into(),
pattern_type: PatternType::Prefixed,
frozen: true,
reason: "DR cutover".into(),
set_by: "User:alice".into(),
set_at_ms: 1_700_000_000_000,
proposal_id: id,
key_id: "alice-yubi".into(),
signature: vec![0xAB; 64],
}),
PRIVATE_TOPIC_FREEZE_TAG,
),
(
"break-glass proposal",
MetadataRecord::V1BreakGlassProposal(
crate::break_glass::BreakGlassProposalRecord {
proposal_id: id,
action: crate::break_glass::BreakGlassAction::ThawTopicFreeze,
target: "literal:orders".into(),
proposer: "User:alice".into(),
reason: "incident 4711 closed".into(),
created_at_ms: 1_700_000_000_000,
expires_at_ms: 1_700_000_600_000,
approvals: vec![crate::break_glass::BreakGlassApproval {
principal: "User:bob".into(),
approved_at_ms: 1_700_000_060_000,
key_id: "bob-yubi".into(),
signature: vec![0xCD; 64],
}],
consumed_at_ms: 1_700_000_120_000,
withdrawn: false,
},
),
PRIVATE_BREAK_GLASS_PROPOSAL_TAG,
),
(
"break-glass tombstone",
MetadataRecord::V1DeleteBreakGlassProposal(id),
PRIVATE_DELETE_BREAK_GLASS_PROPOSAL_TAG,
),
]
}
#[test]
fn freeze_and_break_glass_records_round_trip_through_carriers() {
for (label, rec, _) in freeze_records() {
let values = to_kraft_values(&rec, &img()).unwrap();
check!(values.len() == 1, "{label}");
check!(
from_kraft_value(&values[0], &img()).unwrap() == rec,
"{label}"
);
}
}
/// The tags are a durable on-disk contract, so pin them. 1002 is an
/// unexplained gap in the numbering and stays free.
#[test]
fn freeze_and_break_glass_carriers_use_the_pinned_private_tags() {
for (label, rec, want) in freeze_records() {
let KraftMetadataRecord::NoOp(no_op) = to_kraft(&rec, &img()).unwrap() else {
panic!("{label}: expected a private NoOpRecord carrier");
};
let tags: Vec<u32> = no_op
.unknown_tagged_fields
.0
.iter()
.map(|f| f.tag)
.collect();
check!(tags == vec![want], "{label}");
}
check!(
PRIVATE_TAGS
== [
PRIVATE_FEATURES_EPOCH_TAG,
PRIVATE_PARTITION_OFFSET_ADVANCE_TAG,
PRIVATE_TOPIC_FREEZE_TAG,
PRIVATE_BREAK_GLASS_PROPOSAL_TAG,
PRIVATE_DELETE_BREAK_GLASS_PROPOSAL_TAG,
]
);
check!(PRIVATE_TAGS == [1001, 1003, 1004, 1005, 1006]);
}
/// The apiKeys of Kafka's `MetadataRecordType` table
/// (`metadata/src/main/resources/common/metadata/*Record.json`, trunk).
/// `MetadataRecordSerde.apiMessageFor` throws for any other id, so
/// `kafka-dump-log` and `kafka-metadata-shell` fail on it.
const KAFKA_METADATA_RECORD_API_KEYS: [u32; 27] = [
0, 1, 2, 3, 4, 5, 7, 8, 9, 10, 11, 12, 14, 15, 17, 18, 19, 20, 21, 22, 23, 24, 25, 26, 27,
28, 29,
];
/// One record per assigned private tag, in tag order.
fn private_rows() -> Vec<(&'static str, MetadataRecord, u32)> {
let mut rows: Vec<(&'static str, MetadataRecord, u32)> = vec![
(
"features epoch",
MetadataRecord::V1FeaturesEpoch(FeaturesEpochRecord { epoch: 7 }),
PRIVATE_FEATURES_EPOCH_TAG,
),
(
"diskless offset advance",
MetadataRecord::V1PartitionOffsetAdvance(crate::PartitionOffsetAdvanceRecord {
topic: "t".into(),
partition: 2,
count: 9,
}),
PRIVATE_PARTITION_OFFSET_ADVANCE_TAG,
),
];
rows.extend(freeze_records());
rows
}
/// The version-0 body of a private tag: the big-endian `i16` version,
/// then the wincode record.
fn versioned_body(version: i16, rec: &MetadataRecord) -> Vec<u8> {
let mut body = version.to_be_bytes().to_vec();
body.extend(<serde_wincode::SerdeCompat<MetadataRecord>>::serialize(rec).unwrap());
body
}
/// Every Krabka-only record frames as a Kafka `NoOpRecord` v0 whose body is
/// exactly one tagged field and no trailing bytes. Kafka's reader accepts
/// that shape: the undeclared tag goes to `_unknownTaggedFields`, and
/// `AbstractApiMessageSerde.read` finds no garbage after the record.
#[test]
fn private_records_frame_as_a_kafka_no_op_record() {
for (label, rec, tag) in private_rows() {
let values = to_kraft_values(&rec, &img()).unwrap();
check!(values.len() == 1, "{label}");
let (decoded, version) = KraftMetadataRecord::decode_value(&values[0]).unwrap();
check!(
KAFKA_METADATA_RECORD_API_KEYS.contains(&decoded.api_key()),
"{label}: apiKey {} is not in Kafka's table",
decoded.api_key()
);
check!(
(decoded, version)
== (
KraftMetadataRecord::NoOp(NoOpRecord {
unknown_tagged_fields: UnknownTaggedFields(vec![UnknownTaggedField {
tag,
bytes: Bytes::from(versioned_body(PRIVATE_RECORD_VERSION, &rec)),
}]),
}),
0
),
"{label}"
);
check!(
from_kraft_value(&values[0], &img()).unwrap() == rec,
"{label}"
);
}
}
/// The full `__cluster_metadata` record value of each row of
/// [`private_rows`], in the same order. These bytes are the krabka 1.x
/// on-disk contract: the KIP-631 frame, the `NoOpRecord` tagged field, the
/// `i16` record version 0, then the wincode record.
const PRIVATE_GOLDEN_VALUES: [&str; 5] = [
// features epoch (tag 1001)
"01140001e9070e0000130000000700000000000000",
// diskless offset advance (tag 1003)
"01140001eb071b000015000000010000000000000074020000000900000000000000",
// topic freeze (tag 1004)
"01140001ec07ba01000018000000090000000000000074656e616e742d612e01000000010a000000000000004452206375746f7665720a00000000000000557365723a616c6963650068e5cf8b0100001000000000000000000000000000000000000000000000b10a00000000000000616c6963652d797562694000000000000000abababababababababababababababababababababababababababababababababababababababababababababababababababababababababababababababab",
// break-glass proposal (tag 1005)
"01140001ed07f7010000190000001000000000000000000000000000000000000000000000b1000000000e000000000000006c69746572616c3a6f72646572730a00000000000000557365723a616c6963651400000000000000696e636964656e74203437313120636c6f7365640068e5cf8b010000c08feecf8b01000001000000000000000800000000000000557365723a626f626052e6cf8b0100000800000000000000626f622d797562694000000000000000cdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdcdc03ce7cf8b01000000",
// break-glass tombstone (tag 1006)
"01140001ee071e00001a0000001000000000000000000000000000000000000000000000b1",
];
#[test]
fn private_records_match_the_golden_bytes() {
use crate::wincode_contract::{hex, unhex};
for ((label, rec, _), golden) in private_rows().into_iter().zip(PRIVATE_GOLDEN_VALUES) {
let values = to_kraft_values(&rec, &img()).unwrap();
check!(
values.iter().map(|v| hex(v)).collect::<Vec<_>>() == vec![golden.to_owned()],
"{label}"
);
check!(
from_kraft_value(&unhex(golden), &img()) == Ok(rec),
"{label}"
);
}
}
#[test]
fn private_carrier_rejects_a_bad_tag_or_body() {
let epoch = MetadataRecord::V1FeaturesEpoch(FeaturesEpochRecord { epoch: 7 });
let topic = MetadataRecord::V1Topic(TopicRecord {
name: "t".into(),
topic_id: uuid::Uuid::from_u128(0x7),
partitions: 1,
replication_factor: 1,
});
let unversioned = <serde_wincode::SerdeCompat<MetadataRecord>>::serialize(&epoch).unwrap();
for (label, tag, body, want) in [
(
"version 1",
1001,
versioned_body(1, &epoch),
TranslateError::UnknownPrivateRecordVersion {
tag: 1001,
version: 1,
},
),
(
"future version",
1001,
versioned_body(i16::MAX, &epoch),
TranslateError::UnknownPrivateRecordVersion {
tag: 1001,
version: i16::MAX,
},
),
(
"0.x body without a version",
1001,
unversioned,
TranslateError::UnknownPrivateRecordVersion {
tag: 1001,
version: 0x1300,
},
),
(
"record of another tag",
1004,
versioned_body(0, &epoch),
TranslateError::PrivateTagMismatch {
tag: 1004,
expected_tag: Some(1001),
},
),
(
"record of no tag",
1001,
versioned_body(0, &topic),
TranslateError::PrivateTagMismatch {
tag: 1001,
expected_tag: None,
},
),
(
"trailing byte after the record",
1001,
{
let mut body = versioned_body(0, &epoch);
body.push(0);
body
},
TranslateError::TrailingPrivateRecordBytes {
tag: 1001,
trailing: 1,
},
),
(
"burned tag 1002",
1002,
versioned_body(0, &epoch),
TranslateError::UnknownPrivateTag(1002),
),
(
"unassigned tag 1007",
1007,
versioned_body(0, &epoch),
TranslateError::UnknownPrivateTag(1007),
),
(
"one-byte body",
1001,
vec![0],
TranslateError::Decode(
"krabka-private NoOpRecord tag 1001 body is 1 bytes, too short for its i16 \
version"
.into(),
),
),
] {
let no_op = KraftMetadataRecord::NoOp(NoOpRecord {
unknown_tagged_fields: UnknownTaggedFields(vec![UnknownTaggedField {
tag,
bytes: Bytes::from(body),
}]),
});
check!(from_kraft(&no_op, &img()) == Err(want), "{label}");
}
}
/// A `NoOpRecord` that does not carry exactly one private tag is Kafka's
/// own no-op. It has no Krabka counterpart.
#[test]
fn no_op_without_one_private_tag_has_no_counterpart() {
let field = |tag: u32| UnknownTaggedField {
tag,
bytes: Bytes::from_static(b"x"),
};
for (label, fields) in [
("empty", vec![]),
("foreign tag", vec![field(7)]),
(
"two private tags",
vec![
field(PRIVATE_TOPIC_FREEZE_TAG),
field(PRIVATE_BREAK_GLASS_PROPOSAL_TAG),
],
),
] {
let no_op = KraftMetadataRecord::NoOp(NoOpRecord {
unknown_tagged_fields: UnknownTaggedFields(fields),
});
check!(
from_kraft(&no_op, &img()) == Err(TranslateError::NoCounterpart("NoOp")),
"{label}"
);
}
}
#[test]
fn partition_to_kraft_preserves_reassignment_vectors() {
let topic_id = uuid::Uuid::from_u128(0x11);
let mut image = MetadataImage::new(uuid::Uuid::nil());
image.apply(&MetadataRecord::V1Topic(TopicRecord {
name: "t".into(),
topic_id,
partitions: 1,
replication_factor: 3,
}));
let rec = MetadataRecord::V1Partition(PartitionRecord {
topic: "t".into(),
partition: 0,
leader: NodeId(1),
replicas: vec![NodeId(1), NodeId(2), NodeId(3)],
isr: vec![NodeId(1), NodeId(2)],
leader_epoch: LeaderEpoch(4),
adding_replicas: vec![NodeId(3)],
removing_replicas: vec![NodeId(2)],
directories: vec![],
partition_epoch: 9,
});
let KraftMetadataRecord::Partition(k) = to_kraft(&rec, &image).unwrap() else {
panic!("expected Partition");
};
assert2::assert!(k.adding_replicas == vec![3]);
assert2::assert!(k.removing_replicas == vec![2]);
}
#[test]
fn non_private_unknown_has_no_counterpart() {
let err = from_kraft(
&KraftMetadataRecord::Unknown {
api_key: 999_999,
api_version: 0,
body: bytes::Bytes::from_static(b"not a private carrier"),
},
&img(),
)
.unwrap_err();
assert2::assert!(err == TranslateError::NoCounterpart("Unknown"));
}
#[test]
fn remove_acl_with_unknown_id_is_rejected() {
let remove_unknown =
KraftMetadataRecord::RemoveAccessControlEntry(RemoveAccessControlEntryRecord {
id: to_kuuid(uuid::Uuid::from_u128(0xA11CE)),
..Default::default()
});
let err = from_kraft(&remove_unknown, &img()).unwrap_err();
assert2::assert!(matches!(err, TranslateError::UnknownAclId(_)));
let mut image_with_different_acl = img();
image_with_different_acl.apply(&MetadataRecord::V1AccessControlEntry(acl(
ResourceType::Topic,
"topic-a",
"User:alice",
)));
let err = from_kraft(&remove_unknown, &image_with_different_acl).unwrap_err();
assert2::assert!(matches!(err, TranslateError::UnknownAclId(_)));
}
#[test]
fn kraft_variant_name_reports_unmodeled_variants() {
for (record, want) in [
(
KraftMetadataRecord::BrokerRegistrationChange(
krabka_protocol::owned::broker_registration_change_record::BrokerRegistrationChangeRecord::default(),
),
"BrokerRegistrationChange",
),
(
KraftMetadataRecord::NoOp(
krabka_protocol::owned::no_op_record::NoOpRecord::default(),
),
"NoOp",
),
(
KraftMetadataRecord::BeginTransaction(
krabka_protocol::owned::begin_transaction_record::BeginTransactionRecord::default(),
),
"BeginTransaction",
),
(
KraftMetadataRecord::EndTransaction(
krabka_protocol::owned::end_transaction_record::EndTransactionRecord::default(),
),
"EndTransaction",
),
(
KraftMetadataRecord::RemoveAccessControlEntry(
RemoveAccessControlEntryRecord::default(),
),
"RemoveAccessControlEntry",
),
(
KraftMetadataRecord::PartitionChange(KPartitionChangeRecord::default()),
"PartitionChange",
),
(
KraftMetadataRecord::Unknown {
api_key: 42,
api_version: 0,
body: bytes::Bytes::new(),
},
"Unknown",
),
] {
assert2::assert!(kraft_variant_name(&record) == want);
}
}
#[test]
fn image_derives_partition_count_from_partitions_map() {
let mut image = img();
image.apply(&MetadataRecord::V1Topic(TopicRecord {
name: "t".into(),
topic_id: uuid::Uuid::from_u128(7),
partitions: 3,
replication_factor: 1,
}));
for p in 0..3 {
image.apply(&MetadataRecord::V1Partition(PartitionRecord {
topic: "t".into(),
partition: p,
leader: NodeId(1),
replicas: vec![NodeId(1)],
isr: vec![NodeId(1)],
leader_epoch: LeaderEpoch(0),
adding_replicas: vec![],
removing_replicas: vec![],
directories: vec![],
partition_epoch: 0,
}));
}
assert2::assert!(image.topic_partition_count("t") == 3);
}
/// An image with topic `t` and three one-replica partitions: 0 is
/// ordinary, 1 holds its leader epoch at the maximum and 2 its partition
/// epoch.
fn classification_image() -> (MetadataImage, uuid::Uuid) {
let topic_id = uuid::Uuid::from_u128(0x7E);
let mut image = img();
image.apply(&MetadataRecord::V1Topic(TopicRecord {
name: "t".into(),
topic_id,
partitions: 3,
replication_factor: 1,
}));
for (partition, leader_epoch, partition_epoch) in [
(0, LeaderEpoch(7), 3),
(1, LeaderEpoch(i32::MAX), 3),
(2, LeaderEpoch(7), i32::MAX),
] {
image.apply(&MetadataRecord::V1Partition(PartitionRecord {
topic: "t".into(),
partition,
leader: NodeId(1),
replicas: vec![NodeId(1)],
isr: vec![NodeId(1)],
leader_epoch,
partition_epoch,
adding_replicas: vec![],
removing_replicas: vec![],
directories: vec![],
}));
}
(image, topic_id)
}
/// Pins which `Invalid*` variant a representative of each kind of bad
/// input produces. A reference error depends on what the image holds; a
/// value error is decided by the record's bytes alone.
#[test]
fn invalid_errors_split_into_reference_and_value() {
let (image, topic_id) = classification_image();
let change = |partition_id: i32, edit: fn(&mut KPartitionChangeRecord)| {
let mut record = KPartitionChangeRecord {
topic_id: to_kuuid(topic_id),
partition_id,
..Default::default()
};
edit(&mut record);
from_kraft(&KraftMetadataRecord::PartitionChange(record), &image)
};
let reference = |field, detail: &str| {
Err(TranslateError::InvalidReference {
field,
detail: detail.into(),
})
};
let value = |field, detail: &str| {
Err(TranslateError::InvalidValue {
field,
detail: detail.into(),
})
};
for (case, got, want) in [
(
"decode: unknown partition",
change(5, |_| {}),
reference("partition change", "unknown partition t-5"),
),
(
"decode: directories do not match the image's replicas",
change(0, |c| {
c.directories = Some(vec![KUuid([1; 16]), KUuid([2; 16])]);
}),
reference(
"partition change directories",
"2 directories for 1 replicas",
),
),
(
"decode: directories do not match replicas in the same record",
change(0, |c| {
c.replicas = Some(vec![1, 2]);
c.directories = Some(vec![KUuid([1; 16])]);
}),
value(
"partition change directories",
"1 directories for 2 replicas",
),
),
(
"decode: image leader epoch at its maximum",
change(1, |c| c.leader = 1),
reference("partition change leader epoch", "leader epoch overflow"),
),
(
"encode: directory assignment for an unknown partition",
to_kraft(
&MetadataRecord::V1PartitionDirAssignment(PartitionDirAssignmentRecord {
topic: "t".into(),
partition: 9,
replica: NodeId(1),
directory: uuid::Uuid::from_u128(0xD1),
}),
&image,
)
.map(|_| unreachable!()),
reference("partition directory assignment", "unknown partition t-9"),
),
(
"decode: image partition epoch at its maximum",
change(2, |_| {}),
reference(
"partition change partition epoch",
"partition epoch overflow",
),
),
(
"decode: unknown leader recovery state on a partition record",
from_kraft(
&KraftMetadataRecord::Partition(KPartitionRecord {
topic_id: to_kuuid(topic_id),
partition_id: 0,
leader: 1,
replicas: vec![1],
isr: vec![1],
leader_recovery_state: 9,
..Default::default()
}),
&image,
),
value("partition leader recovery state", "unknown state 9"),
),
(
"decode: unknown leader recovery state on a known partition",
change(0, |c| c.leader_recovery_state = 9),
value("partition change leader recovery state", "unknown state 9"),
),
(
"decode: unknown fenced value",
from_kraft(
&KraftMetadataRecord::BrokerRegistrationChange(
BrokerRegistrationChangeRecord {
broker_id: 1,
fenced: 7,
..Default::default()
},
),
&image,
),
value("broker registration change fenced", "unknown value 7"),
),
(
"decode: negative node id",
from_kraft(
&KraftMetadataRecord::UnregisterBroker(KUnregisterBrokerRecord {
broker_id: -1,
..Default::default()
}),
&image,
),
value("unregister broker id", "-1 is negative"),
),
(
"encode: node id that exceeds i32",
to_kraft(
&MetadataRecord::V1ProducerIds(ProducerIdsRecord {
broker_id: NodeId(1 << 31),
broker_epoch: 0,
next_producer_id: 0,
}),
&image,
)
.map(|_| unreachable!()),
value("producer IDs broker id", "node_id 2147483648 exceeds i32"),
),
] {
check!(got == want, "{case}");
}
}
/// Every byte-content check names its field and the offending value, and
/// reports it as `InvalidValue`, whatever the image holds.
#[test]
fn byte_content_errors_are_invalid_values() {
let image = img();
let decode = |record: KraftMetadataRecord| from_kraft(&record, &image).map(drop);
let encode = |record: MetadataRecord| to_kraft(&record, &image).map(drop);
let scram = |mechanism: i8, iterations: i32| {
decode(KraftMetadataRecord::UserScramCredential(
UserScramCredentialRecord {
name: "alice".into(),
mechanism,
iterations,
..Default::default()
},
))
};
let token = |owner: &str, requester: &str, renewer: &str| {
decode(KraftMetadataRecord::DelegationToken(
KDelegationTokenRecord {
owner: owner.into(),
requester: requester.into(),
renewers: vec![renewer.into()],
..Default::default()
},
))
};
let config = |resource_type: i8, resource_name: &str| {
decode(KraftMetadataRecord::Config(ConfigRecord {
resource_type,
resource_name: resource_name.into(),
name: "k".into(),
..Default::default()
}))
};
let value = |field, detail: &str| {
Err(TranslateError::InvalidValue {
field,
detail: detail.into(),
})
};
for (case, got, want) in [
(
"a non-SCRAM mechanism on encode",
scram_mechanism_to_wire(SaslMechanism::Plain).map(drop),
value("scram mechanism", "Plain is not a SCRAM mechanism"),
),
(
"an unknown SCRAM mechanism byte",
scram(9, 4096),
value("scram mechanism", "unknown SCRAM mechanism wire byte 9"),
),
(
"negative SCRAM iterations",
scram(1, -1),
value("scram iterations", "-1 is negative"),
),
(
"SCRAM iterations that exceed i32 on encode",
encode(MetadataRecord::V1ScramCredential(ScramCredentialRecord {
user: "alice".into(),
mechanism: SaslMechanism::ScramSha256,
salt: vec![],
stored_key: vec![],
server_key: vec![],
iterations: u32::MAX,
})),
value("scram iterations", "4294967295 exceeds i32"),
),
(
"an unknown ACL resource type",
resource_type_from_wire(9).map(drop),
value("acl resource_type", "unknown wire byte 9"),
),
(
"an unknown ACL pattern type",
pattern_type_from_wire(9).map(drop),
value("acl pattern_type", "unknown wire byte 9"),
),
(
"an unknown ACL operation",
operation_from_wire(99).map(drop),
value("acl operation", "unknown wire byte 99"),
),
(
"an unknown ACL permission type",
permission_from_wire(9).map(drop),
value("acl permission_type", "unknown wire byte 9"),
),
(
"an unknown security protocol",
protocol_from_wire(9).map(drop),
value("security_protocol", "unknown wire value 9"),
),
(
"an unregistered broker id that exceeds i32 on encode",
encode(MetadataRecord::V1UnregisterBroker(UnregisterBrokerRecord {
node_id: NodeId(1 << 31),
broker_epoch: 0,
})),
value("broker_id", "node_id 2147483648 exceeds i32"),
),
(
"a registered broker id that exceeds i32 on encode",
encode(MetadataRecord::V1BrokerRegistration(
BrokerRegistrationRecord {
node_id: NodeId(1 << 31),
broker_epoch: 0,
incarnation_id: uuid::Uuid::nil(),
host: "localhost".into(),
port: 9092,
rack: None,
log_dirs: vec![],
fenced: false,
in_controlled_shutdown: false,
cordoned_log_dirs: None,
endpoints: vec![],
features: std::collections::BTreeMap::new(),
},
)),
value("broker_id", "node_id 2147483648 exceeds i32"),
),
(
"a registered controller id that exceeds i32 on encode",
encode(MetadataRecord::V1ControllerRegistration(
ControllerRegistrationRecord {
node_id: NodeId(1 << 31),
incarnation_id: uuid::Uuid::nil(),
zk_migration_ready: false,
endpoints: vec![],
features: std::collections::BTreeMap::new(),
},
)),
value("controller_id", "node_id 2147483648 exceeds i32"),
),
(
"a delegation token owner that does not parse",
token("alice", "User:bob", "User:carol"),
value("delegation token owner", "invalid principal \"alice\""),
),
(
"a delegation token requester that does not parse",
token("User:alice", "bob", "User:carol"),
value("delegation token requester", "invalid principal \"bob\""),
),
(
"a delegation token renewer that does not parse",
token("User:alice", "User:bob", "carol"),
value("delegation token renewer", "invalid principal \"carol\""),
),
(
"a broker config resource name that is not a node id",
config(4, "broker-one"),
value(
"broker config resource_name",
"invalid digit found in string",
),
),
(
"an unsupported config resource type",
config(99, "x"),
value("config resource_type", "unsupported resource_type 99"),
),
] {
check!(got == want, "{case}");
}
}
/// The encode-side checks of partition records and partition updates:
/// a lookup in the image is a reference error, a node id that exceeds
/// `i32` a value error.
#[test]
fn partition_encode_errors_split_into_reference_and_value() {
let (image, _) = classification_image();
let encode = |record: MetadataRecord| to_kraft(&record, &image).map(drop);
let update = |partition: i32, edit: fn(&mut crate::PartitionUpdateRecord)| {
let mut record = crate::PartitionUpdateRecord {
partition: image.partition("t", 0).unwrap().clone(),
eligible_leader_replicas: None,
last_known_elr: None,
recovery_state: None,
};
record.partition.partition = partition;
edit(&mut record);
encode(MetadataRecord::V1PartitionUpdate(record))
};
let reference = |field, detail: &str| {
Err(TranslateError::InvalidReference {
field,
detail: detail.into(),
})
};
let value = |field, detail: &str| {
Err(TranslateError::InvalidValue {
field,
detail: detail.into(),
})
};
for (case, got, want) in [
(
"encode: directory assignment for a broker that is not a replica",
encode(MetadataRecord::V1PartitionDirAssignment(
PartitionDirAssignmentRecord {
topic: "t".into(),
partition: 0,
replica: NodeId(2),
directory: uuid::Uuid::from_u128(0xD1),
},
)),
reference(
"partition directory assignment",
"broker 2 is not a replica of t-0",
),
),
(
"encode: partition update for an unknown partition",
update(9, |_| {}),
reference("partition update", "unknown partition t-9"),
),
(
"encode: partition update with a replica that exceeds i32",
update(0, |u| u.partition.replicas = vec![NodeId(1 << 31)]),
value("partition replicas", "node id 2147483648 exceeds i32"),
),
(
"encode: partition update with an eligible leader that exceeds i32",
update(0, |u| {
u.eligible_leader_replicas = Some(vec![NodeId(1 << 31)]);
}),
value("eligible leader replica", "node id 2147483648 exceeds i32"),
),
(
"encode: partition with a replica that exceeds i32",
encode(MetadataRecord::V1Partition(PartitionRecord {
replicas: vec![NodeId(1 << 31)],
..image.partition("t", 0).unwrap().clone()
})),
value("partition replicas", "node id 2147483648 exceeds i32"),
),
(
"encode: partition with a leader that exceeds i32",
encode(MetadataRecord::V1Partition(PartitionRecord {
leader: NodeId(1 << 31),
..image.partition("t", 0).unwrap().clone()
})),
value("partition leader", "leader 2147483648 exceeds i32"),
),
] {
check!(got == want, "{case}");
}
}
}