krabka_metadata/

kraft_translate.rs

//! 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 std::{collections::BTreeMap, vec};

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_delegation_token_record::RemoveDelegationTokenRecord,
        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},
    directory_id::MIGRATING,
    metadata_version::{
        CORDONED_LOG_DIRS_MIN_LEVEL, DIRECTORY_ASSIGNMENT_MIN_LEVEL, ELR_MIN_LEVEL,
        METADATA_VERSION_MAX,
    },
    records::{
        BrokerConfigRecord, BrokerEndpoint, BrokerRegistrationChangeRecord as ChangeRecord,
        BrokerRegistrationRecord, ClientMetricsConfigRecord, ClientQuotaRecord,
        ControllerRegistrationRecord, DEFAULT_BROKER_CONFIG_NODE_ID, DelegationTokenRecord,
        DeleteDelegationTokenRecord, 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.
///
/// There is one blob 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(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 < DIRECTORY_ASSIGNMENT_MIN_LEVEL {
                    change.log_dirs.clear();
                }
                if 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 >= ELR_MIN_LEVEL => 2,
                        version if version >= DIRECTORY_ASSIGNMENT_MIN_LEVEL => 1,
                        _ => 0,
                    }
                }
                KraftMetadataRecord::RegisterBroker(_) => {
                    if metadata_version >= CORDONED_LOG_DIRS_MIN_LEVEL {
                        4
                    } else if 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: &TopicConfigRecord,
    image: &MetadataImage,
) -> Vec<KraftMetadataRecord> {
    let empty = 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 = 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: &ClientMetricsConfigRecord,
    image: &MetadataImage,
) -> Vec<KraftMetadataRecord> {
    let empty = 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<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(
                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(), MIGRATING);
    if let Some(directory) = directories.get_mut(replica_slot) {
        *directory = 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_owned()))?
        .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) = q.config_value.map_or((0.0, true), |v| (v, false));
    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(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(
                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 krabka_protocol::owned::{
        begin_transaction_record::BeginTransactionRecord,
        end_transaction_record::EndTransactionRecord,
    };

    use super::*;
    use crate::{
        acl::AclEntryFilter,
        break_glass::{BreakGlassAction, BreakGlassApproval, BreakGlassProposalRecord},
        metadata_version::METADATA_VERSION_FEATURE,
        records::{
            FeaturesEpochRecord, KRaftVersionRecord, PartitionDirAssignmentRecord, VotersRecord,
        },
        voters::VoterSet,
        write_freeze::TopicFreezeRecord,
    };

    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: 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: 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: 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: 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: 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: BTreeMap::new(),
        });

        let mut kip858_image = img();
        kip858_image.apply(&MetadataRecord::V1FeatureLevel(FeatureLevelRecord {
            name: METADATA_VERSION_FEATURE.into(),
            level: 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: METADATA_VERSION_FEATURE.into(),
            level: 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: 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: 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: 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: 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: BTreeMap::from([
                ("metadata.version".to_owned(), (7_i16, 25_i16)),
                ("kraft.version".to_owned(), (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![BrokerEndpoint {
                name: "CONTROLLER".into(),
                host: "controller-3".into(),
                port: 9093,
                protocol: krabka_security::ListenerProtocol::SaslSsl,
            }],
            features: BTreeMap::from([("kraft.version".to_owned(), (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_owned(), "60000".to_owned()),
                ("metrics".to_owned(), "org.apache.kafka".to_owned()),
            ]
            .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: BTreeMap::from([
                    ("interval.ms".into(), "60000".into()),
                    ("metrics".into(), "org.apache.kafka".into()),
                ]),
            },
        ));
        let clear = MetadataRecord::V1ClientMetricsConfig(ClientMetricsConfigRecord {
            name: "metrics-sub-1".into(),
            configs: 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: 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_owned(), "9".to_owned())].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_owned(), "9".to_owned()),
                ("segment.bytes".to_owned(), "1048576".to_owned()),
            ]
            .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_owned(), "9".to_owned())].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_owned(), (*v).to_owned()))
                .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_owned(), "9".to_owned())].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: BTreeMap<String, String> = [("a".to_owned(), "9".to_owned())].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: 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: 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: 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(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(BreakGlassProposalRecord {
                    proposal_id: id,
                    action: 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![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(
                    BrokerRegistrationChangeRecord::default(),
                ),
                "BrokerRegistrationChange",
            ),
            (KraftMetadataRecord::NoOp(NoOpRecord::default()), "NoOp"),
            (
                KraftMetadataRecord::BeginTransaction(BeginTransactionRecord::default()),
                "BeginTransaction",
            ),
            (
                KraftMetadataRecord::EndTransaction(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: 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: 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}");
        }
    }
}