kafkit-client 0.1.9

//! Admin client operations for topics, groups, configs, and cluster metadata.
//!
//! ```no_run
//! # async fn example() -> kafkit_client::Result<()> {
//! use kafkit_client::{KafkaClient, NewTopic};
//!
//! let admin = KafkaClient::new("localhost:9092").admin().connect().await?;
//! admin
//!     .create_topics([NewTopic::new("orders", 3, 1)])
//!     .await?;
//! # Ok(())
//! # }
//! ```
//!
use std::collections::BTreeMap;

use anyhow::anyhow;
use bytes::{Buf, Bytes};
use kafka_protocol::error::{ParseResponseErrorCode, ResponseError};
use kafka_protocol::messages::alter_user_scram_credentials_request::ScramCredentialUpsertion;
use kafka_protocol::messages::create_acls_request::AclCreation;
use kafka_protocol::messages::create_partitions_request::{
    CreatePartitionsAssignment, CreatePartitionsTopic,
};
use kafka_protocol::messages::create_topics_request::{CreatableTopic, CreatableTopicConfig};
use kafka_protocol::messages::delete_acls_request::DeleteAclsFilter;
use kafka_protocol::messages::describe_configs_request::DescribeConfigsResource;
use kafka_protocol::messages::describe_log_dirs_request::DescribableLogDirTopic;
use kafka_protocol::messages::incremental_alter_configs_request::{
    AlterConfigsResource, AlterableConfig,
};
use kafka_protocol::messages::metadata_request::MetadataRequestTopic;
use kafka_protocol::messages::update_features_request::FeatureUpdateKey;
use kafka_protocol::messages::{
    AlterUserScramCredentialsRequest, AlterUserScramCredentialsResponse,
    ConsumerGroupDescribeRequest, ConsumerGroupDescribeResponse, ConsumerProtocolAssignment,
    ConsumerProtocolSubscription, CreateAclsRequest, CreateAclsResponse, CreatePartitionsRequest,
    CreatePartitionsResponse, CreateTopicsRequest, CreateTopicsResponse, DeleteAclsRequest,
    DeleteAclsResponse, DeleteGroupsRequest, DeleteGroupsResponse, DeleteTopicsRequest,
    DeleteTopicsResponse, DescribeAclsRequest, DescribeAclsResponse, DescribeClusterRequest,
    DescribeClusterResponse, DescribeConfigsRequest, DescribeConfigsResponse,
    DescribeGroupsRequest, DescribeGroupsResponse, DescribeLogDirsRequest, DescribeLogDirsResponse,
    IncrementalAlterConfigsRequest, IncrementalAlterConfigsResponse, ListGroupsRequest,
    ListGroupsResponse, MetadataRequest, MetadataResponse, ShareGroupDescribeRequest,
    ShareGroupDescribeResponse, UpdateFeaturesRequest, UpdateFeaturesResponse,
};
use kafka_protocol::protocol::{Decodable, Request, StrBytes};
use tracing::{debug, instrument};
use uuid::Uuid;

use crate::config::{AdminConfig, SaslMechanism};
use crate::constants::{
    ALTER_USER_SCRAM_CREDENTIALS_VERSION_CAP, CONSUMER_GROUP_DESCRIBE_VERSION_CAP,
    CREATE_ACLS_VERSION_CAP, CREATE_PARTITIONS_VERSION_CAP, CREATE_TOPICS_VERSION_CAP,
    DELETE_ACLS_VERSION_CAP, DELETE_GROUPS_VERSION_CAP, DELETE_TOPICS_VERSION_CAP,
    DESCRIBE_ACLS_VERSION_CAP, DESCRIBE_CLUSTER_VERSION_CAP, DESCRIBE_CONFIGS_VERSION_CAP,
    DESCRIBE_GROUPS_VERSION_CAP, DESCRIBE_LOG_DIRS_VERSION_CAP,
    INCREMENTAL_ALTER_CONFIGS_VERSION_CAP, LIST_GROUPS_VERSION_CAP, METADATA_VERSION_CAP,
    SHARE_GROUP_DESCRIBE_VERSION_CAP, UPDATE_FEATURES_VERSION_CAP,
};
use crate::network::scram;
use crate::network::{BrokerConnection, connect_to_any_bootstrap, duration_to_i32_ms};
use crate::types::TopicPartition;
use crate::{AdminError, BrokerError, Result, ValidationError};

#[derive(Debug, Clone)]
/// Topic definition used with [`KafkaAdmin::create_topics`].
///
/// `num_partitions` and `replication_factor` must both be positive. Additional
/// topic-level broker configs can be attached with [`NewTopic::with_config`].
///
/// ```no_run
/// # async fn example(admin: kafkit_client::KafkaAdmin) -> kafkit_client::Result<()> {
/// use kafkit_client::NewTopic;
///
/// admin
///     .create_topics([
///         NewTopic::new("orders", 6, 3)
///             .with_config("cleanup.policy", "delete")
///             .with_config("retention.ms", "604800000"),
///     ])
///     .await?;
/// # Ok(())
/// # }
/// ```
pub struct NewTopic {
    /// Topic name to create.
    pub name: String,
    /// Number of partitions for the new topic.
    pub num_partitions: i32,
    /// Number of replicas Kafka should create for each partition.
    pub replication_factor: i16,
    /// Topic-level configuration entries.
    pub configs: BTreeMap<String, String>,
}

impl NewTopic {
    /// Creates a topic definition.
    pub fn new(name: impl Into<String>, num_partitions: i32, replication_factor: i16) -> Self {
        Self {
            name: name.into(),
            num_partitions,
            replication_factor,
            configs: BTreeMap::new(),
        }
    }

    /// Adds a topic-level config entry and returns the updated definition.
    pub fn with_config(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
        self.configs.insert(key.into(), value.into());
        self
    }

    fn into_request_topic(self) -> Result<CreatableTopic> {
        let name = validate_topic_name(self.name)?;
        if self.num_partitions <= 0 {
            return Err(AdminError::InvalidPartitionCount {
                partitions: self.num_partitions,
            }
            .into());
        }
        if self.replication_factor <= 0 {
            return Err(AdminError::InvalidReplicationFactor {
                replication_factor: self.replication_factor,
            }
            .into());
        }

        let configs = self
            .configs
            .into_iter()
            .map(|(key, value)| {
                CreatableTopicConfig::default()
                    .with_name(StrBytes::from_string(key))
                    .with_value(Some(StrBytes::from_string(value)))
            })
            .collect();

        Ok(CreatableTopic::default()
            .with_name(StrBytes::from_string(name).into())
            .with_num_partitions(self.num_partitions)
            .with_replication_factor(self.replication_factor)
            .with_configs(configs))
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Partition increase request used with [`KafkaAdmin::create_partitions`].
///
/// Kafka can only increase a topic's partition count. `total_count` is the new
/// total number of partitions, not the number of partitions to add.
///
/// ```no_run
/// # async fn example(admin: kafkit_client::KafkaAdmin) -> kafkit_client::Result<()> {
/// use kafkit_client::NewPartitions;
///
/// admin
///     .create_partitions([("orders", NewPartitions::increase_to(12))])
///     .await?;
/// # Ok(())
/// # }
/// ```
pub struct NewPartitions {
    /// New total partition count for the topic.
    pub total_count: i32,
    /// Optional per-new-partition replica broker assignments.
    pub assignments: Vec<Vec<i32>>,
}

impl NewPartitions {
    /// Creates a request to increase a topic to `total_count` partitions.
    pub fn increase_to(total_count: i32) -> Self {
        Self {
            total_count,
            assignments: Vec::new(),
        }
    }

    /// Adds an explicit replica assignment for one new partition.
    ///
    /// Call this once for each new partition that needs an explicit broker
    /// assignment.
    pub fn with_assignment<I>(mut self, broker_ids: I) -> Self
    where
        I: IntoIterator<Item = i32>,
    {
        self.assignments.push(broker_ids.into_iter().collect());
        self
    }

    fn into_request_topic(self, topic_name: String) -> Result<CreatePartitionsTopic> {
        let name = validate_topic_name(topic_name)?;
        if self.total_count <= 0 {
            return Err(AdminError::InvalidPartitionCount {
                partitions: self.total_count,
            }
            .into());
        }

        let assignments = (!self.assignments.is_empty()).then(|| {
            self.assignments
                .into_iter()
                .map(|broker_ids| {
                    CreatePartitionsAssignment::default()
                        .with_broker_ids(broker_ids.into_iter().map(Into::into).collect())
                })
                .collect()
        });

        Ok(CreatePartitionsTopic::default()
            .with_name(StrBytes::from_string(name).into())
            .with_count(self.total_count)
            .with_assignments(assignments))
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
/// Type of Kafka resource an ACL applies to.
pub enum ResourceType {
    Unknown,
    Any,
    Topic,
    Group,
    Cluster,
    TransactionalId,
    DelegationToken,
    User,
}

impl ResourceType {
    fn as_protocol_value(self) -> i8 {
        match self {
            Self::Unknown => 0,
            Self::Any => 1,
            Self::Topic => 2,
            Self::Group => 3,
            Self::Cluster => 4,
            Self::TransactionalId => 5,
            Self::DelegationToken => 6,
            Self::User => 7,
        }
    }

    fn from_protocol_value(value: i8) -> Self {
        match value {
            1 => Self::Any,
            2 => Self::Topic,
            3 => Self::Group,
            4 => Self::Cluster,
            5 => Self::TransactionalId,
            6 => Self::DelegationToken,
            7 => Self::User,
            _ => Self::Unknown,
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
/// Resource pattern type used by ACLs.
pub enum PatternType {
    Unknown,
    Any,
    Match,
    Literal,
    Prefixed,
}

impl PatternType {
    fn as_protocol_value(self) -> i8 {
        match self {
            Self::Unknown => 0,
            Self::Any => 1,
            Self::Match => 2,
            Self::Literal => 3,
            Self::Prefixed => 4,
        }
    }

    fn from_protocol_value(value: i8) -> Self {
        match value {
            1 => Self::Any,
            2 => Self::Match,
            3 => Self::Literal,
            4 => Self::Prefixed,
            _ => Self::Unknown,
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
/// Operation granted or denied by an ACL.
pub enum AclOperation {
    Unknown,
    Any,
    All,
    Read,
    Write,
    Create,
    Delete,
    Alter,
    Describe,
    ClusterAction,
    DescribeConfigs,
    AlterConfigs,
    IdempotentWrite,
    CreateTokens,
    DescribeTokens,
    TwoPhaseCommit,
}

impl AclOperation {
    fn as_protocol_value(self) -> i8 {
        match self {
            Self::Unknown => 0,
            Self::Any => 1,
            Self::All => 2,
            Self::Read => 3,
            Self::Write => 4,
            Self::Create => 5,
            Self::Delete => 6,
            Self::Alter => 7,
            Self::Describe => 8,
            Self::ClusterAction => 9,
            Self::DescribeConfigs => 10,
            Self::AlterConfigs => 11,
            Self::IdempotentWrite => 12,
            Self::CreateTokens => 13,
            Self::DescribeTokens => 14,
            Self::TwoPhaseCommit => 15,
        }
    }

    fn from_protocol_value(value: i8) -> Self {
        match value {
            1 => Self::Any,
            2 => Self::All,
            3 => Self::Read,
            4 => Self::Write,
            5 => Self::Create,
            6 => Self::Delete,
            7 => Self::Alter,
            8 => Self::Describe,
            9 => Self::ClusterAction,
            10 => Self::DescribeConfigs,
            11 => Self::AlterConfigs,
            12 => Self::IdempotentWrite,
            13 => Self::CreateTokens,
            14 => Self::DescribeTokens,
            15 => Self::TwoPhaseCommit,
            _ => Self::Unknown,
        }
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
/// Whether an ACL grants or denies access.
pub enum AclPermissionType {
    Unknown,
    Any,
    Deny,
    Allow,
}

impl AclPermissionType {
    fn as_protocol_value(self) -> i8 {
        match self {
            Self::Unknown => 0,
            Self::Any => 1,
            Self::Deny => 2,
            Self::Allow => 3,
        }
    }

    fn from_protocol_value(value: i8) -> Self {
        match value {
            1 => Self::Any,
            2 => Self::Deny,
            3 => Self::Allow,
            _ => Self::Unknown,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Represents a pattern that ACLs use to match resources.
pub struct ResourcePattern {
    pub resource_type: ResourceType,
    pub name: String,
    pub pattern_type: PatternType,
}

impl ResourcePattern {
    pub const WILDCARD_RESOURCE: &'static str = "*";

    pub fn new(
        resource_type: ResourceType,
        name: impl Into<String>,
        pattern_type: PatternType,
    ) -> Self {
        Self {
            resource_type,
            name: name.into(),
            pattern_type,
        }
    }

    pub fn to_filter(&self) -> ResourcePatternFilter {
        ResourcePatternFilter::new(
            self.resource_type,
            Some(self.name.clone()),
            self.pattern_type,
        )
    }

    fn validate(&self) -> Result<()> {
        if matches!(self.resource_type, ResourceType::Any) {
            return Err(ValidationError::InvalidAclResourceType {
                resource_type: "ANY".to_owned(),
            }
            .into());
        }
        if matches!(self.pattern_type, PatternType::Any | PatternType::Match) {
            return Err(ValidationError::InvalidAclPatternType {
                pattern_type: format!("{:?}", self.pattern_type),
            }
            .into());
        }
        if self.name.trim().is_empty() {
            return Err(ValidationError::EmptyResourceName { resource: "ACL" }.into());
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Filter form of [`ResourcePattern`].
pub struct ResourcePatternFilter {
    pub resource_type: ResourceType,
    pub name: Option<String>,
    pub pattern_type: PatternType,
}

impl ResourcePatternFilter {
    pub fn any() -> Self {
        Self::new(ResourceType::Any, None, PatternType::Any)
    }

    pub fn new(
        resource_type: ResourceType,
        name: Option<String>,
        pattern_type: PatternType,
    ) -> Self {
        Self {
            resource_type,
            name,
            pattern_type,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Access control entry: principal, host, operation, and permission type.
pub struct AccessControlEntry {
    pub principal: String,
    pub host: String,
    pub operation: AclOperation,
    pub permission_type: AclPermissionType,
}

impl AccessControlEntry {
    pub fn new(
        principal: impl Into<String>,
        host: impl Into<String>,
        operation: AclOperation,
        permission_type: AclPermissionType,
    ) -> Self {
        Self {
            principal: principal.into(),
            host: host.into(),
            operation,
            permission_type,
        }
    }

    pub fn to_filter(&self) -> AccessControlEntryFilter {
        AccessControlEntryFilter::new(
            Some(self.principal.clone()),
            Some(self.host.clone()),
            self.operation,
            self.permission_type,
        )
    }

    fn validate(&self) -> Result<()> {
        if self.principal.trim().is_empty() {
            return Err(ValidationError::EmptyAclPrincipal.into());
        }
        if self.host.trim().is_empty() {
            return Err(ValidationError::EmptyAclHost.into());
        }
        if matches!(self.operation, AclOperation::Any) {
            return Err(ValidationError::InvalidAclOperation {
                operation: "ANY".to_owned(),
            }
            .into());
        }
        if matches!(self.permission_type, AclPermissionType::Any) {
            return Err(ValidationError::InvalidAclPermissionType {
                permission_type: "ANY".to_owned(),
            }
            .into());
        }
        Ok(())
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Filter form of [`AccessControlEntry`].
pub struct AccessControlEntryFilter {
    pub principal: Option<String>,
    pub host: Option<String>,
    pub operation: AclOperation,
    pub permission_type: AclPermissionType,
}

impl AccessControlEntryFilter {
    pub fn any() -> Self {
        Self::new(None, None, AclOperation::Any, AclPermissionType::Any)
    }

    pub fn new(
        principal: Option<String>,
        host: Option<String>,
        operation: AclOperation,
        permission_type: AclPermissionType,
    ) -> Self {
        Self {
            principal,
            host,
            operation,
            permission_type,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Binding between a resource pattern and an access control entry.
pub struct AclBinding {
    pub pattern: ResourcePattern,
    pub entry: AccessControlEntry,
}

impl AclBinding {
    pub fn new(pattern: ResourcePattern, entry: AccessControlEntry) -> Self {
        Self { pattern, entry }
    }

    pub fn to_filter(&self) -> AclBindingFilter {
        AclBindingFilter::new(self.pattern.to_filter(), self.entry.to_filter())
    }

    fn validate(&self) -> Result<()> {
        self.pattern.validate()?;
        self.entry.validate()
    }

    fn into_creation(self) -> Result<AclCreation> {
        self.validate()?;
        Ok(AclCreation::default()
            .with_resource_type(self.pattern.resource_type.as_protocol_value())
            .with_resource_name(StrBytes::from_string(self.pattern.name))
            .with_resource_pattern_type(self.pattern.pattern_type.as_protocol_value())
            .with_principal(StrBytes::from_string(self.entry.principal))
            .with_host(StrBytes::from_string(self.entry.host))
            .with_operation(self.entry.operation.as_protocol_value())
            .with_permission_type(self.entry.permission_type.as_protocol_value()))
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Filter matching ACL bindings.
pub struct AclBindingFilter {
    pub pattern_filter: ResourcePatternFilter,
    pub entry_filter: AccessControlEntryFilter,
}

impl AclBindingFilter {
    pub fn any() -> Self {
        Self::new(
            ResourcePatternFilter::any(),
            AccessControlEntryFilter::any(),
        )
    }

    pub fn new(
        pattern_filter: ResourcePatternFilter,
        entry_filter: AccessControlEntryFilter,
    ) -> Self {
        Self {
            pattern_filter,
            entry_filter,
        }
    }

    fn to_describe_request(&self) -> DescribeAclsRequest {
        DescribeAclsRequest::default()
            .with_resource_type_filter(self.pattern_filter.resource_type.as_protocol_value())
            .with_resource_name_filter(self.pattern_filter.name.clone().map(StrBytes::from_string))
            .with_pattern_type_filter(self.pattern_filter.pattern_type.as_protocol_value())
            .with_principal_filter(
                self.entry_filter
                    .principal
                    .clone()
                    .map(StrBytes::from_string),
            )
            .with_host_filter(self.entry_filter.host.clone().map(StrBytes::from_string))
            .with_operation(self.entry_filter.operation.as_protocol_value())
            .with_permission_type(self.entry_filter.permission_type.as_protocol_value())
    }

    fn into_delete_filter(self) -> DeleteAclsFilter {
        DeleteAclsFilter::default()
            .with_resource_type_filter(self.pattern_filter.resource_type.as_protocol_value())
            .with_resource_name_filter(self.pattern_filter.name.map(StrBytes::from_string))
            .with_pattern_type_filter(self.pattern_filter.pattern_type.as_protocol_value())
            .with_principal_filter(self.entry_filter.principal.map(StrBytes::from_string))
            .with_host_filter(self.entry_filter.host.map(StrBytes::from_string))
            .with_operation(self.entry_filter.operation.as_protocol_value())
            .with_permission_type(self.entry_filter.permission_type.as_protocol_value())
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// ACLs deleted for one delete filter.
pub struct DeleteAclsResult {
    pub matching_acls: Vec<AclBinding>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Summary returned by [`KafkaAdmin::list_topics`].
pub struct TopicListing {
    /// Topic name.
    pub name: String,
    /// Stable topic ID when the broker returns one.
    pub topic_id: Option<Uuid>,
    /// Whether Kafka marks the topic as internal.
    pub is_internal: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Metadata for one partition in a [`TopicDescription`].
pub struct TopicPartitionDescription {
    /// Partition number.
    pub partition: i32,
    /// Broker ID currently acting as leader.
    pub leader_id: i32,
    /// Leader epoch reported by the broker.
    pub leader_epoch: i32,
    /// Broker IDs hosting replicas for this partition.
    pub replica_nodes: Vec<i32>,
    /// Broker IDs currently in the in-sync replica set.
    pub isr_nodes: Vec<i32>,
    /// Broker IDs for replicas currently offline.
    pub offline_replicas: Vec<i32>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Detailed metadata returned by [`KafkaAdmin::describe_topics`].
pub struct TopicDescription {
    /// Topic name.
    pub name: String,
    /// Stable topic ID when the broker returns one.
    pub topic_id: Option<Uuid>,
    /// Whether Kafka marks the topic as internal.
    pub is_internal: bool,
    /// Partition metadata sorted by partition number.
    pub partitions: Vec<TopicPartitionDescription>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Broker endpoint metadata returned in a [`ClusterDescription`].
pub struct BrokerDescription {
    /// Broker ID assigned by the Kafka cluster.
    pub broker_id: i32,
    /// Broker host advertised by Kafka.
    pub host: String,
    /// Broker port advertised by Kafka.
    pub port: i32,
    /// Optional rack ID.
    pub rack: Option<String>,
    /// Whether the broker is currently fenced.
    pub is_fenced: bool,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Cluster metadata returned by [`KafkaAdmin::describe_cluster`].
pub struct ClusterDescription {
    /// Kafka cluster ID.
    pub cluster_id: String,
    /// Broker ID of the active controller.
    pub controller_id: i32,
    /// Brokers known to the cluster, sorted by broker ID.
    pub brokers: Vec<BrokerDescription>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Finalized feature level reported by the cluster.
pub struct BrokerFeatureLevel {
    /// Feature name.
    pub name: String,
    /// Finalized version level.
    pub level: i16,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Finalized broker feature update.
pub struct FeatureUpdate {
    /// Feature name.
    pub name: String,
    /// New finalized maximum version level. Values below 1 delete the finalized feature.
    pub max_version_level: i16,
    /// How the broker should apply an upgrade or downgrade.
    pub upgrade_type: FeatureUpgradeType,
}

impl FeatureUpdate {
    /// Creates an upgrade-only feature update.
    pub fn upgrade(name: impl Into<String>, max_version_level: i16) -> Self {
        Self {
            name: name.into(),
            max_version_level,
            upgrade_type: FeatureUpgradeType::Upgrade,
        }
    }

    /// Creates a safe downgrade or deletion update.
    pub fn safe_downgrade(name: impl Into<String>, max_version_level: i16) -> Self {
        Self {
            name: name.into(),
            max_version_level,
            upgrade_type: FeatureUpgradeType::SafeDowngrade,
        }
    }

    /// Creates an unsafe downgrade or deletion update.
    pub fn unsafe_downgrade(name: impl Into<String>, max_version_level: i16) -> Self {
        Self {
            name: name.into(),
            max_version_level,
            upgrade_type: FeatureUpgradeType::UnsafeDowngrade,
        }
    }

    fn into_request_update(self, version: i16) -> Result<FeatureUpdateKey> {
        let name = validate_feature_name(self.name)?;
        let allow_downgrade = self.upgrade_type.allows_downgrade();
        let mut update = FeatureUpdateKey::default()
            .with_feature(StrBytes::from_string(name))
            .with_max_version_level(self.max_version_level);
        if version == 0 {
            update = update.with_allow_downgrade(allow_downgrade);
        } else {
            update = update.with_upgrade_type(self.upgrade_type.as_protocol_value());
        }
        Ok(update)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
/// Type of finalized feature update to perform.
pub enum FeatureUpgradeType {
    /// Only allow upgrades.
    Upgrade,
    /// Allow safe, lossless downgrades.
    SafeDowngrade,
    /// Allow unsafe, potentially lossy downgrades.
    UnsafeDowngrade,
}

impl FeatureUpgradeType {
    fn as_protocol_value(self) -> i8 {
        match self {
            Self::Upgrade => 1,
            Self::SafeDowngrade => 2,
            Self::UnsafeDowngrade => 3,
        }
    }

    fn allows_downgrade(self) -> bool {
        !matches!(self, Self::Upgrade)
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Summary returned by [`KafkaAdmin::list_groups`].
pub struct ConsumerGroupListing {
    /// Consumer group ID.
    pub group_id: String,
    /// Protocol type reported by Kafka.
    pub protocol_type: String,
    /// Group state when provided by the broker.
    pub state: Option<String>,
    /// Group type when provided by newer broker versions, such as `classic` or `share`.
    pub group_type: Option<String>,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum DescribeGroupApi {
    Consumer,
    Classic,
    Share,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Detailed group metadata returned by group describe APIs.
pub struct ConsumerGroupDescription {
    /// Consumer group ID.
    pub group_id: String,
    /// Group state reported by Kafka.
    pub state: String,
    /// Protocol type, such as `consumer`, `classic`, or `share`.
    pub protocol_type: String,
    /// Protocol-specific data, usually the assignor or protocol name.
    pub protocol_data: String,
    /// Members currently in the group.
    pub members: Vec<ConsumerGroupMemberDescription>,
    /// Authorized operations bitset when returned by the broker.
    pub authorized_operations: Option<i32>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Metadata for one member of a described consumer group.
pub struct ConsumerGroupMemberDescription {
    /// Kafka-assigned member ID.
    pub member_id: String,
    /// Static membership instance ID, when configured.
    pub group_instance_id: Option<String>,
    /// Client ID reported by the member.
    pub client_id: String,
    /// Host reported by the member.
    pub client_host: String,
    /// Count derived from member metadata. For consumer/share groups this is
    /// the number of subscribed topic names; for classic groups it is decoded
    /// from the classic subscription payload when possible.
    pub member_metadata_bytes: usize,
    /// Count derived from member assignment metadata. For consumer/share groups
    /// this is the number of assigned partitions; for classic groups it is
    /// decoded from the classic assignment payload when possible.
    pub member_assignment_bytes: usize,
}

/// Backwards-compatible alias for [`ConsumerGroupListing`].
pub type GroupListing = ConsumerGroupListing;

/// Backwards-compatible alias for [`ConsumerGroupDescription`].
pub type GroupDescription = ConsumerGroupDescription;

/// Backwards-compatible alias for [`ConsumerGroupMemberDescription`].
pub type GroupMemberDescription = ConsumerGroupMemberDescription;

#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)]
/// Kafka resource type used by config APIs.
pub enum ConfigResourceType {
    /// Unknown or unsupported resource type.
    Unknown,
    /// Topic-level configuration.
    Topic,
    /// Broker configuration.
    Broker,
    /// Broker logger configuration.
    BrokerLogger,
    /// Client metrics configuration.
    ClientMetrics,
    /// Consumer group configuration.
    Group,
}

impl ConfigResourceType {
    fn as_protocol_value(self) -> i8 {
        match self {
            Self::Unknown => 0,
            Self::Topic => 2,
            Self::Broker => 4,
            Self::BrokerLogger => 8,
            Self::ClientMetrics => 16,
            Self::Group => 32,
        }
    }

    fn from_protocol_value(value: i8) -> Self {
        match value {
            2 => Self::Topic,
            4 => Self::Broker,
            8 => Self::BrokerLogger,
            16 => Self::ClientMetrics,
            32 => Self::Group,
            _ => Self::Unknown,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Hash)]
/// Kafka resource identified for config describe or alter operations.
///
/// Use [`ConfigResource::topic`] and [`ConfigResource::group`] for the common
/// cases, or [`ConfigResource::new`] for broker and other resource types.
pub struct ConfigResource {
    /// Type of Kafka resource.
    pub resource_type: ConfigResourceType,
    /// Resource name, such as a topic name, group ID, or broker ID string.
    pub resource_name: String,
}

impl ConfigResource {
    /// Creates a config resource identifier.
    pub fn new(resource_type: ConfigResourceType, resource_name: impl Into<String>) -> Self {
        Self {
            resource_type,
            resource_name: resource_name.into(),
        }
    }

    /// Creates a topic config resource.
    pub fn topic(resource_name: impl Into<String>) -> Self {
        Self::new(ConfigResourceType::Topic, resource_name)
    }

    /// Creates a group config resource.
    pub fn group(resource_name: impl Into<String>) -> Self {
        Self::new(ConfigResourceType::Group, resource_name)
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// One Kafka configuration entry returned by [`KafkaAdmin::describe_configs`].
pub struct ConfigEntry {
    /// Config key.
    pub name: String,
    /// Config value. Sensitive entries are usually returned as `None`.
    pub value: Option<String>,
    /// Whether Kafka reports the config as read-only.
    pub read_only: bool,
    /// Broker protocol config source value.
    pub config_source: i8,
    /// Whether the broker marks the config as sensitive.
    pub is_sensitive: bool,
    /// Broker protocol config type value when supported.
    pub config_type: Option<i8>,
    /// Broker-supplied config documentation when available.
    pub documentation: Option<String>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Described configuration for one Kafka resource.
pub struct ConfigResourceConfig {
    /// Resource that was described.
    pub resource: ConfigResource,
    /// Config entries keyed by config name.
    pub entries: BTreeMap<String, ConfigEntry>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Log directory usage returned by [`KafkaAdmin::describe_log_dirs`].
pub struct BrokerLogDirs {
    /// Broker ID that returned these log directories.
    pub broker_id: i32,
    /// Log directory details reported by the broker.
    pub log_dirs: Vec<LogDirDescription>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// One broker log directory.
pub struct LogDirDescription {
    /// Absolute log directory path.
    pub log_dir: String,
    /// Broker protocol error code for this log directory, if any.
    pub error_code: i16,
    /// Total size of the filesystem containing this log dir when returned by Kafka.
    pub total_bytes: Option<i64>,
    /// Usable size of the filesystem containing this log dir when returned by Kafka.
    pub usable_bytes: Option<i64>,
    /// Replica logs in this directory.
    pub replicas: Vec<ReplicaLogDirDescription>,
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// Size information for one replica log in a broker log directory.
pub struct ReplicaLogDirDescription {
    /// Topic name.
    pub topic: String,
    /// Partition number.
    pub partition: i32,
    /// Log segment size in bytes.
    pub size_bytes: i64,
    /// Offset lag reported by Kafka.
    pub offset_lag: i64,
    /// Whether this is a future replica log.
    pub is_future: bool,
}

impl ConfigResourceConfig {
    /// Returns a config entry by name.
    pub fn entry(&self, name: &str) -> Option<&ConfigEntry> {
        self.entries.get(name)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
/// Operation type for an incremental config change.
pub enum AlterConfigOpType {
    /// Set or replace the config value.
    Set,
    /// Delete the config and fall back to Kafka defaults.
    Delete,
    /// Append to a list-valued config.
    Append,
    /// Remove from a list-valued config.
    Subtract,
}

impl AlterConfigOpType {
    fn as_protocol_value(self) -> i8 {
        match self {
            Self::Set => 0,
            Self::Delete => 1,
            Self::Append => 2,
            Self::Subtract => 3,
        }
    }
}

#[derive(Debug, Clone, PartialEq, Eq)]
/// One config mutation used with [`KafkaAdmin::incremental_alter_configs`].
pub struct AlterConfigOp {
    /// Config key.
    pub name: String,
    /// Operation to apply.
    pub op_type: AlterConfigOpType,
    /// Config value for operations that require one.
    pub value: Option<String>,
}

impl AlterConfigOp {
    /// Creates a config set operation.
    pub fn set(name: impl Into<String>, value: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            op_type: AlterConfigOpType::Set,
            value: Some(value.into()),
        }
    }

    /// Creates a config delete operation.
    pub fn delete(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            op_type: AlterConfigOpType::Delete,
            value: None,
        }
    }
}

#[derive(Debug, Clone)]
/// Client for Kafka admin operations.
///
/// `KafkaAdmin` is an async client for cluster, topic, group, config, feature,
/// and SCRAM credential operations. Construct it either from an [`AdminConfig`]
/// when you need explicit configuration, or through [`KafkaClient::admin`] for
/// the builder-style API.
///
/// ```no_run
/// # async fn example() -> kafkit_client::Result<()> {
/// use kafkit_client::{AdminConfig, KafkaAdmin, NewTopic};
///
/// let admin = KafkaAdmin::connect(AdminConfig::new("localhost:9092")).await?;
///
/// admin.create_topics([NewTopic::new("orders", 3, 1)]).await?;
/// let topics = admin.list_topics().await?;
/// # let _ = topics;
/// # Ok(())
/// # }
/// ```
///
/// [`KafkaClient::admin`]: crate::KafkaClient::admin
pub struct KafkaAdmin {
    config: AdminConfig,
}

impl KafkaAdmin {
    #[instrument(
        name = "admin.connect",
        level = "debug",
        skip(config),
        fields(
            bootstrap_server_count = config.bootstrap_servers.len(),
            client_id = %config.client_id
        )
    )]
    /// Connects to Kafka and returns an admin client.
    ///
    /// This performs a lightweight bootstrap connection so configuration,
    /// security settings, and broker reachability fail before the client is
    /// returned.
    pub async fn connect(config: AdminConfig) -> Result<Self> {
        let admin = Self { config };
        admin.warm_up().await?;
        debug!("admin client connected");
        Ok(admin)
    }

    #[instrument(name = "admin.create_topics", level = "debug", skip(self, topics))]
    /// Creates one or more topics.
    ///
    /// Existing topics are treated as success, matching the common
    /// create-if-missing setup flow. Passing an empty iterator is a no-op.
    pub async fn create_topics<I>(&self, topics: I) -> Result<()>
    where
        I: IntoIterator<Item = NewTopic>,
    {
        let topics = topics
            .into_iter()
            .map(NewTopic::into_request_topic)
            .collect::<Result<Vec<_>>>()?;
        if topics.is_empty() {
            return Ok(());
        }

        let request = CreateTopicsRequest::default()
            .with_topics(topics)
            .with_timeout_ms(duration_to_i32_ms(self.config.request_timeout)?)
            .with_validate_only(false);
        let response: CreateTopicsResponse = self
            .send_request::<CreateTopicsRequest>(CREATE_TOPICS_VERSION_CAP, &request)
            .await?;

        for topic in response.topics {
            let name = topic.name.0.to_string();
            if let Some(error) = topic
                .error_code
                .err()
                .filter(|error| !is_ignorable_create_topic_error(*error))
            {
                return Err(admin_broker_error("create_topic", Some(name), error));
            }
        }

        Ok(())
    }

    #[instrument(name = "admin.delete_topics", level = "debug", skip(self, topics))]
    /// Deletes one or more topics by name.
    ///
    /// Passing an empty iterator is a no-op. Kafka must have topic deletion
    /// enabled for the broker to complete the operation.
    pub async fn delete_topics<I, S>(&self, topics: I) -> Result<()>
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        let topic_names = topics
            .into_iter()
            .map(|topic| validate_topic_name(topic.into()))
            .collect::<Result<Vec<_>>>()?;
        if topic_names.is_empty() {
            return Ok(());
        }

        let request = DeleteTopicsRequest::default()
            .with_topic_names(
                topic_names
                    .iter()
                    .cloned()
                    .map(StrBytes::from_string)
                    .map(Into::into)
                    .collect(),
            )
            .with_timeout_ms(duration_to_i32_ms(self.config.request_timeout)?);
        let response: DeleteTopicsResponse = self
            .send_request::<DeleteTopicsRequest>(DELETE_TOPICS_VERSION_CAP, &request)
            .await?;

        for topic in response.responses {
            let name = topic
                .name
                .as_ref()
                .map(|name| name.0.to_string())
                .unwrap_or_else(|| "<unknown>".to_owned());
            if let Some(error) = topic.error_code.err() {
                return Err(admin_broker_error("delete_topic", Some(name), error));
            }
        }

        Ok(())
    }

    #[instrument(
        name = "admin.create_partitions",
        level = "debug",
        skip(self, partitions)
    )]
    /// Increases partition counts for one or more topics.
    ///
    /// Each item is a `(topic_name, NewPartitions)` pair. Kafka only supports
    /// increasing the total partition count; it cannot shrink a topic.
    pub async fn create_partitions<I, S>(&self, partitions: I) -> Result<()>
    where
        I: IntoIterator<Item = (S, NewPartitions)>,
        S: Into<String>,
    {
        let topics = partitions
            .into_iter()
            .map(|(topic, new_partitions)| new_partitions.into_request_topic(topic.into()))
            .collect::<Result<Vec<_>>>()?;
        if topics.is_empty() {
            return Ok(());
        }

        let request = CreatePartitionsRequest::default()
            .with_topics(topics)
            .with_timeout_ms(duration_to_i32_ms(self.config.request_timeout)?)
            .with_validate_only(false);
        let response: CreatePartitionsResponse = self
            .send_request::<CreatePartitionsRequest>(CREATE_PARTITIONS_VERSION_CAP, &request)
            .await?;

        for topic in response.results {
            let name = topic.name.0.to_string();
            if let Some(error) = topic.error_code.err() {
                return Err(admin_broker_error("create_partitions", Some(name), error));
            }
        }

        Ok(())
    }

    #[instrument(name = "admin.list_topics", level = "debug", skip(self))]
    /// Lists topics known to the cluster.
    ///
    /// The returned list is sorted by topic name.
    pub async fn list_topics(&self) -> Result<Vec<TopicListing>> {
        let response = self.fetch_metadata(None).await?;
        let mut topics = Vec::new();

        for topic in response.topics {
            let Some(name) = topic.name.as_ref().map(|name| name.0.to_string()) else {
                continue;
            };
            if let Some(error) = topic.error_code.err() {
                return Err(admin_broker_error("list_topics", Some(name), error));
            }

            topics.push(TopicListing {
                name,
                topic_id: (!topic.topic_id.is_nil()).then_some(topic.topic_id),
                is_internal: topic.is_internal,
            });
        }

        topics.sort_by(|left, right| left.name.cmp(&right.name));
        Ok(topics)
    }

    #[instrument(name = "admin.describe_topics", level = "debug", skip(self, topics))]
    /// Describes selected topics.
    ///
    /// The returned descriptions preserve the input topic order. Passing an
    /// empty iterator returns an empty vector.
    pub async fn describe_topics<I, S>(&self, topics: I) -> Result<Vec<TopicDescription>>
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        let requested_topics = topics
            .into_iter()
            .map(|topic| validate_topic_name(topic.into()))
            .collect::<Result<Vec<_>>>()?;
        if requested_topics.is_empty() {
            return Ok(Vec::new());
        }

        let response = self.fetch_metadata(Some(&requested_topics)).await?;
        let mut descriptions = BTreeMap::new();

        for topic in response.topics {
            let name = topic
                .name
                .as_ref()
                .map(|name| name.0.to_string())
                .unwrap_or_default();
            if let Some(error) = topic.error_code.err() {
                let label = if name.is_empty() { "<unknown>" } else { &name };
                return Err(admin_broker_error(
                    "describe_topic",
                    Some(label.to_owned()),
                    error,
                ));
            }

            let mut partitions = topic
                .partitions
                .into_iter()
                .map(|partition| {
                    if let Some(error) = partition.error_code.err() {
                        return Err(admin_broker_error(
                            "describe_topic_partition",
                            Some(format!("{name}:{}", partition.partition_index)),
                            error,
                        ));
                    }

                    Ok(TopicPartitionDescription {
                        partition: partition.partition_index,
                        leader_id: partition.leader_id.0,
                        leader_epoch: partition.leader_epoch,
                        replica_nodes: partition.replica_nodes.into_iter().map(|id| id.0).collect(),
                        isr_nodes: partition.isr_nodes.into_iter().map(|id| id.0).collect(),
                        offline_replicas: partition
                            .offline_replicas
                            .into_iter()
                            .map(|id| id.0)
                            .collect(),
                    })
                })
                .collect::<Result<Vec<_>>>()?;
            partitions.sort_by_key(|partition| partition.partition);

            descriptions.insert(
                name.clone(),
                TopicDescription {
                    name,
                    topic_id: (!topic.topic_id.is_nil()).then_some(topic.topic_id),
                    is_internal: topic.is_internal,
                    partitions,
                },
            );
        }

        requested_topics
            .into_iter()
            .map(|topic| {
                descriptions.remove(&topic).ok_or_else(|| {
                    anyhow!("metadata response did not include topic '{topic}'").into()
                })
            })
            .collect()
    }

    #[instrument(name = "admin.describe_cluster", level = "debug", skip(self))]
    /// Describes cluster ID, controller, and broker endpoints.
    pub async fn describe_cluster(&self) -> Result<ClusterDescription> {
        let (mut connection, version) = self
            .connect_with_version::<DescribeClusterRequest>(DESCRIBE_CLUSTER_VERSION_CAP)
            .await?;
        let mut request =
            DescribeClusterRequest::default().with_include_cluster_authorized_operations(false);
        if version >= 1 {
            request = request.with_endpoint_type(1);
        }
        if version >= 2 {
            request = request.with_include_fenced_brokers(true);
        }

        let response: DescribeClusterResponse = connection
            .send_request::<DescribeClusterRequest>(&self.config.client_id, version, &request)
            .await?;
        if let Some(error) = response.error_code.err() {
            return Err(admin_broker_error(
                "describe_cluster",
                None::<String>,
                error,
            ));
        }

        let mut brokers = response
            .brokers
            .into_iter()
            .map(|broker| BrokerDescription {
                broker_id: broker.broker_id.0,
                host: broker.host.to_string(),
                port: broker.port,
                rack: broker.rack.map(|rack| rack.to_string()),
                is_fenced: broker.is_fenced,
            })
            .collect::<Vec<_>>();
        brokers.sort_by_key(|broker| broker.broker_id);

        Ok(ClusterDescription {
            cluster_id: response.cluster_id.to_string(),
            controller_id: response.controller_id.0,
            brokers,
        })
    }

    #[instrument(name = "admin.describe_log_dirs", level = "debug", skip(self, brokers))]
    /// Describes broker log directories and replica log sizes.
    ///
    /// Kafka serves this API from each broker. Pass broker metadata returned by
    /// [`KafkaAdmin::describe_cluster`]. When `topics` is `None`, Kafka returns
    /// all topic-partitions hosted by each broker.
    pub async fn describe_log_dirs<I>(
        &self,
        brokers: I,
        topics: Option<&[TopicPartition]>,
    ) -> Result<Vec<BrokerLogDirs>>
    where
        I: IntoIterator<Item = BrokerDescription>,
    {
        let request_topics = topics.map(describe_log_dirs_topics);
        let mut handles = Vec::new();
        for broker in brokers {
            let config = self.config.clone();
            let request_topics = request_topics.clone();
            let broker_id = broker.broker_id;
            handles.push((
                broker_id,
                tokio::spawn(async move {
                    describe_log_dirs_for_broker(config, broker, request_topics).await
                }),
            ));
        }

        let mut results = Vec::with_capacity(handles.len());
        let mut first_error = None;
        for (broker_id, handle) in handles {
            match handle.await {
                Ok(Ok(broker_result)) => results.push(broker_result),
                Ok(Err(error)) => {
                    first_error.get_or_insert(error);
                }
                Err(error) => {
                    first_error.get_or_insert(
                        anyhow!("describe log dirs for broker {broker_id} task failed: {error}")
                            .into(),
                    );
                }
            }
        }
        if let Some(error) = first_error {
            return Err(error);
        }
        results.sort_by_key(|broker| broker.broker_id);
        Ok(results)
    }

    #[instrument(name = "admin.list_groups", level = "debug", skip(self))]
    /// Lists consumer groups known to the cluster.
    ///
    /// The returned list is sorted by group ID.
    pub async fn list_groups(&self) -> Result<Vec<ConsumerGroupListing>> {
        let response: ListGroupsResponse = self
            .send_request::<ListGroupsRequest>(
                LIST_GROUPS_VERSION_CAP,
                &ListGroupsRequest::default(),
            )
            .await?;
        if let Some(error) = response.error_code.err() {
            return Err(admin_broker_error("list_groups", None::<String>, error));
        }

        let mut groups = response
            .groups
            .into_iter()
            .map(|group| ConsumerGroupListing {
                group_id: group.group_id.to_string(),
                protocol_type: group.protocol_type.to_string(),
                state: (!group.group_state.is_empty()).then(|| group.group_state.to_string()),
                group_type: (!group.group_type.is_empty()).then(|| group.group_type.to_string()),
            })
            .collect::<Vec<_>>();
        groups.sort_by(|left, right| left.group_id.cmp(&right.group_id));
        Ok(groups)
    }

    #[instrument(name = "admin.list_consumer_groups", level = "debug", skip(self))]
    /// Alias for [`KafkaAdmin::list_groups`].
    pub async fn list_consumer_groups(&self) -> Result<Vec<ConsumerGroupListing>> {
        self.list_groups().await
    }

    #[instrument(name = "admin.describe_groups", level = "debug", skip(self, groups))]
    /// Describes groups, routing each group to the appropriate Kafka describe API.
    ///
    /// Newer brokers report group type in `ListGroups`; this method uses that
    /// metadata to describe consumer, share, and classic groups. The returned
    /// descriptions preserve the input group order.
    pub async fn describe_groups<I, S>(&self, groups: I) -> Result<Vec<GroupDescription>>
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        let group_ids = groups
            .into_iter()
            .map(|group| validate_group_id(group.into()))
            .collect::<Result<Vec<_>>>()?;
        if group_ids.is_empty() {
            return Ok(Vec::new());
        }

        let listed_group_types = self
            .list_groups()
            .await?
            .into_iter()
            .map(|group| (group.group_id, group.group_type))
            .collect::<BTreeMap<_, _>>();
        let mut consumer_groups = Vec::new();
        let mut share_groups = Vec::new();
        let mut classic_groups = Vec::new();
        for group_id in &group_ids {
            match describe_group_api_for_listed_type(
                listed_group_types
                    .get(group_id)
                    .and_then(|group_type| group_type.as_deref()),
            ) {
                DescribeGroupApi::Share => {
                    share_groups.push(group_id.clone());
                }
                DescribeGroupApi::Classic => {
                    classic_groups.push(group_id.clone());
                }
                DescribeGroupApi::Consumer => {
                    consumer_groups.push(group_id.clone());
                }
            }
        }

        let mut descriptions = BTreeMap::new();
        for group in self.describe_consumer_groups(consumer_groups).await? {
            descriptions.insert(group.group_id.clone(), group);
        }
        for group in self.describe_share_groups(share_groups).await? {
            descriptions.insert(group.group_id.clone(), group);
        }
        for group in self.describe_classic_groups(classic_groups).await? {
            descriptions.insert(group.group_id.clone(), group);
        }

        group_ids
            .into_iter()
            .map(|group_id| {
                descriptions.remove(&group_id).ok_or_else(|| {
                    anyhow!("describe groups response did not include group '{group_id}'").into()
                })
            })
            .collect()
    }

    #[instrument(
        name = "admin.describe_consumer_groups",
        level = "debug",
        skip(self, groups)
    )]
    /// Describes KIP-848 consumer groups.
    ///
    /// Use [`KafkaAdmin::describe_groups`] when the group type is not known.
    pub async fn describe_consumer_groups<I, S>(
        &self,
        groups: I,
    ) -> Result<Vec<ConsumerGroupDescription>>
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        let group_ids = groups
            .into_iter()
            .map(|group| validate_group_id(group.into()))
            .collect::<Result<Vec<_>>>()?;
        if group_ids.is_empty() {
            return Ok(Vec::new());
        }

        let request = ConsumerGroupDescribeRequest::default()
            .with_group_ids(
                group_ids
                    .iter()
                    .cloned()
                    .map(StrBytes::from_string)
                    .map(Into::into)
                    .collect(),
            )
            .with_include_authorized_operations(false);
        let response: ConsumerGroupDescribeResponse = self
            .send_request::<ConsumerGroupDescribeRequest>(
                CONSUMER_GROUP_DESCRIBE_VERSION_CAP,
                &request,
            )
            .await?;

        let mut descriptions = BTreeMap::new();
        for group in response.groups {
            let group_id = group.group_id.to_string();
            if let Some(error) = group.error_code.err() {
                return Err(admin_broker_error(
                    "describe_consumer_group",
                    Some(group_id),
                    error,
                ));
            }

            descriptions.insert(
                group_id.clone(),
                ConsumerGroupDescription {
                    group_id,
                    state: group.group_state.to_string(),
                    protocol_type: "consumer".to_owned(),
                    protocol_data: group.assignor_name.to_string(),
                    members: group
                        .members
                        .into_iter()
                        .map(|member| ConsumerGroupMemberDescription {
                            member_id: member.member_id.to_string(),
                            group_instance_id: member
                                .instance_id
                                .map(|instance_id| instance_id.to_string()),
                            client_id: member.client_id.to_string(),
                            client_host: member.client_host.to_string(),
                            member_metadata_bytes: member.subscribed_topic_names.len(),
                            member_assignment_bytes: assignment_partition_count(&member.assignment),
                        })
                        .collect(),
                    authorized_operations: (group.authorized_operations != i32::MIN)
                        .then_some(group.authorized_operations),
                },
            );
        }

        group_ids
            .into_iter()
            .map(|group_id| {
                descriptions.remove(&group_id).ok_or_else(|| {
                    anyhow!("describe groups response did not include group '{group_id}'").into()
                })
            })
            .collect()
    }

    #[instrument(
        name = "admin.describe_classic_groups",
        level = "debug",
        skip(self, groups)
    )]
    /// Describes classic consumer groups with Kafka's legacy describe-groups API.
    ///
    /// Use [`KafkaAdmin::describe_groups`] when the group type is not known.
    pub async fn describe_classic_groups<I, S>(
        &self,
        groups: I,
    ) -> Result<Vec<ConsumerGroupDescription>>
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        let group_ids = groups
            .into_iter()
            .map(|group| validate_group_id(group.into()))
            .collect::<Result<Vec<_>>>()?;
        if group_ids.is_empty() {
            return Ok(Vec::new());
        }

        let request = DescribeGroupsRequest::default()
            .with_groups(
                group_ids
                    .iter()
                    .cloned()
                    .map(StrBytes::from_string)
                    .map(Into::into)
                    .collect(),
            )
            .with_include_authorized_operations(false);
        let response: DescribeGroupsResponse = self
            .send_request::<DescribeGroupsRequest>(DESCRIBE_GROUPS_VERSION_CAP, &request)
            .await?;

        let mut descriptions = BTreeMap::new();
        for group in response.groups {
            let group_id = group.group_id.to_string();
            if let Some(error) = group.error_code.err() {
                return Err(admin_broker_error(
                    "describe_classic_group",
                    Some(group_id),
                    error,
                ));
            }

            descriptions.insert(
                group_id.clone(),
                ConsumerGroupDescription {
                    group_id,
                    state: group.group_state.to_string(),
                    protocol_type: group.protocol_type.to_string(),
                    protocol_data: group.protocol_data.to_string(),
                    members: group
                        .members
                        .into_iter()
                        .map(|member| {
                            let member_metadata_bytes =
                                classic_subscription_topic_count(&member.member_metadata);
                            let member_assignment_bytes =
                                classic_assignment_partition_count(&member.member_assignment);
                            ConsumerGroupMemberDescription {
                                member_id: member.member_id.to_string(),
                                group_instance_id: member
                                    .group_instance_id
                                    .map(|instance_id| instance_id.to_string()),
                                client_id: member.client_id.to_string(),
                                client_host: member.client_host.to_string(),
                                member_metadata_bytes,
                                member_assignment_bytes,
                            }
                        })
                        .collect(),
                    authorized_operations: (group.authorized_operations != i32::MIN)
                        .then_some(group.authorized_operations),
                },
            );
        }

        group_ids
            .into_iter()
            .map(|group_id| {
                descriptions.remove(&group_id).ok_or_else(|| {
                    anyhow!("describe classic groups response did not include group '{group_id}'")
                        .into()
                })
            })
            .collect()
    }

    #[instrument(
        name = "admin.describe_share_groups",
        level = "debug",
        skip(self, groups)
    )]
    /// Describes Kafka share groups.
    ///
    /// Use [`KafkaAdmin::describe_groups`] when the group type is not known.
    pub async fn describe_share_groups<I, S>(
        &self,
        groups: I,
    ) -> Result<Vec<ConsumerGroupDescription>>
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        let group_ids = groups
            .into_iter()
            .map(|group| validate_group_id(group.into()))
            .collect::<Result<Vec<_>>>()?;
        if group_ids.is_empty() {
            return Ok(Vec::new());
        }

        let request = ShareGroupDescribeRequest::default()
            .with_group_ids(
                group_ids
                    .iter()
                    .cloned()
                    .map(StrBytes::from_string)
                    .map(Into::into)
                    .collect(),
            )
            .with_include_authorized_operations(false);
        let response: ShareGroupDescribeResponse = self
            .send_request::<ShareGroupDescribeRequest>(SHARE_GROUP_DESCRIBE_VERSION_CAP, &request)
            .await?;

        let mut descriptions = BTreeMap::new();
        for group in response.groups {
            let group_id = group.group_id.to_string();
            if let Some(error) = group.error_code.err() {
                return Err(admin_broker_error(
                    "describe_share_group",
                    Some(group_id),
                    error,
                ));
            }

            descriptions.insert(
                group_id.clone(),
                ConsumerGroupDescription {
                    group_id,
                    state: group.group_state.to_string(),
                    protocol_type: "share".to_owned(),
                    protocol_data: group.assignor_name.to_string(),
                    members: group
                        .members
                        .into_iter()
                        .map(|member| ConsumerGroupMemberDescription {
                            member_id: member.member_id.to_string(),
                            group_instance_id: None,
                            client_id: member.client_id.to_string(),
                            client_host: member.client_host.to_string(),
                            member_metadata_bytes: member.subscribed_topic_names.len(),
                            member_assignment_bytes: share_assignment_partition_count(
                                &member.assignment,
                            ),
                        })
                        .collect(),
                    authorized_operations: (group.authorized_operations != i32::MIN)
                        .then_some(group.authorized_operations),
                },
            );
        }

        group_ids
            .into_iter()
            .map(|group_id| {
                descriptions.remove(&group_id).ok_or_else(|| {
                    anyhow!("describe share groups response did not include group '{group_id}'")
                        .into()
                })
            })
            .collect()
    }

    #[instrument(name = "admin.delete_groups", level = "debug", skip(self, groups))]
    /// Deletes one or more groups by ID.
    ///
    /// Passing an empty iterator is a no-op.
    pub async fn delete_groups<I, S>(&self, groups: I) -> Result<()>
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        let group_ids = groups
            .into_iter()
            .map(|group| validate_group_id(group.into()))
            .collect::<Result<Vec<_>>>()?;
        if group_ids.is_empty() {
            return Ok(());
        }

        let request = DeleteGroupsRequest::default().with_groups_names(
            group_ids
                .iter()
                .cloned()
                .map(StrBytes::from_string)
                .map(Into::into)
                .collect(),
        );
        let response: DeleteGroupsResponse = self
            .send_request::<DeleteGroupsRequest>(DELETE_GROUPS_VERSION_CAP, &request)
            .await?;

        for result in response.results {
            if let Some(error) = result.error_code.err() {
                return Err(admin_broker_error(
                    "delete_group",
                    Some(result.group_id.to_string()),
                    error,
                ));
            }
        }
        Ok(())
    }

    #[instrument(
        name = "admin.delete_consumer_groups",
        level = "debug",
        skip(self, groups)
    )]
    /// Alias for [`KafkaAdmin::delete_groups`].
    pub async fn delete_consumer_groups<I, S>(&self, groups: I) -> Result<()>
    where
        I: IntoIterator<Item = S>,
        S: Into<String>,
    {
        self.delete_groups(groups).await
    }

    #[instrument(name = "admin.describe_acls", level = "debug", skip(self, filter))]
    /// Describes ACL bindings matching a filter.
    ///
    /// The shape mirrors Kafka's `Admin#describeAcls(AclBindingFilter)`.
    pub async fn describe_acls(&self, filter: AclBindingFilter) -> Result<Vec<AclBinding>> {
        let request = filter.to_describe_request();
        let response: DescribeAclsResponse = self
            .send_request::<DescribeAclsRequest>(DESCRIBE_ACLS_VERSION_CAP, &request)
            .await?;
        if let Some(error) = response.error_code.err() {
            return Err(admin_broker_error("describe_acls", None::<String>, error));
        }

        let mut bindings = Vec::new();
        for resource in response.resources {
            let pattern = ResourcePattern::new(
                ResourceType::from_protocol_value(resource.resource_type),
                resource.resource_name.to_string(),
                PatternType::from_protocol_value(resource.pattern_type),
            );
            for acl in resource.acls {
                bindings.push(AclBinding::new(
                    pattern.clone(),
                    AccessControlEntry::new(
                        acl.principal.to_string(),
                        acl.host.to_string(),
                        AclOperation::from_protocol_value(acl.operation),
                        AclPermissionType::from_protocol_value(acl.permission_type),
                    ),
                ));
            }
        }
        bindings.sort_by(|left, right| {
            left.pattern
                .name
                .cmp(&right.pattern.name)
                .then_with(|| left.entry.principal.cmp(&right.entry.principal))
                .then_with(|| left.entry.host.cmp(&right.entry.host))
        });
        Ok(bindings)
    }

    #[instrument(name = "admin.create_acls", level = "debug", skip(self, acls))]
    /// Creates ACL bindings.
    ///
    /// The shape mirrors Kafka's `Admin#createAcls(Collection<AclBinding>)`.
    pub async fn create_acls<I>(&self, acls: I) -> Result<()>
    where
        I: IntoIterator<Item = AclBinding>,
    {
        let creations = acls
            .into_iter()
            .map(AclBinding::into_creation)
            .collect::<Result<Vec<_>>>()?;
        if creations.is_empty() {
            return Ok(());
        }

        let request = CreateAclsRequest::default().with_creations(creations);
        let response: CreateAclsResponse = self
            .send_request::<CreateAclsRequest>(CREATE_ACLS_VERSION_CAP, &request)
            .await?;
        for result in response.results {
            if let Some(error) = result.error_code.err() {
                return Err(admin_broker_error("create_acl", None::<String>, error));
            }
        }
        Ok(())
    }

    #[instrument(name = "admin.delete_acls", level = "debug", skip(self, filters))]
    /// Deletes ACL bindings matching each filter.
    ///
    /// The shape mirrors Kafka's `Admin#deleteAcls(Collection<AclBindingFilter>)`.
    pub async fn delete_acls<I>(&self, filters: I) -> Result<Vec<DeleteAclsResult>>
    where
        I: IntoIterator<Item = AclBindingFilter>,
    {
        let filters = filters
            .into_iter()
            .map(AclBindingFilter::into_delete_filter)
            .collect::<Vec<_>>();
        if filters.is_empty() {
            return Ok(Vec::new());
        }

        let request = DeleteAclsRequest::default().with_filters(filters);
        let response: DeleteAclsResponse = self
            .send_request::<DeleteAclsRequest>(DELETE_ACLS_VERSION_CAP, &request)
            .await?;
        let mut results = Vec::new();
        for result in response.filter_results {
            if let Some(error) = result.error_code.err() {
                return Err(admin_broker_error("delete_acls", None::<String>, error));
            }
            let mut matching_acls = Vec::new();
            for acl in result.matching_acls {
                if let Some(error) = acl.error_code.err() {
                    return Err(admin_broker_error(
                        "delete_acl",
                        Some(acl.resource_name.to_string()),
                        error,
                    ));
                }
                matching_acls.push(AclBinding::new(
                    ResourcePattern::new(
                        ResourceType::from_protocol_value(acl.resource_type),
                        acl.resource_name.to_string(),
                        PatternType::from_protocol_value(acl.pattern_type),
                    ),
                    AccessControlEntry::new(
                        acl.principal.to_string(),
                        acl.host.to_string(),
                        AclOperation::from_protocol_value(acl.operation),
                        AclPermissionType::from_protocol_value(acl.permission_type),
                    ),
                ));
            }
            results.push(DeleteAclsResult { matching_acls });
        }
        Ok(results)
    }

    #[instrument(
        name = "admin.describe_configs",
        level = "debug",
        skip(self, resources)
    )]
    /// Describes configuration entries for Kafka resources.
    ///
    /// ```no_run
    /// # async fn example(admin: kafkit_client::KafkaAdmin) -> kafkit_client::Result<()> {
    /// use kafkit_client::ConfigResource;
    ///
    /// let configs = admin
    ///     .describe_configs([ConfigResource::topic("orders")])
    ///     .await?;
    ///
    /// if let Some(retention) = configs[0].entry("retention.ms") {
    ///     println!("retention: {:?}", retention.value);
    /// }
    /// # Ok(())
    /// # }
    /// ```
    pub async fn describe_configs<I>(&self, resources: I) -> Result<Vec<ConfigResourceConfig>>
    where
        I: IntoIterator<Item = ConfigResource>,
    {
        let resources = resources.into_iter().collect::<Vec<_>>();
        if resources.is_empty() {
            return Ok(Vec::new());
        }

        let request = DescribeConfigsRequest::default()
            .with_resources(
                resources
                    .iter()
                    .map(|resource| {
                        DescribeConfigsResource::default()
                            .with_resource_type(resource.resource_type.as_protocol_value())
                            .with_resource_name(StrBytes::from_string(
                                resource.resource_name.clone(),
                            ))
                            .with_configuration_keys(None)
                    })
                    .collect(),
            )
            .with_include_synonyms(false);
        let response: DescribeConfigsResponse = self
            .send_request::<DescribeConfigsRequest>(DESCRIBE_CONFIGS_VERSION_CAP, &request)
            .await?;

        let mut described = BTreeMap::new();
        for resource in response.results {
            let resource_type = ConfigResourceType::from_protocol_value(resource.resource_type);
            let resource_name = resource.resource_name.to_string();
            if let Some(error) = resource.error_code.err() {
                return Err(admin_broker_error(
                    "describe_configs",
                    Some(format!("{resource_type:?}:{resource_name}")),
                    error,
                ));
            }

            let entries = resource
                .configs
                .into_iter()
                .map(|entry| {
                    let name = entry.name.to_string();
                    let config_entry = ConfigEntry {
                        name: name.clone(),
                        value: entry.value.map(|value| value.to_string()),
                        read_only: entry.read_only,
                        config_source: entry.config_source,
                        is_sensitive: entry.is_sensitive,
                        config_type: (response_supported_config_type(entry.config_type))
                            .then_some(entry.config_type),
                        documentation: entry.documentation.map(|doc| doc.to_string()),
                    };
                    (name, config_entry)
                })
                .collect();

            described.insert(
                (resource_type, resource_name.clone()),
                ConfigResourceConfig {
                    resource: ConfigResource::new(resource_type, resource_name),
                    entries,
                },
            );
        }

        resources
            .into_iter()
            .map(|resource| {
                described
                    .remove(&(resource.resource_type, resource.resource_name.clone()))
                    .ok_or_else(|| {
                        anyhow!("describe configs response did not include {:?}", resource).into()
                    })
            })
            .collect()
    }

    #[instrument(
        name = "admin.incremental_alter_configs",
        level = "debug",
        skip(self, resources)
    )]
    /// Applies incremental config changes to Kafka resources.
    ///
    /// ```no_run
    /// # async fn example(admin: kafkit_client::KafkaAdmin) -> kafkit_client::Result<()> {
    /// use kafkit_client::{AlterConfigOp, ConfigResource};
    ///
    /// admin
    ///     .incremental_alter_configs([(
    ///         ConfigResource::topic("orders"),
    ///         vec![AlterConfigOp::set("retention.ms", "604800000")],
    ///     )])
    ///     .await?;
    /// # Ok(())
    /// # }
    /// ```
    pub async fn incremental_alter_configs<I>(&self, resources: I) -> Result<()>
    where
        I: IntoIterator<Item = (ConfigResource, Vec<AlterConfigOp>)>,
    {
        let resources = resources
            .into_iter()
            .map(|(resource, ops)| {
                AlterConfigsResource::default()
                    .with_resource_type(resource.resource_type.as_protocol_value())
                    .with_resource_name(StrBytes::from_string(resource.resource_name))
                    .with_configs(
                        ops.into_iter()
                            .map(|op| {
                                AlterableConfig::default()
                                    .with_name(StrBytes::from_string(op.name))
                                    .with_config_operation(op.op_type.as_protocol_value())
                                    .with_value(op.value.map(StrBytes::from_string))
                            })
                            .collect(),
                    )
            })
            .collect::<Vec<_>>();
        if resources.is_empty() {
            return Ok(());
        }

        let request = IncrementalAlterConfigsRequest::default()
            .with_resources(resources)
            .with_validate_only(false);
        let response: IncrementalAlterConfigsResponse = self
            .send_request::<IncrementalAlterConfigsRequest>(
                INCREMENTAL_ALTER_CONFIGS_VERSION_CAP,
                &request,
            )
            .await?;

        for resource in response.responses {
            if let Some(error) = resource.error_code.err() {
                let resource_type = ConfigResourceType::from_protocol_value(resource.resource_type);
                return Err(admin_broker_error(
                    "incremental_alter_configs",
                    Some(format!("{resource_type:?}:{}", resource.resource_name)),
                    error,
                ));
            }
        }

        Ok(())
    }

    #[instrument(
        name = "admin.upsert_scram_credential",
        level = "debug",
        skip(self, user, password)
    )]
    /// Creates or replaces a SCRAM credential for a user.
    ///
    /// `mechanism` must be [`SaslMechanism::ScramSha256`] or
    /// [`SaslMechanism::ScramSha512`]. Use
    /// [`KafkaAdmin::upsert_scram_credential_with_iterations`] to override the
    /// default iteration count.
    pub async fn upsert_scram_credential(
        &self,
        user: impl Into<String>,
        mechanism: SaslMechanism,
        password: impl AsRef<[u8]>,
    ) -> Result<()> {
        self.upsert_scram_credential_with_iterations(
            user,
            mechanism,
            password,
            scram::MIN_ITERATIONS,
        )
        .await
    }

    #[instrument(
        name = "admin.upsert_scram_credential_with_iterations",
        level = "debug",
        skip(self, user, password)
    )]
    /// Creates or replaces a SCRAM credential with an explicit iteration count.
    pub async fn upsert_scram_credential_with_iterations(
        &self,
        user: impl Into<String>,
        mechanism: SaslMechanism,
        password: impl AsRef<[u8]>,
        iterations: i32,
    ) -> Result<()> {
        let user = user.into();
        let mechanism_type = mechanism
            .scram_type()
            .ok_or_else(|| anyhow!("SCRAM credential upsertion requires a SCRAM mechanism"))?;
        let salt = scram::secure_random_bytes()?;
        let salted_password =
            scram::salted_password(mechanism, password.as_ref(), &salt, iterations)?;
        let request = AlterUserScramCredentialsRequest::default().with_upsertions(vec![
            ScramCredentialUpsertion::default()
                .with_name(StrBytes::from_string(user.clone()))
                .with_mechanism(mechanism_type)
                .with_iterations(iterations)
                .with_salt(Bytes::from(salt))
                .with_salted_password(Bytes::from(salted_password)),
        ]);
        let response: AlterUserScramCredentialsResponse = self
            .send_request::<AlterUserScramCredentialsRequest>(
                ALTER_USER_SCRAM_CREDENTIALS_VERSION_CAP,
                &request,
            )
            .await?;

        for result in response.results {
            if let Some(error) = result.error_code.err() {
                return Err(admin_broker_error(
                    "alter_scram_credential",
                    Some(result.user.to_string()),
                    error,
                ));
            }
        }

        Ok(())
    }

    /// Returns the configuration used by this admin client.
    pub fn config(&self) -> &AdminConfig {
        &self.config
    }

    /// Returns finalized broker feature levels visible from a bootstrap broker.
    pub async fn finalized_feature_levels(&self) -> Result<Vec<BrokerFeatureLevel>> {
        let connection = connect_to_any_bootstrap(
            &self.config.bootstrap_servers,
            &self.config.client_id,
            self.config.request_timeout,
            self.config.security_protocol,
            &self.config.tls,
            &self.config.sasl,
            &self.config.tcp_connector,
        )
        .await?;
        Ok(connection
            .finalized_feature_levels()
            .into_iter()
            .map(|(name, level)| BrokerFeatureLevel { name, level })
            .collect())
    }

    /// Updates finalized broker feature levels.
    ///
    /// Feature updates are cluster-level operations. Consider calling
    /// [`KafkaAdmin::validate_feature_updates`] first when supported by the
    /// broker.
    pub async fn update_features<I>(&self, updates: I) -> Result<()>
    where
        I: IntoIterator<Item = FeatureUpdate>,
    {
        self.update_features_inner(updates, false).await
    }

    /// Validates finalized broker feature updates without applying them.
    pub async fn validate_feature_updates<I>(&self, updates: I) -> Result<()>
    where
        I: IntoIterator<Item = FeatureUpdate>,
    {
        self.update_features_inner(updates, true).await
    }

    async fn update_features_inner<I>(&self, updates: I, validate_only: bool) -> Result<()>
    where
        I: IntoIterator<Item = FeatureUpdate>,
    {
        let (mut connection, version) = self
            .connect_with_version::<UpdateFeaturesRequest>(UPDATE_FEATURES_VERSION_CAP)
            .await?;
        let feature_updates = updates
            .into_iter()
            .map(|update| update.into_request_update(version))
            .collect::<Result<Vec<_>>>()?;
        if feature_updates.is_empty() {
            return Ok(());
        }

        let mut request = UpdateFeaturesRequest::default()
            .with_timeout_ms(duration_to_i32_ms(self.config.request_timeout)?)
            .with_feature_updates(feature_updates);
        if version >= 1 {
            request = request.with_validate_only(validate_only);
        } else if validate_only {
            return Err(anyhow!("validate-only feature updates require UpdateFeatures v1+").into());
        }

        let response: UpdateFeaturesResponse = connection
            .send_request::<UpdateFeaturesRequest>(&self.config.client_id, version, &request)
            .await?;
        if let Some(error) = response.error_code.err() {
            return Err(admin_broker_error("update_features", None::<String>, error));
        }

        for result in response.results {
            if let Some(error) = result.error_code.err() {
                let feature = result.feature.to_string();
                return Err(admin_broker_error("update_feature", Some(feature), error));
            }
        }

        Ok(())
    }

    async fn warm_up(&self) -> Result<()> {
        let _ = connect_to_any_bootstrap(
            &self.config.bootstrap_servers,
            &self.config.client_id,
            self.config.request_timeout,
            self.config.security_protocol,
            &self.config.tls,
            &self.config.sasl,
            &self.config.tcp_connector,
        )
        .await?;
        Ok(())
    }

    async fn fetch_metadata(&self, topics: Option<&[String]>) -> Result<MetadataResponse> {
        let (mut connection, version) = self
            .connect_with_version::<MetadataRequest>(METADATA_VERSION_CAP)
            .await?;
        let request = MetadataRequest::default()
            .with_topics(topics.map(|topics| {
                topics
                    .iter()
                    .cloned()
                    .map(StrBytes::from_string)
                    .map(|name| MetadataRequestTopic::default().with_name(Some(name.into())))
                    .collect()
            }))
            .with_allow_auto_topic_creation(false)
            .with_include_cluster_authorized_operations(false)
            .with_include_topic_authorized_operations(false);
        Ok(connection
            .send_request::<MetadataRequest>(&self.config.client_id, version, &request)
            .await?)
    }

    async fn send_request<Req>(&self, version_cap: i16, request: &Req) -> Result<Req::Response>
    where
        Req: Request,
    {
        let (mut connection, version) = self.connect_with_version::<Req>(version_cap).await?;
        Ok(connection
            .send_request::<Req>(&self.config.client_id, version, request)
            .await?)
    }

    async fn connect_with_version<Req>(&self, version_cap: i16) -> Result<(BrokerConnection, i16)>
    where
        Req: Request,
    {
        let connection = connect_to_any_bootstrap(
            &self.config.bootstrap_servers,
            &self.config.client_id,
            self.config.request_timeout,
            self.config.security_protocol,
            &self.config.tls,
            &self.config.sasl,
            &self.config.tcp_connector,
        )
        .await?;
        let version = connection.version_with_cap::<Req>(version_cap)?;
        Ok((connection, version))
    }
}

fn validate_topic_name(topic: String) -> Result<String> {
    let topic = topic.trim();
    if topic.is_empty() {
        return Err(AdminError::EmptyTopicName.into());
    }

    Ok(topic.to_owned())
}

fn admin_broker_error(
    operation: &'static str,
    resource: impl Into<Option<String>>,
    error: ResponseError,
) -> crate::Error {
    BrokerError::response(operation, resource, error).into()
}

async fn describe_log_dirs_for_broker(
    mut config: AdminConfig,
    broker: BrokerDescription,
    request_topics: Option<Vec<DescribableLogDirTopic>>,
) -> Result<BrokerLogDirs> {
    let address = format!("{}:{}", broker.host, broker.port);
    config.bootstrap_servers = vec![address];
    let broker_admin = KafkaAdmin { config };
    let response: DescribeLogDirsResponse = broker_admin
        .send_request::<DescribeLogDirsRequest>(
            DESCRIBE_LOG_DIRS_VERSION_CAP,
            &DescribeLogDirsRequest::default().with_topics(request_topics),
        )
        .await?;
    if let Some(error) = ResponseError::try_from_code(response.error_code) {
        return Err(admin_broker_error(
            "describe_log_dirs",
            Some(format!("broker {}", broker.broker_id)),
            error,
        ));
    }

    let mut log_dirs = Vec::new();
    for result in response.results {
        let replicas = result
            .topics
            .into_iter()
            .flat_map(|topic| {
                let topic_name = topic.name.to_string();
                topic
                    .partitions
                    .into_iter()
                    .map(move |partition| ReplicaLogDirDescription {
                        topic: topic_name.clone(),
                        partition: partition.partition_index,
                        size_bytes: partition.partition_size.max(0),
                        offset_lag: partition.offset_lag,
                        is_future: partition.is_future_key,
                    })
            })
            .collect();
        log_dirs.push(LogDirDescription {
            log_dir: result.log_dir.to_string(),
            error_code: result.error_code,
            total_bytes: (result.total_bytes >= 0).then_some(result.total_bytes),
            usable_bytes: (result.usable_bytes >= 0).then_some(result.usable_bytes),
            replicas,
        });
    }
    log_dirs.sort_by(|left, right| left.log_dir.cmp(&right.log_dir));
    Ok(BrokerLogDirs {
        broker_id: broker.broker_id,
        log_dirs,
    })
}

fn describe_log_dirs_topics(partitions: &[TopicPartition]) -> Vec<DescribableLogDirTopic> {
    let mut topics = BTreeMap::<String, Vec<i32>>::new();
    for partition in partitions {
        topics
            .entry(partition.topic.clone())
            .or_default()
            .push(partition.partition);
    }
    topics
        .into_iter()
        .map(|(topic, mut partitions)| {
            partitions.sort_unstable();
            partitions.dedup();
            DescribableLogDirTopic::default()
                .with_topic(StrBytes::from_string(topic).into())
                .with_partitions(partitions)
        })
        .collect()
}

fn validate_group_id(group_id: String) -> Result<String> {
    let group_id = group_id.trim();
    if group_id.is_empty() {
        return Err(ValidationError::EmptyConsumerGroupId.into());
    }

    Ok(group_id.to_owned())
}

fn describe_group_api_for_listed_type(group_type: Option<&str>) -> DescribeGroupApi {
    match group_type {
        Some(group_type) if group_type.eq_ignore_ascii_case("consumer") => {
            DescribeGroupApi::Consumer
        }
        Some(group_type) if group_type.eq_ignore_ascii_case("share") => DescribeGroupApi::Share,
        _ => DescribeGroupApi::Classic,
    }
}

fn validate_feature_name(feature: String) -> Result<String> {
    let feature = feature.trim();
    if feature.is_empty() {
        return Err(ValidationError::EmptyFeatureName.into());
    }

    Ok(feature.to_owned())
}

fn assignment_partition_count(
    assignment: &kafka_protocol::messages::consumer_group_describe_response::Assignment,
) -> usize {
    assignment
        .topic_partitions
        .iter()
        .map(|topic| topic.partitions.len())
        .sum()
}

fn share_assignment_partition_count(
    assignment: &kafka_protocol::messages::share_group_describe_response::Assignment,
) -> usize {
    assignment
        .topic_partitions
        .iter()
        .map(|topic| topic.partitions.len())
        .sum()
}

fn classic_subscription_topic_count(metadata: &Bytes) -> usize {
    let Some((version, mut body)) = classic_protocol_body(metadata) else {
        return 0;
    };
    ConsumerProtocolSubscription::decode(&mut body, version)
        .map(|subscription| subscription.topics.len())
        .unwrap_or(0)
}

fn classic_assignment_partition_count(assignment: &Bytes) -> usize {
    let Some((version, mut body)) = classic_protocol_body(assignment) else {
        return 0;
    };
    ConsumerProtocolAssignment::decode(&mut body, version)
        .map(|assignment| {
            assignment
                .assigned_partitions
                .iter()
                .map(|topic| topic.partitions.len())
                .sum()
        })
        .unwrap_or(0)
}

fn classic_protocol_body(bytes: &Bytes) -> Option<(i16, Bytes)> {
    let mut body = bytes.clone();
    if body.remaining() < 2 {
        return None;
    }
    let version = body.get_i16();
    Some((version, body))
}

fn is_ignorable_create_topic_error(error: ResponseError) -> bool {
    error == ResponseError::TopicAlreadyExists
}

fn response_supported_config_type(config_type: i8) -> bool {
    config_type >= 0
}

#[cfg(test)]
mod tests {
    use super::*;
    use bytes::{BufMut, BytesMut};
    use kafka_protocol::protocol::Encodable;

    #[test]
    fn new_topic_maps_to_create_topics_request() {
        let topic = NewTopic::new("orders", 3, 2)
            .with_config("cleanup.policy", "compact")
            .into_request_topic()
            .expect("topic should be valid");

        assert_eq!(topic.name.0.to_string(), "orders");
        assert_eq!(topic.num_partitions, 3);
        assert_eq!(topic.replication_factor, 2);
        assert_eq!(topic.configs.len(), 1);
        assert_eq!(topic.configs[0].name.to_string(), "cleanup.policy");
    }

    #[test]
    fn new_topic_rejects_invalid_partition_count() {
        let error = NewTopic::new("orders", 0, 1)
            .into_request_topic()
            .expect_err("invalid topic should fail");
        assert!(
            error
                .to_string()
                .contains("topic partition count must be positive")
        );
    }

    #[test]
    fn new_topic_rejects_empty_names_and_invalid_replication_factor() {
        let error = NewTopic::new("  ", 1, 1).into_request_topic().unwrap_err();
        assert!(matches!(
            error,
            crate::Error::Admin(AdminError::EmptyTopicName)
        ));

        let error = NewTopic::new("orders", 1, 0)
            .into_request_topic()
            .unwrap_err();
        assert!(matches!(
            error,
            crate::Error::Admin(AdminError::InvalidReplicationFactor {
                replication_factor: 0
            })
        ));
    }

    #[test]
    fn new_partitions_maps_assignments_and_rejects_invalid_input() {
        let topic = NewPartitions::increase_to(4)
            .with_assignment([1, 2])
            .with_assignment([2, 3])
            .into_request_topic("orders".to_owned())
            .unwrap();
        assert_eq!(topic.name.to_string(), "orders");
        assert_eq!(topic.count, 4);
        assert_eq!(topic.assignments.unwrap().len(), 2);

        let error = NewPartitions::increase_to(0)
            .into_request_topic("orders".to_owned())
            .unwrap_err();
        assert!(matches!(
            error,
            crate::Error::Admin(AdminError::InvalidPartitionCount { partitions: 0 })
        ));

        let error = NewPartitions::increase_to(1)
            .into_request_topic(" ".to_owned())
            .unwrap_err();
        assert!(matches!(
            error,
            crate::Error::Admin(AdminError::EmptyTopicName)
        ));
    }

    #[test]
    fn config_resource_and_operation_protocol_values_are_stable() {
        assert_eq!(ConfigResourceType::Unknown.as_protocol_value(), 0);
        assert_eq!(ConfigResourceType::Topic.as_protocol_value(), 2);
        assert_eq!(ConfigResourceType::Broker.as_protocol_value(), 4);
        assert_eq!(ConfigResourceType::BrokerLogger.as_protocol_value(), 8);
        assert_eq!(ConfigResourceType::ClientMetrics.as_protocol_value(), 16);
        assert_eq!(ConfigResourceType::Group.as_protocol_value(), 32);
        assert_eq!(
            ConfigResourceType::from_protocol_value(2),
            ConfigResourceType::Topic
        );
        assert_eq!(
            ConfigResourceType::from_protocol_value(99),
            ConfigResourceType::Unknown
        );

        assert_eq!(
            AlterConfigOp::set("cleanup.policy", "compact")
                .op_type
                .as_protocol_value(),
            0
        );
        assert_eq!(
            AlterConfigOp::delete("retention.ms")
                .op_type
                .as_protocol_value(),
            1
        );
        assert_eq!(AlterConfigOpType::Append.as_protocol_value(), 2);
        assert_eq!(AlterConfigOpType::Subtract.as_protocol_value(), 3);
        assert!(response_supported_config_type(0));
        assert!(!response_supported_config_type(-1));
    }

    #[test]
    fn config_resource_config_entry_lookup_returns_named_entry() {
        let mut entries = BTreeMap::new();
        entries.insert(
            "cleanup.policy".to_owned(),
            ConfigEntry {
                name: "cleanup.policy".to_owned(),
                value: Some("compact".to_owned()),
                read_only: false,
                config_source: 1,
                is_sensitive: false,
                config_type: Some(2),
                documentation: None,
            },
        );
        let config = ConfigResourceConfig {
            resource: ConfigResource::topic("orders"),
            entries,
        };

        assert_eq!(
            config.entry("cleanup.policy").unwrap().value.as_deref(),
            Some("compact")
        );
        assert!(config.entry("missing").is_none());
    }

    #[test]
    fn describe_group_api_routing_uses_legacy_api_for_classic_groups() {
        assert_eq!(
            describe_group_api_for_listed_type(None),
            DescribeGroupApi::Classic
        );
        assert_eq!(
            describe_group_api_for_listed_type(Some("classic")),
            DescribeGroupApi::Classic
        );
        assert_eq!(
            describe_group_api_for_listed_type(Some("consumer")),
            DescribeGroupApi::Consumer
        );
        assert_eq!(
            describe_group_api_for_listed_type(Some("share")),
            DescribeGroupApi::Share
        );
    }

    #[test]
    fn acl_binding_creation_validates_and_maps_protocol_values() {
        let creation = AclBinding::new(
            ResourcePattern::new(ResourceType::Topic, "orders", PatternType::Prefixed),
            AccessControlEntry::new(
                "User:alice",
                "*",
                AclOperation::Read,
                AclPermissionType::Allow,
            ),
        )
        .into_creation()
        .unwrap();

        assert_eq!(creation.resource_type, 2);
        assert_eq!(creation.resource_name.to_string(), "orders");
        assert_eq!(creation.resource_pattern_type, 4);
        assert_eq!(creation.principal.to_string(), "User:alice");
        assert_eq!(creation.host.to_string(), "*");
        assert_eq!(creation.operation, 3);
        assert_eq!(creation.permission_type, 3);

        let error = AclBinding::new(
            ResourcePattern::new(ResourceType::Any, "orders", PatternType::Literal),
            AccessControlEntry::new(
                "User:alice",
                "*",
                AclOperation::Read,
                AclPermissionType::Allow,
            ),
        )
        .into_creation()
        .unwrap_err();
        assert!(error.to_string().contains("resource_type must not be ANY"));

        let error = AclBinding::new(
            ResourcePattern::new(ResourceType::Topic, "orders", PatternType::Literal),
            AccessControlEntry::new(
                "User:alice",
                "*",
                AclOperation::Any,
                AclPermissionType::Allow,
            ),
        )
        .into_creation()
        .unwrap_err();
        assert!(error.to_string().contains("operation must not be ANY"));
    }

    #[test]
    fn acl_binding_filters_map_to_describe_and_delete_requests() {
        let filter = AclBindingFilter::new(
            ResourcePatternFilter::new(
                ResourceType::Topic,
                Some("orders".to_owned()),
                PatternType::Literal,
            ),
            AccessControlEntryFilter::new(
                Some("User:alice".to_owned()),
                Some("*".to_owned()),
                AclOperation::Read,
                AclPermissionType::Allow,
            ),
        );

        let describe = filter.to_describe_request();
        assert_eq!(describe.resource_type_filter, 2);
        assert_eq!(
            describe
                .resource_name_filter
                .as_ref()
                .map(ToString::to_string),
            Some("orders".to_owned())
        );
        assert_eq!(describe.pattern_type_filter, 3);
        assert_eq!(
            describe.principal_filter.as_ref().map(ToString::to_string),
            Some("User:alice".to_owned())
        );
        assert_eq!(
            describe.host_filter.as_ref().map(ToString::to_string),
            Some("*".to_owned())
        );
        assert_eq!(describe.operation, 3);
        assert_eq!(describe.permission_type, 3);

        let delete = filter.into_delete_filter();
        assert_eq!(delete.resource_type_filter, 2);
        assert_eq!(
            delete
                .resource_name_filter
                .as_ref()
                .map(ToString::to_string),
            Some("orders".to_owned())
        );
        assert_eq!(delete.pattern_type_filter, 3);
        assert_eq!(
            delete.principal_filter.as_ref().map(ToString::to_string),
            Some("User:alice".to_owned())
        );
        assert_eq!(
            delete.host_filter.as_ref().map(ToString::to_string),
            Some("*".to_owned())
        );
        assert_eq!(delete.operation, 3);
        assert_eq!(delete.permission_type, 3);
    }

    #[test]
    fn classic_group_payload_counts_decode_subscription_and_assignment() {
        let subscription = ConsumerProtocolSubscription::default().with_topics(vec![
            StrBytes::from_static_str("orders"),
            StrBytes::from_static_str("payments"),
        ]);
        let subscription_bytes = encode_classic_protocol(&subscription, 3);
        assert_eq!(classic_subscription_topic_count(&subscription_bytes), 2);

        let assignment = ConsumerProtocolAssignment::default().with_assigned_partitions(vec![
            kafka_protocol::messages::consumer_protocol_assignment::TopicPartition::default()
                .with_topic(StrBytes::from_static_str("orders").into())
                .with_partitions(vec![0, 1]),
            kafka_protocol::messages::consumer_protocol_assignment::TopicPartition::default()
                .with_topic(StrBytes::from_static_str("payments").into())
                .with_partitions(vec![0]),
        ]);
        let assignment_bytes = encode_classic_protocol(&assignment, 3);
        assert_eq!(classic_assignment_partition_count(&assignment_bytes), 3);

        assert_eq!(
            classic_subscription_topic_count(&Bytes::from_static(b"\x00")),
            0
        );
        assert_eq!(
            classic_assignment_partition_count(&Bytes::from_static(b"\x00\x63")),
            0
        );
    }

    #[test]
    fn feature_update_maps_to_modern_update_features_request() {
        let update = FeatureUpdate::upgrade("share.version", 1)
            .into_request_update(2)
            .unwrap();
        assert_eq!(update.feature.to_string(), "share.version");
        assert_eq!(update.max_version_level, 1);
        assert_eq!(update.upgrade_type, 1);
        assert!(!update.allow_downgrade);
    }

    #[test]
    fn feature_update_maps_to_legacy_downgrade_flag() {
        let update = FeatureUpdate::safe_downgrade("metadata.version", 0)
            .into_request_update(0)
            .unwrap();
        assert_eq!(update.feature.to_string(), "metadata.version");
        assert_eq!(update.max_version_level, 0);
        assert_eq!(update.upgrade_type, 1);
        assert!(update.allow_downgrade);
    }

    #[test]
    fn feature_update_rejects_empty_name() {
        let error = FeatureUpdate::upgrade(" ", 1)
            .into_request_update(2)
            .unwrap_err();
        assert!(
            error
                .to_string()
                .contains("feature names must be non-empty")
        );
    }

    #[test]
    fn topic_already_exists_is_ignorable() {
        assert!(is_ignorable_create_topic_error(
            ResponseError::TopicAlreadyExists
        ));
        assert!(!is_ignorable_create_topic_error(
            ResponseError::UnknownTopicOrPartition
        ));
    }

    fn encode_classic_protocol<T>(message: &T, version: i16) -> Bytes
    where
        T: Encodable,
    {
        let mut bytes = BytesMut::new();
        bytes.put_i16(version);
        message.encode(&mut bytes, version).unwrap();
        bytes.freeze()
    }
}