krabka_client_admin/

lib.rs

//! Admin client for Krabka operators and control-plane services.
//!
//! The client targets the active controller and retries selected RPCs on a
//! refreshed controller connection when the broker returns `NOT_CONTROLLER`.
//! It supports plaintext by default and the same client-side TLS/SASL security
//! surface as [`krabka_client_core`] through [`AdminClient::connect_secured`].
//!
//! The client is built on `krabka_client_core::Connection`'s typed
//! `send::<R: ProtocolRequest>`, so request-version negotiation is automatic
//! through the `ApiVersionTable` that connect time populates. The public
//! modules cover topic CRUD, listing and description, partition expansion,
//! config changes and config-resource listing, SCRAM user credentials, ACLs,
//! client quotas, delegation tokens, metadata-quorum voters, and log-dir
//! inspection.

use std::{
    any::Any,
    collections::BTreeMap,
    io,
    net::SocketAddr,
    sync::{Arc, Mutex, PoisonError},
};

#[cfg(test)]
use krabka_client_core::{ClientDnsTimeout, MockBroker};
use krabka_client_core::{
    ClientError, Connection, ConnectionOptions, MetadataRecoveryRebootstrapTrigger,
    MetadataRecoveryStrategy, ProtocolRequest as _, connection_target_host, security,
    telemetry::{
        ClientMetrics, ClientTelemetry, ClientTelemetryConfig, NetworkMetrics, TelemetryClientType,
        TelemetryTiming,
    },
    transport,
};
use krabka_protocol::owned::{
    describe_cluster_response::DescribeClusterResponse, metadata_request, metadata_response,
};
use krabka_units::{Time, convert::TimeExt as _};
use thiserror::Error;
use tokio::{
    sync::RwLock,
    time::{Instant, timeout},
};

pub mod brokers;
pub mod cluster;
mod config;
pub mod config_resources;
pub mod configs;
pub mod consumer_group_listings;
pub mod delegation_tokens;
pub mod elections;
pub mod features;
mod group_coordinators;
pub mod group_deletions;
pub mod group_descriptions;
pub mod groups;
pub mod log_dirs;
pub mod offsets;
mod partition_leaders;
pub mod quorum;
pub mod quotas;
mod retry;
pub mod share_group_offsets;
mod telemetry;
pub mod topic_descriptions;
pub mod topics;
pub mod transactions;

#[cfg(test)]
#[macro_export]
macro_rules! encode_test_struct_field {
    ($encoder:ident, $response:ident, $field:ident, $value:expr, $version:expr, $flexible:expr $(,)?) => {
        $encoder(
            &$response {
                $field: $value,
                ..Default::default()
            },
            $version,
            $flexible,
        )
    };
}

#[cfg(test)]
#[macro_export]
macro_rules! encode_test_struct_fields {
    ($encoder:ident, $response:ident, { $($field:ident: $value:expr),+ $(,)? }, $version:expr, $flexible:expr $(,)?) => {
        $encoder(
            &$response {
                $($field: $value,)+
                ..Default::default()
            },
            $version,
            $flexible,
        )
    };
}
pub mod users;

/// Result of applying a `metadata.version` feature update.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct MetadataVersionUpdate {
    pub level: i16,
}

pub use cluster::{ClusterDescription, ClusterNode, DescribeClusterOptions};
pub use config::{
    AdminClientConfig, DEFAULT_ADMIN_CONNECTIONS_MAX_IDLE, DEFAULT_ADMIN_REQUEST_TIMEOUT,
    DEFAULT_API_TIMEOUT, DEFAULT_ENABLE_METRICS_PUSH, DEFAULT_RETRY_BACKOFF,
    DEFAULT_RETRY_BACKOFF_MAX, DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT,
    DEFAULT_SOCKET_CONNECTION_SETUP_TIMEOUT_MAX,
};
pub use config_resources::{ConfigResource, ConfigResourceType};
pub use configs::{
    AlterConfigOp, AlterConfigOpType, AlterConfigsResults, Config, ConfigEntry, ConfigSource,
    ConfigSynonym, ConfigType, DescribeConfigsOptions, DescribeConfigsResults,
    IncrementalAlterConfigsOptions,
};
pub use consumer_group_listings::{
    ConsumerGroupListing, ListConsumerGroupsOptions, ListConsumerGroupsResult,
};
pub use delegation_tokens::{
    CreateDelegationTokenOptions, DelegationToken, DescribeDelegationTokenOptions,
    ExpireDelegationTokenOptions, RenewDelegationTokenOptions,
};
pub use elections::{ElectionResults, ElectionType};
pub use features::{
    DescribeFeaturesOptions, FeatureMetadata, FeatureRange, FeatureUpdate, UpdateFeaturesOptions,
    UpdateFeaturesResults, UpgradeType,
};
pub use group_deletions::GroupDeletions;
pub use group_descriptions::{
    ClassicGroupDescription, ConsumerGroupDescription, DescribeGroupsOptions, GroupDescriptions,
    MemberDescription, ShareGroupDescription, StreamsGroupDescription,
};
pub use groups::{
    ConsumerGroupOffsetOutcome, MemberToRemove, RemoveMembersOptions, RemovedMember, RemovedMembers,
};
pub use log_dirs::{
    BrokerResult, LogDirInfo, LogDirPartitionInfo, LogDirTopicInfo, ReplicaLogDir,
    ReplicaLogDirInfo, TopicPartitionReplica,
};
pub use offsets::{IsolationLevel, ListedOffset, OffsetSpec};
pub use quorum::{MetadataQuorum, QuorumNode, QuorumReplica, RaftVoterEndpoint};
pub use quotas::{
    ClientQuotaAlteration, ClientQuotaEntity, ClientQuotaFilter, ClientQuotaFilterComponent,
    ClientQuotaMatch, ClientQuotas, ENTITY_CLIENT_ID, ENTITY_IP, ENTITY_USER, QuotaOp,
    UserQuotaConfig, diff_user_quotas,
};
pub use share_group_offsets::{
    ListShareGroupOffsetsSpec, ShareGroupOffsets, SharePartitionOffsetInfo,
};
pub use topic_descriptions::{
    DescribeTopicsOptions, ListTopicsOptions, TopicDescription, TopicDescriptions, TopicListing,
    TopicPartitionInfo,
};
pub use topics::{
    AlterPartitionReassignmentsOptions, CreatePartitionsOp, CreatePartitionsOutcome,
    CreateTopicOutcome, CreateTopicSpec, DeleteRecordsOp, DeleteRecordsOutcome, DeleteTopicOutcome,
    PartitionAssignment, PartitionAssignmentOutcome, TopicMetadata, TopicMetadataEntry,
    TopicMutationOptions, TopicReplicationStatus,
};
pub use transactions::{
    AbortTransactionSpec, ListTransactionsFilter, ProducerStateInfo, TransactionDescription,
    TransactionListing,
};
pub use users::{
    AclEntry, AclEntryFilter, AclOperation, CreateAclOutcome, DEFAULT_SCRAM_ITERATIONS,
    DeleteAclFilterOutcome, MAX_SCRAM_ITERATIONS, MIN_SCRAM_ITERATIONS, PatternType,
    PermissionType, ResourceType, ScramDeletion, ScramIterations, ScramUpsertion, ScramUserOutcome,
    UserScramCredential, UserScramCredentials,
};

/// Test seam for `AdminClient`.
///
/// The operator's reconcile only needs dynamic dispatch through this trait.
/// Production code wraps a concrete `AdminClient`, and tests substitute a
/// fake.
///
/// Methods take `&mut self` because the underlying `AdminClient`'s
/// `NOT_CONTROLLER` retry path reconnects the inner `Connection` in place,
/// which needs unique access.
#[async_trait::async_trait]
pub trait AdminClientLike: Send {
    /// Finalize `metadata.version` at `level` through `UpdateFeatures`.
    ///
    /// # Errors
    /// Returns a transport, protocol, or broker error when Kafka rejects the
    /// requested level.
    async fn update_metadata_version(
        &mut self,
        _level: i16,
        _upgrade_type: UpgradeType,
        _timeout: Time,
    ) -> Result<MetadataVersionUpdate, AdminError> {
        Err(AdminError::Protocol(
            "UpdateFeatures is not implemented by this admin client".into(),
        ))
    }
    /// Return the live `__cluster_metadata` partition-zero quorum view.
    ///
    /// # Errors
    /// Returns a transport, protocol, or Kafka error from `DescribeQuorum`.
    async fn describe_metadata_quorum(&mut self) -> Result<MetadataQuorum, AdminError> {
        Err(AdminError::Protocol(
            "DescribeQuorum is not implemented by this admin client".into(),
        ))
    }
    /// Remove one exact node and directory identity from the metadata quorum,
    /// as Kafka's `removeRaftVoter` operation does.
    ///
    /// `cluster_id` is Kafka's `RemoveRaftVoterOptions.clusterId`, the
    /// cluster ID string as `meta.properties` holds it; `None` sends a null
    /// cluster ID, which the controller does not check.
    ///
    /// `NOT_CONTROLLER` (41) makes the client find the controller again and
    /// resend the request with the backoff until `default.api.timeout.ms`
    /// (60 s), as Kafka's `handleNotControllerError` does.
    ///
    /// # Errors
    /// Returns a transport, protocol, or Kafka error from `RemoveRaftVoter`,
    /// and `REQUEST_TIMED_OUT` (7) at the deadline.
    async fn remove_raft_voter(
        &mut self,
        _cluster_id: Option<&str>,
        _node_id: i32,
        _directory_id: uuid::Uuid,
    ) -> Result<(), AdminError> {
        Err(AdminError::Protocol(
            "RemoveRaftVoter is not implemented by this admin client".into(),
        ))
    }
    /// Metadata for the named topics. Pass an empty slice to fetch all
    /// topics, per Kafka semantics.
    ///
    /// # Errors
    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
    async fn metadata(&mut self, topics: &[&str]) -> Result<TopicMetadata, AdminError>;
    /// Converges every partition of `topic` to `replication_factor` through
    /// `AlterPartitionReassignments`.
    ///
    /// Existing replicas are retained where possible. Added replicas rotate
    /// across the broker list, and decreases preserve the leading replica
    /// order. An existing reassignment is never overwritten.
    ///
    /// # Errors
    /// Returns an error when the factor is invalid for the live broker set,
    /// metadata cannot be read, Kafka rejects a reassignment, or transport I/O
    /// fails.
    async fn reconcile_topic_replication_factor(
        &mut self,
        topic: &str,
        replication_factor: i32,
        timeout: Time,
    ) -> Result<TopicReplicationStatus, AdminError>;
    /// Returns current replica membership for every selected partition.
    ///
    /// Pass an empty topic slice to inspect the whole cluster before broker
    /// evacuation.
    ///
    /// # Errors
    /// Returns a transport or protocol error.
    async fn describe_partition_assignments(
        &mut self,
        _topics: &[&str],
    ) -> Result<Vec<PartitionAssignment>, AdminError> {
        Err(AdminError::Protocol(
            "Describe partition assignments is not implemented by this admin client".into(),
        ))
    }
    /// Lists active partition reassignments and their replica membership.
    ///
    /// An empty filter lists every active reassignment.
    ///
    /// # Errors
    /// Returns a transport, protocol, or top-level broker error.
    async fn list_partition_reassignments(
        &mut self,
        _partitions: &BTreeMap<String, Vec<i32>>,
        _timeout: Time,
    ) -> Result<Vec<PartitionAssignment>, AdminError> {
        Err(AdminError::Protocol(
            "ListPartitionReassignments is not implemented by this admin client".into(),
        ))
    }
    /// Submits exact replica lists; `None` cancels an active reassignment.
    /// A reassignment may change a partition's replication factor, Kafka's
    /// default; [`AdminClient::alter_partition_assignments_with`] can refuse that.
    ///
    /// # Errors
    /// Returns a transport, protocol, or top-level broker error.
    async fn alter_partition_assignments(
        &mut self,
        _assignments: &BTreeMap<(String, i32), Option<Vec<i32>>>,
        _timeout: Time,
    ) -> Result<Vec<PartitionAssignmentOutcome>, AdminError> {
        Err(AdminError::Protocol(
            "AlterPartitionReassignments is not implemented by this admin client".into(),
        ))
    }
    /// Unregisters one broker from the active `KRaft` controller.
    ///
    /// Kafka's `KafkaAdminClient.unregisterBroker` retries
    /// `REQUEST_TIMED_OUT` (7) with the retry backoff until the call deadline
    /// (`default.api.timeout.ms`, 60 s), and fails at once on every other
    /// code, `NOT_CONTROLLER` (41) too. This call does the same.
    ///
    /// # Errors
    /// Returns a transport, protocol, or broker error. At the deadline it
    /// gives `REQUEST_TIMED_OUT` (7).
    async fn unregister_broker(&mut self, _broker_id: i32) -> Result<(), AdminError> {
        Err(AdminError::Protocol(
            "UnregisterBroker is not implemented by this admin client".into(),
        ))
    }
    /// Creates topics through the active controller, as Kafka's
    /// `KafkaAdminClient.createTopics` does.
    ///
    /// The call retries until the deadline of `options`:
    ///
    /// - `NOT_CONTROLLER` (41) for any topic: find the controller again, wait
    ///   for the backoff, and send every topic that has no outcome again
    ///   (`handleNotControllerError`). With controller bootstrap,
    ///   `NOT_LEADER_OR_FOLLOWER` (6) does the same.
    /// - `THROTTLING_QUOTA_EXCEEDED` (89): when
    ///   [`TopicMutationOptions::retry_on_quota_violation`] is set, wait for the
    ///   throttle time of the response and send only the throttled topics
    ///   again. A topic that has an outcome is not sent again.
    ///
    /// At the deadline a throttled topic keeps code 89, with the throttle time
    /// that remains. Another topic without an outcome gets
    /// `REQUEST_TIMED_OUT` (7). The outcomes keep the order of `specs`.
    ///
    /// # Errors
    /// Returns an error when configuration is invalid, protocol encoding fails, or transport I/O fails.
    async fn create_topics(
        &mut self,
        specs: &[CreateTopicSpec],
        options: TopicMutationOptions,
    ) -> Result<Vec<CreateTopicOutcome>, AdminError>;
    /// Deletes topics through the active controller, as Kafka's
    /// `KafkaAdminClient.deleteTopics` does. The call retries
    /// `NOT_CONTROLLER` (41) and `THROTTLING_QUOTA_EXCEEDED` (89) as
    /// [`AdminClientLike::create_topics`] does.
    ///
    /// # Errors
    /// Returns an error when configuration is invalid, protocol encoding fails, or transport I/O fails.
    async fn delete_topics(
        &mut self,
        names: &[&str],
        options: TopicMutationOptions,
    ) -> Result<Vec<DeleteTopicOutcome>, AdminError>;
    /// Adds partitions through the active controller, as Kafka's
    /// `KafkaAdminClient.createPartitions` does. The call retries
    /// `NOT_CONTROLLER` (41) and `THROTTLING_QUOTA_EXCEEDED` (89) as
    /// [`AdminClientLike::create_topics`] does.
    ///
    /// # Errors
    /// Returns an error when configuration is invalid, protocol encoding fails, or transport I/O fails.
    async fn create_partitions(
        &mut self,
        ops: &[CreatePartitionsOp],
        options: TopicMutationOptions,
    ) -> Result<Vec<CreatePartitionsOutcome>, AdminError>;
    /// Deletes records below each requested partition offset.
    ///
    /// `offset == -1` follows Kafka `DeleteRecords` semantics and truncates to
    /// the partition high watermark. The returned `low_watermark` is the
    /// broker's resulting log-start offset for that partition.
    ///
    /// # Errors
    ///
    /// Returns an [`AdminError`] when metadata lookup, leader routing, transport,
    /// or protocol handling fails. Kafka partition-level failures remain in the
    /// returned [`DeleteRecordsOutcome::error_code`].
    async fn delete_records(
        &mut self,
        ops: &[DeleteRecordsOp],
        timeout: Time,
    ) -> Result<Vec<DeleteRecordsOutcome>, AdminError>;
    /// Describes the configs of each resource of `resources`, as Kafka's
    /// `describeConfigs` operation does.
    ///
    /// The client sends one `DescribeConfigs` request to each broker that a
    /// `BROKER` resource with a broker id or a `BROKER_LOGGER` resource
    /// names, and one request with every other resource on its connection,
    /// all at the same time. Each request asks for every config of its
    /// resources. The result has one entry for each resource:
    ///
    /// - Its [`Config`], or the error the broker gave for it, such as
    ///   `UNKNOWN_TOPIC_OR_PARTITION` (3).
    /// - `UNKNOWN_SERVER_ERROR` (-1) when the answer has no result for it.
    /// - The error of its request when that request fails: `REQUEST_TIMED_OUT`
    ///   (7) at `default.api.timeout.ms` (60 s), or `UNSUPPORTED_VERSION` (35)
    ///   when the node does not support `DescribeConfigs` v3 and
    ///   `include_documentation` is set. A broker is found by id in fresh
    ///   metadata, and a missing broker or a lost connection is tried again
    ///   until the deadline, as in [`AdminClient::describe_log_dirs`].
    ///
    /// # Errors
    /// Returns [`AdminError::InvalidArgument`], and sends nothing, when the
    /// name of a `BROKER` resource is neither empty nor a broker id, or the
    /// name of a `BROKER_LOGGER` resource is not a broker id, as Kafka's
    /// `nodeFor` throws a `NumberFormatException`.
    async fn describe_configs(
        &mut self,
        resources: &[ConfigResource],
        options: DescribeConfigsOptions,
    ) -> Result<DescribeConfigsResults, AdminError>;
    /// Changes the configs of each resource of `configs`, as Kafka's
    /// `incrementalAlterConfigs` operation does. With
    /// [`IncrementalAlterConfigsOptions::validate_only`] the broker
    /// validates the changes and applies none.
    ///
    /// A `BROKER` resource with a broker id goes to that broker, and a
    /// `BROKER_LOGGER` resource to its broker, each in a request of its own.
    /// With a controller bootstrap only a `BROKER_LOGGER` resource goes to
    /// its node. Every other resource goes in one request on the client's
    /// connection. A `NOT_CONTROLLER` (41) answer, or with a controller
    /// bootstrap a `NOT_LEADER_OR_FOLLOWER` (6) answer, makes the client find
    /// the controller again and send the request again until the deadline.
    /// The result has one entry for each resource:
    ///
    /// - `Ok(())`, or the error the broker gave for it, such as
    ///   `INVALID_CONFIG` (40).
    /// - `UNKNOWN_SERVER_ERROR` (-1) when the answer has no result for it.
    /// - The error of its request when that request fails, with
    ///   `REQUEST_TIMED_OUT` (7) at `default.api.timeout.ms` (60 s).
    ///
    /// # Errors
    /// Returns [`AdminError::InvalidArgument`], and sends nothing, for a
    /// resource name that [`Self::describe_configs`] refuses.
    async fn incremental_alter_configs(
        &mut self,
        configs: &BTreeMap<ConfigResource, Vec<AlterConfigOp>>,
        options: IncrementalAlterConfigsOptions,
    ) -> Result<AlterConfigsResults, AdminError>;
    /// Upserts SCRAM-SHA-512 credentials, deletes them, or both, in a single
    /// call.
    ///
    /// `upsertions` carry plaintext passwords. The function generates a fresh
    /// 16-byte salt per row and computes the KIP-554 wire `salted_password`
    /// (PBKDF2-HMAC-SHA-512) client-side with
    /// `krabka_security::pbkdf2_salted`. The broker never sees the raw
    /// password.
    ///
    /// # Errors
    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
    async fn alter_user_scram_credentials_sha512(
        &mut self,
        upsertions: &[ScramUpsertion],
        deletions: &[ScramDeletion],
    ) -> Result<Vec<ScramUserOutcome>, AdminError>;
    /// SCRAM-SHA-256 sibling of
    /// [`Self::alter_user_scram_credentials_sha512`].
    ///
    /// The iteration counts, the salt generation, and the salted-password
    /// derivation are identical to the SHA-512 path. Only the mechanism wire
    /// byte and the HMAC algorithm differ.
    ///
    /// # Errors
    ///
    /// Same as `_sha512`: returns [`AdminError::Protocol`] when the
    /// system RNG fails, otherwise propagates the broker's
    /// per-username outcome rows.
    async fn alter_user_scram_credentials_sha256(
        &mut self,
        upsertions: &[ScramUpsertion],
        deletions: &[ScramDeletion],
    ) -> Result<Vec<ScramUserOutcome>, AdminError>;
    /// Lists ACLs that match `filter`.
    ///
    /// The broker's response is resource-grouped on the wire, with one block
    /// per `(resource_type, resource_name, pattern_type)`. This method
    /// flattens it back into `AclEntry` rows for the diff.
    ///
    /// # Errors
    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
    async fn describe_acls(&mut self, filter: &AclEntryFilter)
    -> Result<Vec<AclEntry>, AdminError>;
    /// Creates the supplied ACLs.
    ///
    /// # Errors
    /// Returns [`AdminError::InvalidArgument`], before anything is sent, for
    /// an entry whose pattern type is [`PatternType::Match`]: Kafka's
    /// `ResourcePattern` refuses `patternType must not be MATCH`. Otherwise
    /// returns an error when protocol encoding fails, the broker rejects the
    /// request, or transport I/O fails.
    async fn create_acls(
        &mut self,
        creations: &[AclEntry],
    ) -> Result<Vec<CreateAclOutcome>, AdminError>;
    /// Deletes every ACL that matches any of `filters`.
    ///
    /// Each filter's response surfaces the matched ACL set, so callers can
    /// confirm that the deletion converged on the expected rows.
    ///
    /// # Errors
    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
    async fn delete_acls(
        &mut self,
        filters: &[AclEntryFilter],
    ) -> Result<Vec<DeleteAclFilterOutcome>, AdminError>;
    /// Reads the quotas of the user entity `{user: username}`.
    ///
    /// The filter is strict on the single component `("user", username)`, as
    /// `ClientQuotaFilter.containsOnly`: an entry whose entity also has a
    /// `client-id` does not match.
    ///
    /// # Errors
    /// See [`AdminClient::describe_client_quotas`].
    async fn describe_user_quotas(&mut self, username: &str)
    -> Result<UserQuotaConfig, AdminError>;
    /// Applies `ops` to the user entity `{user: username}`. Returns the
    /// error the broker gave for the entity, or `None` on success. No
    /// request is sent for an empty `ops`.
    ///
    /// # Errors
    /// See [`AdminClient::alter_client_quotas`].
    async fn alter_user_quotas(
        &mut self,
        username: &str,
        ops: &[QuotaOp],
        validate_only: bool,
    ) -> Result<Option<KafkaError>, AdminError>;

    // ── delegation-token RPCs (KIP-48) ────────────────────────────────
    //
    // Each method runs the `_impl` call of the same name in
    // `delegation_tokens.rs`.

    /// Creates a delegation token, as Kafka's `createDelegationToken`
    /// operation does.
    ///
    /// The broker does not return the renewers, so the result has the
    /// renewers of `options`, as Kafka's result does. With an
    /// [`owner`](CreateDelegationTokenOptions::owner) the client negotiates
    /// `CreateDelegationToken` v3 or higher, as Kafka's request refuses to
    /// write the owner fields below v3.
    ///
    /// # Errors
    ///
    /// Returns [`AdminError::Broker`] for a non-zero error code, such as
    /// `DELEGATION_TOKEN_AUTH_DISABLED` (61),
    /// `DELEGATION_TOKEN_REQUEST_NOT_ALLOWED` (64),
    /// `DELEGATION_TOKEN_AUTHORIZATION_FAILED` (65), or
    /// `INVALID_PRINCIPAL_TYPE` (67). Returns [`AdminError::Transport`] when
    /// the connection fails, or when an owner is set and the broker does not
    /// support v3.
    async fn create_delegation_token(
        &mut self,
        options: &CreateDelegationTokenOptions,
    ) -> Result<DelegationToken, AdminError>;
    /// Renews the delegation token with `hmac`, as Kafka's
    /// `renewDelegationToken` operation does, and returns its new expiry in
    /// epoch milliseconds.
    ///
    /// # Errors
    ///
    /// Returns [`AdminError::Broker`] for a non-zero error code, such as
    /// `DELEGATION_TOKEN_NOT_FOUND` (62), `DELEGATION_TOKEN_OWNER_MISMATCH`
    /// (63), or `DELEGATION_TOKEN_EXPIRED` (66). Returns
    /// [`AdminError::Transport`] when the connection fails.
    async fn renew_delegation_token(
        &mut self,
        hmac: &[u8],
        options: RenewDelegationTokenOptions,
    ) -> Result<i64, AdminError>;
    /// Expires the delegation token with `hmac`, as Kafka's
    /// `expireDelegationToken` operation does, and returns the expiry that
    /// the broker set, in epoch milliseconds. A token that expires now gets
    /// the broker's clock at the time of the request.
    ///
    /// # Errors
    ///
    /// Returns [`AdminError::Broker`] for a non-zero error code, such as
    /// `DELEGATION_TOKEN_NOT_FOUND` (62), `DELEGATION_TOKEN_OWNER_MISMATCH`
    /// (63), or `DELEGATION_TOKEN_EXPIRED` (66). Returns
    /// [`AdminError::Transport`] when the connection fails.
    async fn expire_delegation_token(
        &mut self,
        hmac: &[u8],
        options: ExpireDelegationTokenOptions,
    ) -> Result<i64, AdminError>;
    /// Describes the delegation tokens that `options` selects, as Kafka's
    /// `describeDelegationToken` operation does.
    ///
    /// # Errors
    ///
    /// Returns [`AdminError::Broker`] for a non-zero error code, such as
    /// `DELEGATION_TOKEN_AUTH_DISABLED` (61) or
    /// `DELEGATION_TOKEN_REQUEST_NOT_ALLOWED` (64). Returns
    /// [`AdminError::Transport`] when the connection fails.
    async fn describe_delegation_token(
        &mut self,
        options: &DescribeDelegationTokenOptions,
    ) -> Result<Vec<DelegationToken>, AdminError>;
}

#[async_trait::async_trait]
impl AdminClientLike for AdminClient {
    async fn update_metadata_version(
        &mut self,
        level: i16,
        upgrade_type: UpgradeType,
        timeout: Time,
    ) -> Result<MetadataVersionUpdate, AdminError> {
        Self::update_metadata_version_impl(self, level, upgrade_type, timeout).await
    }

    async fn describe_metadata_quorum(&mut self) -> Result<MetadataQuorum, AdminError> {
        Self::describe_metadata_quorum_impl(self).await
    }

    async fn remove_raft_voter(
        &mut self,
        cluster_id: Option<&str>,
        node_id: i32,
        directory_id: uuid::Uuid,
    ) -> Result<(), AdminError> {
        Self::remove_raft_voter_impl(self, cluster_id, node_id, directory_id).await
    }

    async fn metadata(&mut self, topics: &[&str]) -> Result<TopicMetadata, AdminError> {
        Self::metadata_impl(self, topics).await
    }
    async fn reconcile_topic_replication_factor(
        &mut self,
        topic: &str,
        replication_factor: i32,
        timeout: Time,
    ) -> Result<TopicReplicationStatus, AdminError> {
        Self::reconcile_topic_replication_factor_impl(self, topic, replication_factor, timeout)
            .await
    }
    async fn describe_partition_assignments(
        &mut self,
        topics: &[&str],
    ) -> Result<Vec<PartitionAssignment>, AdminError> {
        Self::describe_partition_assignments_impl(self, topics).await
    }
    async fn list_partition_reassignments(
        &mut self,
        partitions: &BTreeMap<String, Vec<i32>>,
        timeout: Time,
    ) -> Result<Vec<PartitionAssignment>, AdminError> {
        Self::list_partition_reassignments_impl(self, partitions, timeout).await
    }
    async fn alter_partition_assignments(
        &mut self,
        assignments: &BTreeMap<(String, i32), Option<Vec<i32>>>,
        timeout: Time,
    ) -> Result<Vec<PartitionAssignmentOutcome>, AdminError> {
        Self::alter_partition_assignments_impl(self, assignments, timeout).await
    }
    async fn unregister_broker(&mut self, broker_id: i32) -> Result<(), AdminError> {
        Self::unregister_broker_impl(self, broker_id).await
    }
    async fn create_topics(
        &mut self,
        specs: &[CreateTopicSpec],
        options: TopicMutationOptions,
    ) -> Result<Vec<CreateTopicOutcome>, AdminError> {
        Self::create_topics_impl(self, specs, options).await
    }
    async fn delete_topics(
        &mut self,
        names: &[&str],
        options: TopicMutationOptions,
    ) -> Result<Vec<DeleteTopicOutcome>, AdminError> {
        Self::delete_topics_impl(self, names, options).await
    }
    async fn create_partitions(
        &mut self,
        ops: &[CreatePartitionsOp],
        options: TopicMutationOptions,
    ) -> Result<Vec<CreatePartitionsOutcome>, AdminError> {
        Self::create_partitions_impl(self, ops, options).await
    }
    async fn delete_records(
        &mut self,
        ops: &[DeleteRecordsOp],
        timeout: Time,
    ) -> Result<Vec<DeleteRecordsOutcome>, AdminError> {
        Self::delete_records_impl(self, ops, timeout).await
    }
    async fn describe_configs(
        &mut self,
        resources: &[ConfigResource],
        options: DescribeConfigsOptions,
    ) -> Result<DescribeConfigsResults, AdminError> {
        Self::describe_configs_impl(self, resources, options).await
    }
    async fn incremental_alter_configs(
        &mut self,
        configs: &BTreeMap<ConfigResource, Vec<AlterConfigOp>>,
        options: IncrementalAlterConfigsOptions,
    ) -> Result<AlterConfigsResults, AdminError> {
        Self::incremental_alter_configs_impl(self, configs, options).await
    }
    async fn alter_user_scram_credentials_sha512(
        &mut self,
        upsertions: &[ScramUpsertion],
        deletions: &[ScramDeletion],
    ) -> Result<Vec<ScramUserOutcome>, AdminError> {
        Self::alter_user_scram_credentials_sha512_impl(self, upsertions, deletions).await
    }
    async fn alter_user_scram_credentials_sha256(
        &mut self,
        upsertions: &[ScramUpsertion],
        deletions: &[ScramDeletion],
    ) -> Result<Vec<ScramUserOutcome>, AdminError> {
        Self::alter_user_scram_credentials_sha256_impl(self, upsertions, deletions).await
    }
    async fn describe_acls(
        &mut self,
        filter: &AclEntryFilter,
    ) -> Result<Vec<AclEntry>, AdminError> {
        Self::describe_acls_impl(self, filter).await
    }
    async fn create_acls(
        &mut self,
        creations: &[AclEntry],
    ) -> Result<Vec<CreateAclOutcome>, AdminError> {
        Self::create_acls_impl(self, creations).await
    }
    async fn delete_acls(
        &mut self,
        filters: &[AclEntryFilter],
    ) -> Result<Vec<DeleteAclFilterOutcome>, AdminError> {
        Self::delete_acls_impl(self, filters).await
    }
    async fn describe_user_quotas(
        &mut self,
        username: &str,
    ) -> Result<UserQuotaConfig, AdminError> {
        Self::describe_user_quotas_impl(self, username).await
    }
    async fn alter_user_quotas(
        &mut self,
        username: &str,
        ops: &[QuotaOp],
        validate_only: bool,
    ) -> Result<Option<KafkaError>, AdminError> {
        Self::alter_user_quotas_impl(self, username, ops, validate_only).await
    }

    async fn create_delegation_token(
        &mut self,
        options: &CreateDelegationTokenOptions,
    ) -> Result<DelegationToken, AdminError> {
        Self::create_delegation_token_impl(self, options).await
    }
    async fn renew_delegation_token(
        &mut self,
        hmac: &[u8],
        options: RenewDelegationTokenOptions,
    ) -> Result<i64, AdminError> {
        Self::renew_delegation_token_impl(self, hmac, options).await
    }
    async fn expire_delegation_token(
        &mut self,
        hmac: &[u8],
        options: ExpireDelegationTokenOptions,
    ) -> Result<i64, AdminError> {
        Self::expire_delegation_token_impl(self, hmac, options).await
    }
    async fn describe_delegation_token(
        &mut self,
        options: &DescribeDelegationTokenOptions,
    ) -> Result<Vec<DelegationToken>, AdminError> {
        Self::describe_delegation_token_impl(self, options).await
    }
}

#[derive(Debug, Error)]
pub enum AdminError {
    /// No bootstrap address gave a usable connection. `source` holds the
    /// error from the last address tried, and is `None` only when the list
    /// was empty. Match it to tell an unreachable or silent broker
    /// ([`ClientError::Connect`] or [`ClientError::Timeout`]) from a TLS or
    /// SASL failure with no verdict from the peer ([`ClientError::Tls`] or
    /// [`ClientError::Sasl`]).
    ///
    /// A rejected authentication does not give this error. The client stops
    /// at the first address that rejects it and returns
    /// `AdminError::Transport(ClientError::Authentication { .. })`, as Kafka's
    /// `AdminMetadataManager` makes an `AuthenticationException` fatal.
    #[error("no bootstrap address connected: tried {tried}{cause}",
            cause = .source.as_ref().map(|e| format!("; last error: {e}")).unwrap_or_default())]
    Connect {
        tried: usize,
        #[source]
        source: Option<Box<Self>>,
    },
    /// A setting of [`AdminClientConfig`] is invalid, as Kafka raises a
    /// `ConfigException` when it builds the admin client.
    #[error("invalid admin client configuration: {0}")]
    InvalidConfig(String),
    /// An argument of a call is invalid, as Kafka's admin types throw an
    /// `IllegalArgumentException` before anything is sent.
    #[error("invalid argument: {0}")]
    InvalidArgument(String),
    #[error("broker returned error: api={api} code={code} ({name}){detail}",
            detail = .message.as_deref().map(|m| format!(" {m:?}")).unwrap_or_default())]
    Broker {
        api: &'static str,
        code: i16,
        name: &'static str,
        message: Option<String>,
    },
    #[error("client-core: {0}")]
    Transport(#[from] ClientError),
    #[error("protocol: {0}")]
    Protocol(String),
}

impl AdminError {
    /// Whether a peer rejected authentication. Kafka's admin client fails the
    /// call with the `AuthenticationException` and does not retry
    /// (`KafkaAdminClient` `handleResponses`, `AdminMetadataManager.updateFailed`).
    #[must_use]
    pub const fn is_authentication_failure(&self) -> bool {
        matches!(self, Self::Transport(error) if error.is_authentication_failure())
    }
}

/// A Kafka-level error attached to a single per-resource outcome.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct KafkaError {
    pub code: i16,
    pub name: &'static str,
    pub message: Option<String>,
}

pub(crate) fn kafka_error_if(code: i16, message: Option<String>) -> Option<KafkaError> {
    if code == 0 {
        None
    } else {
        Some(KafkaError {
            code,
            name: kafka_error_name(code),
            message,
        })
    }
}

async fn lookup_first<F, I>(
    host_port: &str,
    dns_timeout: krabka_client_core::ClientDnsTimeout,
    lookup: F,
) -> Result<SocketAddr, AdminError>
where
    F: Future<Output = io::Result<I>>,
    I: IntoIterator<Item = SocketAddr>,
{
    let addrs = timeout(dns_timeout.time().to_std(), lookup)
        .await
        .map_err(|_| {
            AdminError::Protocol(format!(
                "DNS lookup {host_port} timed out after {} ms",
                dns_timeout.milliseconds(),
            ))
        })?
        .map_err(|error| AdminError::Protocol(format!("DNS lookup {host_port}: {error}")))?;
    addrs
        .into_iter()
        .next()
        .ok_or_else(|| AdminError::Protocol(format!("no addresses for {host_port}")))
}

fn format_host_port(host: &str, port: i32) -> String {
    if host.contains(':') && !host.starts_with('[') {
        format!("[{host}]:{port}")
    } else {
        format!("{host}:{port}")
    }
}

/// Short-lived admin client that targets one cluster's controller. It can
/// negotiate TLS/SASL through [`AdminClient::connect_secured`].
///
/// As Kafka's admin client with `enable.metrics.push` on (the default), it
/// pushes the client metrics that a broker's client metrics subscription
/// names (KIP-714) on its bootstrap connection, and its connections count in
/// the `admin-client-metrics` network metrics.
/// [`AdminClientConfig::enable_metrics_push`] turns this off. The
/// constructors without an [`AdminClientConfig`] keep Kafka's default.
/// [`close`](Self::close) sends the terminating push and waits for it;
/// dropping the client sends it in the background.
pub struct AdminClient {
    pub(crate) conn: Arc<RecoveringConnection>,
    bootstrap_addrs: Vec<String>,
    /// Full connection template carried forward so reconnects preserve
    /// caller-supplied identity, security, and timeouts. With metrics push
    /// on, it carries the network metrics that each new connection counts in.
    options: ConnectionOptions,
    /// The deadline, backoff and retry count of each call
    /// (`default.api.timeout.ms`, `retry.backoff.ms`, `retry.backoff.max.ms`,
    /// `retries`).
    pub(crate) retry: retry::RetryPolicy,
    /// The KIP-714 reporter, `None` with `enable_metrics_push` off. Dropping
    /// it starts the terminating push.
    telemetry: Option<ClientTelemetry>,
}

#[derive(Clone, Copy, Debug, Eq, PartialEq)]
enum BootstrapTarget {
    Brokers,
    Controllers,
}

struct ConnectedTarget {
    connection: Connection,
    discovered_addrs: Vec<String>,
}

/// One controller connection with KIP-919/KIP-1102 bootstrap recovery.
pub(crate) struct RecoveringConnection {
    inner: RwLock<Arc<Connection>>,
    bootstrap_addrs: Vec<String>,
    options: ConnectionOptions,
    strategy: MetadataRecoveryStrategy,
    trigger: MetadataRecoveryRebootstrapTrigger,
    target: BootstrapTarget,
    known_addrs: Mutex<Vec<String>>,
    first_metadata_attempt: Mutex<Option<Instant>>,
}

impl RecoveringConnection {
    fn new(
        connection: Connection,
        bootstrap_addrs: Vec<String>,
        options: ConnectionOptions,
        strategy: MetadataRecoveryStrategy,
        trigger: MetadataRecoveryRebootstrapTrigger,
        target: BootstrapTarget,
        known_addrs: Vec<String>,
    ) -> Self {
        Self {
            inner: RwLock::new(Arc::new(connection)),
            bootstrap_addrs,
            options,
            strategy,
            trigger,
            target,
            known_addrs: Mutex::new(known_addrs),
            first_metadata_attempt: Mutex::new(None),
        }
    }

    pub(crate) async fn send<R>(&self, request: R) -> Result<R::Response, AdminError>
    where
        R: krabka_protocol::ProtocolRequest + Clone,
        R::Response: 'static,
    {
        self.send_with_min_version(request, None).await
    }

    pub(crate) async fn send_at_least<R>(
        &self,
        request: R,
        min_version: i16,
    ) -> Result<R::Response, AdminError>
    where
        R: krabka_protocol::ProtocolRequest + Clone,
        R::Response: 'static,
    {
        self.send_with_min_version(request, Some(min_version)).await
    }

    async fn send_with_min_version<R>(
        &self,
        request: R,
        min_version: Option<i16>,
    ) -> Result<R::Response, AdminError>
    where
        R: krabka_protocol::ProtocolRequest + Clone,
        R::Response: 'static,
    {
        self.require_advertised_controller_api::<R>().await?;
        self.begin_metadata_attempt::<R>();
        let first = self.send_current(request.clone(), min_version).await;
        match first {
            Ok(response) if Self::response_requires_rebootstrap(&response) => {
                if self.strategy == MetadataRecoveryStrategy::None {
                    return Err(AdminError::Transport(ClientError::Server {
                        error_code: 129,
                    }));
                }
                self.send_after_rebootstrap(request, min_version).await
            }
            Ok(response) if self.metadata_attempt_timed_out(&response) => {
                self.send_after_rebootstrap(request, min_version).await
            }
            Ok(response) => {
                self.observe_cluster_endpoints(&response);
                self.complete_metadata_attempt(&response);
                Ok(response)
            }
            Err(error) if AdminClient::is_retriable_transport_error(&error) => {
                match self
                    .send_after_current_metadata(request.clone(), min_version)
                    .await
                {
                    Ok(response) => Ok(response),
                    Err(retry_error) if retry_error.is_authentication_failure() => Err(retry_error),
                    Err(_) if self.strategy == MetadataRecoveryStrategy::Rebootstrap => {
                        self.send_after_rebootstrap(request, min_version).await
                    }
                    Err(_) => Err(AdminError::from(error)),
                }
            }
            Err(error) => Err(AdminError::from(error)),
        }
    }

    async fn replace(&self, connection: Connection) {
        *self.inner.write().await = Arc::new(connection);
    }

    async fn send_current<R>(
        &self,
        request: R,
        min_version: Option<i16>,
    ) -> Result<R::Response, ClientError>
    where
        R: krabka_protocol::ProtocolRequest,
    {
        let connection = self.inner.read().await;
        if let Some(min_version) = min_version {
            return send_connection_at_least(&connection, request, min_version).await;
        }
        connection.send(request).await
    }

    async fn send_after_rebootstrap<R>(
        &self,
        request: R,
        min_version: Option<i16>,
    ) -> Result<R::Response, AdminError>
    where
        R: krabka_protocol::ProtocolRequest,
        R::Response: 'static,
    {
        self.rebootstrap().await?;
        self.require_advertised_controller_api::<R>().await?;
        self.begin_metadata_attempt::<R>();
        let response = self.send_current(request, min_version).await?;
        if Self::response_requires_rebootstrap(&response) {
            return Err(AdminError::Transport(ClientError::Server {
                error_code: 129,
            }));
        }
        self.observe_cluster_endpoints(&response);
        self.complete_metadata_attempt(&response);
        Ok(response)
    }

    async fn send_after_current_metadata<R>(
        &self,
        request: R,
        min_version: Option<i16>,
    ) -> Result<R::Response, AdminError>
    where
        R: krabka_protocol::ProtocolRequest + Clone,
        R::Response: 'static,
    {
        self.reconnect_current_metadata().await?;
        self.require_advertised_controller_api::<R>().await?;
        let response = self.send_current(request.clone(), min_version).await?;
        if Self::response_requires_rebootstrap(&response) {
            if self.strategy == MetadataRecoveryStrategy::Rebootstrap {
                return self.send_after_rebootstrap(request, min_version).await;
            }
            return Err(AdminError::Transport(ClientError::Server {
                error_code: 129,
            }));
        }
        if self.metadata_attempt_timed_out(&response) {
            return self.send_after_rebootstrap(request, min_version).await;
        }
        self.observe_cluster_endpoints(&response);
        self.complete_metadata_attempt(&response);
        Ok(response)
    }

    async fn require_advertised_controller_api<R>(&self) -> Result<(), AdminError>
    where
        R: krabka_protocol::ProtocolRequest,
    {
        const UNSUPPORTED_ENDPOINT_TYPE: i16 = 115;
        if self.target == BootstrapTarget::Controllers
            && self
                .inner
                .read()
                .await
                .advertised_api_range(R::API_KEY)
                .is_none()
        {
            return Err(AdminError::Broker {
                api: "ControllerEndpoint",
                code: UNSUPPORTED_ENDPOINT_TYPE,
                name: kafka_error_name(UNSUPPORTED_ENDPOINT_TYPE),
                message: Some(format!(
                    "api_key {} is not supported by the controller listener",
                    R::API_KEY
                )),
            });
        }
        Ok(())
    }

    pub(crate) async fn rebootstrap(&self) -> Result<(), AdminError> {
        let mut last_error = None;
        for host_port in &self.bootstrap_addrs {
            match AdminClient::connect_target_one_with_discovery(
                host_port,
                self.options.clone(),
                self.target,
            )
            .await
            {
                Ok(connected) => {
                    self.replace(connected.connection).await;
                    self.replace_known_addrs(if connected.discovered_addrs.is_empty() {
                        self.bootstrap_addrs.clone()
                    } else {
                        connected.discovered_addrs
                    });
                    self.clear_metadata_attempt();
                    return Ok(());
                }
                Err(error) if error.is_authentication_failure() => return Err(error),
                Err(error) => last_error = Some(Box::new(error)),
            }
        }
        Err(AdminError::Connect {
            tried: self.bootstrap_addrs.len(),
            source: last_error,
        })
    }

    async fn reconnect_current_metadata(&self) -> Result<(), AdminError> {
        let known_addrs = self
            .known_addrs
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .clone();
        let mut last_error = None;
        for host_port in &known_addrs {
            match AdminClient::connect_target_one_with_discovery(
                host_port,
                self.options.clone(),
                self.target,
            )
            .await
            {
                Ok(connected) => {
                    self.replace(connected.connection).await;
                    if !connected.discovered_addrs.is_empty() {
                        self.replace_known_addrs(connected.discovered_addrs);
                    }
                    return Ok(());
                }
                Err(error) if error.is_authentication_failure() => return Err(error),
                Err(error) => last_error = Some(Box::new(error)),
            }
        }
        Err(AdminError::Connect {
            tried: known_addrs.len(),
            source: last_error,
        })
    }

    fn replace_known_addrs(&self, addrs: Vec<String>) {
        *self
            .known_addrs
            .lock()
            .unwrap_or_else(PoisonError::into_inner) = addrs;
    }

    fn observe_cluster_endpoints<T: Any>(&self, response: &T) {
        if let Some(metadata) =
            (response as &dyn Any).downcast_ref::<metadata_response::MetadataResponse>()
        {
            let endpoints = metadata
                .brokers
                .iter()
                .filter(|broker| !broker.host.is_empty() && broker.port > 0)
                .map(|broker| format_host_port(&broker.host, broker.port))
                .collect::<Vec<_>>();
            if !endpoints.is_empty() {
                self.replace_known_addrs(endpoints);
            }
            return;
        }
        if self.target == BootstrapTarget::Controllers
            && let Some(cluster) = (response as &dyn Any).downcast_ref::<DescribeClusterResponse>()
        {
            let endpoints = cluster
                .brokers
                .iter()
                .filter(|controller| !controller.host.is_empty() && controller.port > 0)
                .map(|controller| format_host_port(&controller.host, controller.port))
                .collect::<Vec<_>>();
            if !endpoints.is_empty() {
                self.replace_known_addrs(endpoints);
            }
        }
    }

    pub(crate) fn uses_controller_bootstrap(&self) -> bool {
        self.target == BootstrapTarget::Controllers
    }

    /// The `(min, max)` versions of `api_key` that the current peer
    /// advertised, or `None` when it did not advertise the API.
    pub(crate) async fn advertised_api_range(&self, api_key: i16) -> Option<(i16, i16)> {
        self.inner.read().await.advertised_api_range(api_key)
    }

    fn begin_metadata_attempt<R: krabka_protocol::ProtocolRequest>(&self) {
        if R::API_KEY != metadata_request::MetadataRequest::API_KEY
            || self.strategy == MetadataRecoveryStrategy::None
        {
            return;
        }
        self.first_metadata_attempt
            .lock()
            .unwrap_or_else(PoisonError::into_inner)
            .get_or_insert_with(Instant::now);
    }

    fn response_requires_rebootstrap<T: Any>(response: &T) -> bool {
        (response as &dyn Any)
            .downcast_ref::<metadata_response::MetadataResponse>()
            .is_some_and(|metadata| metadata.error_code == 129)
    }

    fn metadata_attempt_timed_out<T: Any>(&self, response: &T) -> bool {
        if self.strategy == MetadataRecoveryStrategy::None {
            return false;
        }
        let Some(metadata) =
            (response as &dyn Any).downcast_ref::<metadata_response::MetadataResponse>()
        else {
            return false;
        };
        metadata.brokers.is_empty()
            && self
                .first_metadata_attempt
                .lock()
                .unwrap_or_else(PoisonError::into_inner)
                .is_some_and(|started| started.elapsed() >= self.trigger.time().to_std())
    }

    fn complete_metadata_attempt<T: Any>(&self, response: &T) {
        let has_brokers = (response as &dyn Any)
            .downcast_ref::<metadata_response::MetadataResponse>()
            .is_some_and(|metadata| !metadata.brokers.is_empty());
        if has_brokers {
            self.clear_metadata_attempt();
        }
    }

    fn clear_metadata_attempt(&self) {
        *self
            .first_metadata_attempt
            .lock()
            .unwrap_or_else(PoisonError::into_inner) = None;
    }
}

async fn send_connection_at_least<R>(
    connection: &Connection,
    request: R,
    min_version: i16,
) -> Result<R::Response, ClientError>
where
    R: krabka_protocol::ProtocolRequest,
{
    let (broker_min, broker_max) = connection
        .advertised_api_range(R::API_KEY)
        .unwrap_or((0, 0));
    let client_min = R::MIN_VERSION.max(min_version);
    let chosen = R::LATEST_STABLE_VERSION.min(broker_max);
    if chosen < client_min || chosen < broker_min {
        return Err(ClientError::IncompatibleVersion {
            api_key: R::API_KEY,
            broker_min,
            broker_max,
            client_min,
            client_max: R::LATEST_STABLE_VERSION,
        });
    }
    connection.send(request).await
}

impl AdminClient {
    /// The connection options of Kafka's admin client defaults with the
    /// supplied security policy and a generated `adminclient-<n>` client id.
    fn opts(security: Option<security::ClientSecurity>) -> Result<ConnectionOptions, AdminError> {
        Ok(AdminClientConfig {
            security,
            ..AdminClientConfig::default()
        }
        .resolve()?
        .connection)
    }

    /// Connects with Kafka's `AdminClientConfig` settings. Each call of the
    /// client uses the call deadline, retry backoff and retry count of
    /// `config`.
    ///
    /// # Errors
    /// Returns [`AdminError::InvalidConfig`] for invalid settings, and
    /// [`AdminError::Connect`] with the last cause if no bootstrap address
    /// connects.
    pub async fn connect_with_config(
        bootstrap_addrs: &[String],
        config: AdminClientConfig,
    ) -> Result<Self, AdminError> {
        let resolved = config.resolve()?;
        Self::connect_with_metadata_recovery_target(
            bootstrap_addrs,
            resolved.connection,
            resolved.retry,
            (
                resolved.metadata_recovery_strategy,
                resolved.metadata_recovery_rebootstrap_trigger,
            ),
            BootstrapTarget::Brokers,
            resolved.enable_metrics_push,
        )
        .await
    }

    /// Controller-bootstrap variant of [`Self::connect_with_config`]
    /// (KIP-919).
    ///
    /// # Errors
    /// Returns [`AdminError::InvalidConfig`] for invalid settings, or a
    /// connection, protocol, or broker error when no controller bootstrap
    /// endpoint can identify and connect to the active controller.
    pub async fn connect_controller_with_config(
        bootstrap_controllers: &[String],
        config: AdminClientConfig,
    ) -> Result<Self, AdminError> {
        let resolved = config.resolve()?;
        Self::connect_with_metadata_recovery_target(
            bootstrap_controllers,
            resolved.connection,
            resolved.retry,
            (
                resolved.metadata_recovery_strategy,
                resolved.metadata_recovery_rebootstrap_trigger,
            ),
            BootstrapTarget::Controllers,
            resolved.enable_metrics_push,
        )
        .await
    }

    /// Connects and applies optional client security. `None` means plaintext,
    /// which is identical to [`AdminClient::connect`].
    ///
    /// # Errors
    /// Returns [`AdminError::Connect`] if no bootstrap address accepted the
    /// (optionally secured) connection. Its `source` holds the last cause.
    pub async fn connect_secured(
        bootstrap_addrs: &[String],
        security: Option<security::ClientSecurity>,
    ) -> Result<Self, AdminError> {
        Self::connect_with_options(bootstrap_addrs, Self::opts(security)?).await
    }

    /// Connects with the supplied security and DNS deadline, and keeps the
    /// standard admin identity, TCP-connect timeout, and request timeout.
    ///
    /// # Errors
    /// Returns [`AdminError::Connect`] with the last cause if no bootstrap
    /// address connects.
    pub async fn connect_secured_with_dns_timeout(
        bootstrap_addrs: &[String],
        security: Option<security::ClientSecurity>,
        dns_timeout: krabka_client_core::ClientDnsTimeout,
    ) -> Result<Self, AdminError> {
        let mut options = Self::opts(security)?;
        options.dns_timeout = dns_timeout;
        Self::connect_with_options(bootstrap_addrs, options).await
    }

    /// Connects with the standard plaintext admin policy and a custom DNS
    /// deadline.
    ///
    /// # Errors
    /// Returns [`AdminError::Connect`] with the last cause if no bootstrap
    /// address connects.
    pub async fn connect_with_dns_timeout(
        bootstrap_addrs: &[String],
        dns_timeout: krabka_client_core::ClientDnsTimeout,
    ) -> Result<Self, AdminError> {
        Self::connect_secured_with_dns_timeout(bootstrap_addrs, None, dns_timeout).await
    }

    /// Connects with a complete connection-options template.
    ///
    /// # Errors
    /// Returns [`AdminError::Connect`] with the last cause if no bootstrap
    /// address accepted the connection.
    pub async fn connect_with_options(
        bootstrap_addrs: &[String],
        options: ConnectionOptions,
    ) -> Result<Self, AdminError> {
        let retry = config::retry_for_connection_options(&options);
        Self::connect_with_metadata_recovery_target(
            bootstrap_addrs,
            options,
            retry,
            (
                MetadataRecoveryStrategy::default(),
                krabka_client_core::DEFAULT_METADATA_RECOVERY_REBOOTSTRAP_TRIGGER,
            ),
            BootstrapTarget::Brokers,
            DEFAULT_ENABLE_METRICS_PUSH,
        )
        .await
    }

    /// Connects through KIP-919 controller bootstrap addresses rather than
    /// broker bootstrap addresses.
    ///
    /// # Errors
    /// Returns a connection, protocol, or broker error when no controller
    /// bootstrap endpoint can identify and connect to the active controller.
    pub async fn connect_controller(bootstrap_controllers: &[String]) -> Result<Self, AdminError> {
        Self::connect_controller_secured(bootstrap_controllers, None).await
    }

    /// Controller-bootstrap variant of [`Self::connect_secured`].
    ///
    /// # Errors
    /// Returns a connection, protocol, or broker error when no controller
    /// bootstrap endpoint can identify and connect to the active controller.
    pub async fn connect_controller_secured(
        bootstrap_controllers: &[String],
        security: Option<security::ClientSecurity>,
    ) -> Result<Self, AdminError> {
        Self::connect_controller_with_options(bootstrap_controllers, Self::opts(security)?).await
    }

    /// Controller-bootstrap variant of [`Self::connect_with_options`].
    ///
    /// # Errors
    /// Returns a connection, protocol, or broker error when no controller
    /// bootstrap endpoint can identify and connect to the active controller.
    pub async fn connect_controller_with_options(
        bootstrap_controllers: &[String],
        options: ConnectionOptions,
    ) -> Result<Self, AdminError> {
        let retry = config::retry_for_connection_options(&options);
        Self::connect_with_metadata_recovery_target(
            bootstrap_controllers,
            options,
            retry,
            (
                MetadataRecoveryStrategy::default(),
                krabka_client_core::DEFAULT_METADATA_RECOVERY_REBOOTSTRAP_TRIGGER,
            ),
            BootstrapTarget::Controllers,
            DEFAULT_ENABLE_METRICS_PUSH,
        )
        .await
    }

    /// Connect with explicit KIP-1102 metadata recovery settings.
    ///
    /// # Errors
    /// Returns an invalid-configuration error for a negative or fractional
    /// millisecond trigger, or a connection error when no bootstrap endpoint
    /// is reachable.
    pub async fn connect_with_metadata_recovery(
        bootstrap_addrs: &[String],
        options: ConnectionOptions,
        strategy: MetadataRecoveryStrategy,
        rebootstrap_trigger: Time,
    ) -> Result<Self, AdminError> {
        let retry = config::retry_for_connection_options(&options);
        Self::connect_with_metadata_recovery_target(
            bootstrap_addrs,
            options,
            retry,
            (strategy, rebootstrap_trigger),
            BootstrapTarget::Brokers,
            DEFAULT_ENABLE_METRICS_PUSH,
        )
        .await
    }

    async fn connect_with_metadata_recovery_target(
        bootstrap_addrs: &[String],
        mut options: ConnectionOptions,
        retry: retry::RetryPolicy,
        (strategy, rebootstrap_trigger): (MetadataRecoveryStrategy, Time),
        target: BootstrapTarget,
        enable_metrics_push: bool,
    ) -> Result<Self, AdminError> {
        let trigger = MetadataRecoveryRebootstrapTrigger::new(rebootstrap_trigger)
            .map_err(AdminError::Protocol)?;
        // Kafka's `CommonClientConfigs.telemetryReporter`: a reporter only
        // with `enable.metrics.push`. Every connection of the client counts
        // in its network metrics.
        let metrics = enable_metrics_push.then(ClientMetrics::new);
        options.network_metrics = metrics.as_ref().map(|metrics| {
            NetworkMetrics::register(metrics, TelemetryClientType::Admin.metric_group())
        });
        let mut last_error = None;
        for host_port in bootstrap_addrs {
            match Self::connect_target_one_with_discovery(host_port, options.clone(), target).await
            {
                Ok(connected) => {
                    let known_addrs = if connected.discovered_addrs.is_empty() {
                        bootstrap_addrs.to_vec()
                    } else {
                        connected.discovered_addrs
                    };
                    let conn = Arc::new(RecoveringConnection::new(
                        connected.connection,
                        bootstrap_addrs.to_vec(),
                        options.clone(),
                        strategy,
                        trigger,
                        target,
                        known_addrs,
                    ));
                    let telemetry = metrics.map(|metrics| {
                        ClientTelemetry::start(
                            telemetry::AdminTelemetryConnection(Arc::clone(&conn)),
                            &ClientTelemetryConfig::admin(),
                            metrics,
                            TelemetryTiming {
                                request_timeout: options.request_timeout.to_std(),
                                reconnect_backoff: options.reconnect_backoff.to_std(),
                            },
                        )
                    });
                    return Ok(Self {
                        conn,
                        bootstrap_addrs: bootstrap_addrs.to_vec(),
                        options,
                        retry,
                        telemetry,
                    });
                }
                Err(e) if e.is_authentication_failure() => return Err(e),
                Err(e) => {
                    tracing::debug!(
                        target: "krabka_client_admin",
                        addr = %host_port,
                        error = %e,
                        "bootstrap connect failed",
                    );
                    last_error = Some(e);
                }
            }
        }
        match (target, last_error) {
            (BootstrapTarget::Controllers, Some(error)) => Err(error),
            (_, last_error) => Err(AdminError::Connect {
                tried: bootstrap_addrs.len(),
                source: last_error.map(Box::new),
            }),
        }
    }

    /// Tries each bootstrap address in order.
    ///
    /// Each entry is `host:port`, which
    /// [`krabka_client_core::transport::resolve`] resolves.
    /// The first successful connect wins. Returns
    /// [`AdminError::Connect`] with the last cause if none responded. The connection is
    /// plaintext. See [`AdminClient::connect_secured`].
    ///
    /// # Errors
    /// Returns an error when configuration is invalid, protocol encoding fails, the broker rejects the request, or transport I/O fails.
    pub async fn connect(bootstrap_addrs: &[String]) -> Result<Self, AdminError> {
        Self::connect_secured(bootstrap_addrs, None).await
    }

    async fn connect_one(
        host_port: &str,
        mut opts: ConnectionOptions,
    ) -> Result<Connection, AdminError> {
        if let Some(security) = opts.security.as_mut() {
            **security = security.for_target_host(connection_target_host(host_port));
        }
        let addr = lookup_first(host_port, opts.dns_timeout, transport::resolve(host_port)).await?;
        Connection::connect_with_options(addr, opts)
            .await
            .map_err(AdminError::from)
    }

    #[cfg(test)]
    async fn connect_target_one(
        host_port: &str,
        options: ConnectionOptions,
        target: BootstrapTarget,
    ) -> Result<Connection, AdminError> {
        Self::connect_target_one_with_discovery(host_port, options, target)
            .await
            .map(|connected| connected.connection)
    }

    async fn connect_target_one_with_discovery(
        host_port: &str,
        options: ConnectionOptions,
        target: BootstrapTarget,
    ) -> Result<ConnectedTarget, AdminError> {
        match target {
            BootstrapTarget::Brokers => Ok(ConnectedTarget {
                connection: Self::connect_one(host_port, options).await?,
                discovered_addrs: Vec::new(),
            }),
            BootstrapTarget::Controllers => {
                Self::connect_controller_one_with_discovery(host_port, options).await
            }
        }
    }

    async fn connect_controller_one_with_discovery(
        host_port: &str,
        options: ConnectionOptions,
    ) -> Result<ConnectedTarget, AdminError> {
        use krabka_protocol::owned::describe_cluster_request::DescribeClusterRequest;

        const ENDPOINT_TYPE_CONTROLLERS: i8 = 2;
        let bootstrap = Self::connect_one(host_port, options.clone()).await?;
        let response = send_connection_at_least(
            &bootstrap,
            DescribeClusterRequest {
                endpoint_type: ENDPOINT_TYPE_CONTROLLERS,
                ..Default::default()
            },
            1,
        )
        .await?;
        if response.error_code != 0 {
            return Err(AdminError::Broker {
                api: "DescribeCluster",
                code: response.error_code,
                name: kafka_error_name(response.error_code),
                message: response.error_message,
            });
        }
        if response.endpoint_type != ENDPOINT_TYPE_CONTROLLERS {
            return Err(AdminError::Protocol(format!(
                "DescribeCluster returned endpoint_type={}, expected CONTROLLERS",
                response.endpoint_type
            )));
        }
        let discovered_addrs = response
            .brokers
            .iter()
            .filter(|controller| !controller.host.is_empty() && controller.port > 0)
            .map(|controller| format_host_port(&controller.host, controller.port))
            .collect::<Vec<_>>();
        let controller = response
            .brokers
            .into_iter()
            .find(|controller| controller.broker_id == response.controller_id)
            .ok_or_else(|| {
                AdminError::Protocol(format!(
                    "DescribeCluster omitted controller_id={} from controller endpoints",
                    response.controller_id
                ))
            })?;
        if controller.host.is_empty() || controller.port <= 0 {
            return Err(AdminError::Protocol(format!(
                "DescribeCluster returned invalid controller endpoint {}:{}",
                controller.host, controller.port
            )));
        }
        let endpoint = format_host_port(&controller.host, controller.port);
        Ok(ConnectedTarget {
            connection: Self::connect_one(&endpoint, options).await?,
            discovered_addrs,
        })
    }

    /// Replaces the underlying connection. The `NOT_CONTROLLER` retry path
    /// uses it internally to reconnect to the current controller.
    pub(crate) async fn reconnect(&self, host_port: &str) -> Result<(), AdminError> {
        let opts = self.options.clone();
        self.conn
            .replace(Self::connect_one(host_port, opts).await?)
            .await;
        Ok(())
    }

    #[cfg(test)]
    pub(crate) async fn reconnect_bootstrap(&self) -> Result<(), AdminError> {
        let opts = self.options.clone();
        let mut last_error = None;
        for host_port in &self.bootstrap_addrs {
            match Self::connect_target_one(host_port, opts.clone(), self.conn.target).await {
                Ok(conn) => {
                    self.conn.replace(conn).await;
                    return Ok(());
                }
                Err(error) if error.is_authentication_failure() => return Err(error),
                Err(error) => {
                    tracing::debug!(
                        target: "krabka_client_admin",
                        addr = %host_port,
                        error = %error,
                        "bootstrap reconnect failed",
                    );
                    last_error = Some(Box::new(error));
                }
            }
        }
        Err(AdminError::Connect {
            tried: self.bootstrap_addrs.len(),
            source: last_error,
        })
    }

    pub(crate) fn is_retriable_transport_error(error: &ClientError) -> bool {
        matches!(
            error,
            ClientError::Timeout(_) | ClientError::Disconnected | ClientError::Io(_)
        )
    }
}

/// Kafka error code: the broker is not the controller (KIP-129). The
/// admin client refreshes its controller endpoint and retries until the call
/// deadline.
pub(crate) const NOT_CONTROLLER: i16 = 41;

/// Maps a Kafka error code to the name of its `Errors` constant, for
/// human-friendly `AdminError::Broker` output.
///
/// The table holds every code of Apache Kafka's
/// `org.apache.kafka.common.protocol.Errors` enum at apache/kafka@3eb0b8c.
/// A code that Kafka does not define gives `"UNKNOWN"`.
pub(crate) fn kafka_error_name(code: i16) -> &'static str {
    match code {
        -1 => "UNKNOWN_SERVER_ERROR",
        0 => "NONE",
        1 => "OFFSET_OUT_OF_RANGE",
        2 => "CORRUPT_MESSAGE",
        3 => "UNKNOWN_TOPIC_OR_PARTITION",
        4 => "INVALID_FETCH_SIZE",
        5 => "LEADER_NOT_AVAILABLE",
        6 => "NOT_LEADER_OR_FOLLOWER",
        7 => "REQUEST_TIMED_OUT",
        8 => "BROKER_NOT_AVAILABLE",
        9 => "REPLICA_NOT_AVAILABLE",
        10 => "MESSAGE_TOO_LARGE",
        11 => "STALE_CONTROLLER_EPOCH",
        12 => "OFFSET_METADATA_TOO_LARGE",
        13 => "NETWORK_EXCEPTION",
        14 => "COORDINATOR_LOAD_IN_PROGRESS",
        15 => "COORDINATOR_NOT_AVAILABLE",
        16 => "NOT_COORDINATOR",
        17 => "INVALID_TOPIC_EXCEPTION",
        18 => "RECORD_LIST_TOO_LARGE",
        19 => "NOT_ENOUGH_REPLICAS",
        20 => "NOT_ENOUGH_REPLICAS_AFTER_APPEND",
        21 => "INVALID_REQUIRED_ACKS",
        22 => "ILLEGAL_GENERATION",
        23 => "INCONSISTENT_GROUP_PROTOCOL",
        24 => "INVALID_GROUP_ID",
        25 => "UNKNOWN_MEMBER_ID",
        26 => "INVALID_SESSION_TIMEOUT",
        27 => "REBALANCE_IN_PROGRESS",
        28 => "INVALID_COMMIT_OFFSET_SIZE",
        29 => "TOPIC_AUTHORIZATION_FAILED",
        30 => "GROUP_AUTHORIZATION_FAILED",
        31 => "CLUSTER_AUTHORIZATION_FAILED",
        32 => "INVALID_TIMESTAMP",
        33 => "UNSUPPORTED_SASL_MECHANISM",
        34 => "ILLEGAL_SASL_STATE",
        35 => "UNSUPPORTED_VERSION",
        36 => "TOPIC_ALREADY_EXISTS",
        37 => "INVALID_PARTITIONS",
        38 => "INVALID_REPLICATION_FACTOR",
        39 => "INVALID_REPLICA_ASSIGNMENT",
        40 => "INVALID_CONFIG",
        41 => "NOT_CONTROLLER",
        42 => "INVALID_REQUEST",
        43 => "UNSUPPORTED_FOR_MESSAGE_FORMAT",
        44 => "POLICY_VIOLATION",
        45 => "OUT_OF_ORDER_SEQUENCE_NUMBER",
        46 => "DUPLICATE_SEQUENCE_NUMBER",
        47 => "INVALID_PRODUCER_EPOCH",
        48 => "INVALID_TXN_STATE",
        49 => "INVALID_PRODUCER_ID_MAPPING",
        50 => "INVALID_TRANSACTION_TIMEOUT",
        51 => "CONCURRENT_TRANSACTIONS",
        52 => "TRANSACTION_COORDINATOR_FENCED",
        53 => "TRANSACTIONAL_ID_AUTHORIZATION_FAILED",
        54 => "SECURITY_DISABLED",
        55 => "OPERATION_NOT_ATTEMPTED",
        56 => "KAFKA_STORAGE_ERROR",
        57 => "LOG_DIR_NOT_FOUND",
        58 => "SASL_AUTHENTICATION_FAILED",
        59 => "UNKNOWN_PRODUCER_ID",
        60 => "REASSIGNMENT_IN_PROGRESS",
        61 => "DELEGATION_TOKEN_AUTH_DISABLED",
        62 => "DELEGATION_TOKEN_NOT_FOUND",
        63 => "DELEGATION_TOKEN_OWNER_MISMATCH",
        64 => "DELEGATION_TOKEN_REQUEST_NOT_ALLOWED",
        65 => "DELEGATION_TOKEN_AUTHORIZATION_FAILED",
        66 => "DELEGATION_TOKEN_EXPIRED",
        67 => "INVALID_PRINCIPAL_TYPE",
        68 => "NON_EMPTY_GROUP",
        69 => "GROUP_ID_NOT_FOUND",
        70 => "FETCH_SESSION_ID_NOT_FOUND",
        71 => "INVALID_FETCH_SESSION_EPOCH",
        72 => "LISTENER_NOT_FOUND",
        73 => "TOPIC_DELETION_DISABLED",
        74 => "FENCED_LEADER_EPOCH",
        75 => "UNKNOWN_LEADER_EPOCH",
        76 => "UNSUPPORTED_COMPRESSION_TYPE",
        77 => "STALE_BROKER_EPOCH",
        78 => "OFFSET_NOT_AVAILABLE",
        79 => "MEMBER_ID_REQUIRED",
        80 => "PREFERRED_LEADER_NOT_AVAILABLE",
        81 => "GROUP_MAX_SIZE_REACHED",
        82 => "FENCED_INSTANCE_ID",
        83 => "ELIGIBLE_LEADERS_NOT_AVAILABLE",
        84 => "ELECTION_NOT_NEEDED",
        85 => "NO_REASSIGNMENT_IN_PROGRESS",
        86 => "GROUP_SUBSCRIBED_TO_TOPIC",
        87 => "INVALID_RECORD",
        88 => "UNSTABLE_OFFSET_COMMIT",
        89 => "THROTTLING_QUOTA_EXCEEDED",
        90 => "PRODUCER_FENCED",
        91 => "RESOURCE_NOT_FOUND",
        92 => "DUPLICATE_RESOURCE",
        93 => "UNACCEPTABLE_CREDENTIAL",
        94 => "INCONSISTENT_VOTER_SET",
        95 => "INVALID_UPDATE_VERSION",
        96 => "FEATURE_UPDATE_FAILED",
        97 => "PRINCIPAL_DESERIALIZATION_FAILURE",
        98 => "SNAPSHOT_NOT_FOUND",
        99 => "POSITION_OUT_OF_RANGE",
        100 => "UNKNOWN_TOPIC_ID",
        101 => "DUPLICATE_BROKER_REGISTRATION",
        102 => "BROKER_ID_NOT_REGISTERED",
        103 => "INCONSISTENT_TOPIC_ID",
        104 => "INCONSISTENT_CLUSTER_ID",
        105 => "TRANSACTIONAL_ID_NOT_FOUND",
        106 => "FETCH_SESSION_TOPIC_ID_ERROR",
        107 => "INELIGIBLE_REPLICA",
        108 => "NEW_LEADER_ELECTED",
        109 => "OFFSET_MOVED_TO_TIERED_STORAGE",
        110 => "FENCED_MEMBER_EPOCH",
        111 => "UNRELEASED_INSTANCE_ID",
        112 => "UNSUPPORTED_ASSIGNOR",
        113 => "STALE_MEMBER_EPOCH",
        114 => "MISMATCHED_ENDPOINT_TYPE",
        115 => "UNSUPPORTED_ENDPOINT_TYPE",
        116 => "UNKNOWN_CONTROLLER_ID",
        117 => "UNKNOWN_SUBSCRIPTION_ID",
        118 => "TELEMETRY_TOO_LARGE",
        119 => "INVALID_REGISTRATION",
        120 => "TRANSACTION_ABORTABLE",
        121 => "INVALID_RECORD_STATE",
        122 => "SHARE_SESSION_NOT_FOUND",
        123 => "INVALID_SHARE_SESSION_EPOCH",
        124 => "FENCED_STATE_EPOCH",
        125 => "INVALID_VOTER_KEY",
        126 => "DUPLICATE_VOTER",
        127 => "VOTER_NOT_FOUND",
        128 => "INVALID_REGULAR_EXPRESSION",
        129 => "REBOOTSTRAP_REQUIRED",
        130 => "STREAMS_INVALID_TOPOLOGY",
        131 => "STREAMS_INVALID_TOPOLOGY_EPOCH",
        132 => "STREAMS_TOPOLOGY_FENCED",
        133 => "SHARE_SESSION_LIMIT_REACHED",
        134 => "GROUP_DELETION_FAILED",
        135 => "STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED",
        136 => "CONTROLLER_ID_NOT_REGISTERED",
        _ => "UNKNOWN",
    }
}

#[cfg(test)]
mod tests {
    use std::{
        future::pending,
        hint,
        sync::{
            Arc, Mutex,
            atomic::{AtomicU16, AtomicUsize, Ordering},
        },
        time::Duration,
        vec,
    };

    use bytes::{BufMut, BytesMut};
    use krabka_protocol::{
        Decode, Encode,
        owned::{
            api_versions_request,
            api_versions_response::{ApiVersion, ApiVersionsResponse},
            describe_cluster_request::{self, DescribeClusterRequest},
            describe_cluster_response::{DescribeClusterBroker, DescribeClusterResponse},
            metadata_request,
            metadata_response::{
                FLEXIBLE_MIN as METADATA_FLEXIBLE_MIN, MetadataResponse, MetadataResponseBroker,
            },
            sasl_authenticate_request,
            sasl_authenticate_response::SaslAuthenticateResponse,
            sasl_handshake_request,
            sasl_handshake_response::SaslHandshakeResponse,
        },
    };
    use krabka_security::ListenerProtocol;
    use krabka_units::secs;
    use security::{ClientSecurity, SaslCredentials};
    use tokio::{
        io::{AsyncReadExt, AsyncWriteExt},
        net::{TcpListener, TcpStream},
        task::yield_now,
        time::sleep,
    };
    use tokio_util::sync::CancellationToken;

    use super::*;

    struct ObservedAdminBroker {
        addr: SocketAddr,
        shutdown: CancellationToken,
        connections: Arc<AtomicUsize>,
        sasl_handshakes: Arc<AtomicUsize>,
        client_ids: Arc<Mutex<Vec<String>>>,
    }

    impl ObservedAdminBroker {
        async fn start(api_versions_delay: Duration) -> Self {
            let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = listener.local_addr().unwrap();
            let shutdown = CancellationToken::new();
            let connections = Arc::new(AtomicUsize::new(0));
            let sasl_handshakes = Arc::new(AtomicUsize::new(0));
            let client_ids = Arc::new(Mutex::new(Vec::new()));
            let task_shutdown = shutdown.clone();
            let task_connections = Arc::clone(&connections);
            let task_sasl_handshakes = Arc::clone(&sasl_handshakes);
            let task_client_ids = Arc::clone(&client_ids);

            tokio::spawn(async move {
                loop {
                    let Ok((mut stream, _)) = listener.accept().await else {
                        break;
                    };
                    if task_shutdown.is_cancelled() {
                        break;
                    }
                    task_connections.fetch_add(1, Ordering::SeqCst);
                    let conn_sasl_handshakes = Arc::clone(&task_sasl_handshakes);
                    let conn_client_ids = Arc::clone(&task_client_ids);
                    tokio::spawn(async move {
                        while let Ok(frame_len) = stream.read_u32().await {
                            let mut request = vec![0_u8; frame_len as usize];
                            if stream.read_exact(&mut request).await.is_err() || request.len() < 10
                            {
                                break;
                            }
                            let api_key = i16::from_be_bytes([request[0], request[1]]);
                            let correlation_id =
                                i32::from_be_bytes(request[4..8].try_into().unwrap());
                            let client_id_len =
                                usize::from(u16::from_be_bytes([request[8], request[9]]));
                            if request.len() < 10 + client_id_len {
                                break;
                            }
                            conn_client_ids.lock().unwrap().push(
                                String::from_utf8_lossy(&request[10..10 + client_id_len]).into(),
                            );

                            let (body, flexible_header) = match api_key {
                                sasl_handshake_request::API_KEY => {
                                    conn_sasl_handshakes.fetch_add(1, Ordering::SeqCst);
                                    let mut body = BytesMut::new();
                                    SaslHandshakeResponse {
                                        error_code: 0,
                                        ..Default::default()
                                    }
                                    .encode(&mut body, 1)
                                    .unwrap();
                                    (body, false)
                                }
                                sasl_authenticate_request::API_KEY => {
                                    let mut body = BytesMut::new();
                                    SaslAuthenticateResponse {
                                        error_code: 0,
                                        ..Default::default()
                                    }
                                    .encode(&mut body, 2)
                                    .unwrap();
                                    (body, true)
                                }
                                api_versions_request::API_KEY => {
                                    sleep(api_versions_delay).await;
                                    let mut body = BytesMut::new();
                                    ApiVersionsResponse {
                                        error_code: 0,
                                        api_keys: vec![
                                            ApiVersion {
                                                api_key: api_versions_request::API_KEY,
                                                min_version: 0,
                                                max_version: 3,
                                                ..Default::default()
                                            },
                                            ApiVersion {
                                                api_key: metadata_request::API_KEY,
                                                min_version: 0,
                                                max_version: 12,
                                                ..Default::default()
                                            },
                                            ApiVersion {
                                                api_key: sasl_handshake_request::API_KEY,
                                                min_version: 0,
                                                max_version: 1,
                                                ..Default::default()
                                            },
                                            ApiVersion {
                                                api_key: sasl_authenticate_request::API_KEY,
                                                min_version: 0,
                                                max_version: 2,
                                                ..Default::default()
                                            },
                                        ],
                                        ..Default::default()
                                    }
                                    .encode(&mut body, 0)
                                    .unwrap();
                                    (body, false)
                                }
                                _ => continue,
                            };
                            let mut response = BytesMut::new();
                            response.put_i32(correlation_id);
                            if flexible_header {
                                response.put_u8(0);
                            }
                            response.extend_from_slice(&body);
                            if stream
                                .write_u32(u32::try_from(response.len()).unwrap())
                                .await
                                .is_err()
                                || stream.write_all(&response).await.is_err()
                            {
                                break;
                            }
                        }
                    });
                }
            });

            Self {
                addr,
                shutdown,
                connections,
                sasl_handshakes,
                client_ids,
            }
        }

        fn observed_custom_security_and_id(&self) -> bool {
            self.sasl_handshakes.load(Ordering::SeqCst) > 0
                && self
                    .client_ids
                    .lock()
                    .unwrap()
                    .iter()
                    .any(|client_id| client_id == "custom-admin")
        }

        fn stop(self) {
            self.shutdown.cancel();
        }
    }

    struct ObservedController {
        addr: SocketAddr,
        connections: Arc<AtomicUsize>,
        endpoint_types: Arc<Mutex<Vec<i8>>>,
    }

    impl ObservedController {
        async fn start(describe_cluster_max_version: i16) -> Self {
            let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
            let addr = listener.local_addr().unwrap();
            let connections = Arc::new(AtomicUsize::new(0));
            let endpoint_types = Arc::new(Mutex::new(Vec::new()));
            let task_connections = Arc::clone(&connections);
            let task_endpoint_types = Arc::clone(&endpoint_types);

            tokio::spawn(async move {
                while let Ok((mut stream, _)) = listener.accept().await {
                    task_connections.fetch_add(1, Ordering::SeqCst);
                    let endpoint_types = Arc::clone(&task_endpoint_types);
                    tokio::spawn(async move {
                        while let Ok(frame_len) = stream.read_u32().await {
                            let mut request = vec![0_u8; frame_len as usize];
                            if stream.read_exact(&mut request).await.is_err() || request.len() < 10
                            {
                                break;
                            }
                            let api_key = i16::from_be_bytes([request[0], request[1]]);
                            let api_version = i16::from_be_bytes([request[2], request[3]]);
                            let correlation_id =
                                i32::from_be_bytes(request[4..8].try_into().unwrap());
                            let client_id_len =
                                usize::from(u16::from_be_bytes([request[8], request[9]]));
                            let header_len = 10 + client_id_len;
                            if request.len() < header_len {
                                break;
                            }

                            let (body, flexible_header) = match api_key {
                                api_versions_request::API_KEY => {
                                    let mut body = BytesMut::new();
                                    ApiVersionsResponse {
                                        api_keys: vec![
                                            ApiVersion {
                                                api_key: api_versions_request::API_KEY,
                                                min_version: 0,
                                                max_version: 4,
                                                ..Default::default()
                                            },
                                            ApiVersion {
                                                api_key: describe_cluster_request::API_KEY,
                                                min_version: 0,
                                                max_version: describe_cluster_max_version,
                                                ..Default::default()
                                            },
                                        ],
                                        ..Default::default()
                                    }
                                    .encode(&mut body, 0)
                                    .unwrap();
                                    (body, false)
                                }
                                describe_cluster_request::API_KEY => {
                                    let mut encoded = &request[header_len + 1..];
                                    let decoded =
                                        DescribeClusterRequest::decode(&mut encoded, api_version)
                                            .unwrap();
                                    endpoint_types.lock().unwrap().push(decoded.endpoint_type);
                                    let mut body = BytesMut::new();
                                    DescribeClusterResponse {
                                        endpoint_type: 2,
                                        cluster_id: "controller-test".into(),
                                        controller_id: 1,
                                        brokers: vec![DescribeClusterBroker {
                                            broker_id: 1,
                                            host: addr.ip().to_string(),
                                            port: i32::from(addr.port()),
                                            ..Default::default()
                                        }],
                                        ..Default::default()
                                    }
                                    .encode(&mut body, api_version)
                                    .unwrap();
                                    (body, true)
                                }
                                _ => continue,
                            };
                            let mut response = BytesMut::new();
                            response.put_i32(correlation_id);
                            if flexible_header {
                                response.put_u8(0);
                            }
                            response.extend_from_slice(&body);
                            if stream
                                .write_u32(u32::try_from(response.len()).unwrap())
                                .await
                                .is_err()
                                || stream.write_all(&response).await.is_err()
                            {
                                break;
                            }
                        }
                    });
                }
            });

            Self {
                addr,
                connections,
                endpoint_types,
            }
        }
    }

    fn custom_admin_options() -> ConnectionOptions {
        ConnectionOptions {
            dns_timeout: ClientDnsTimeout::default(),
            client_id: "custom-admin".into(),
            socket_connection_setup_timeout: krabka_units::millis(100),
            request_timeout: krabka_units::millis(150),
            dispatch_queue_capacity: krabka_client_core::ConnectionDispatchQueueCapacity::new(7)
                .unwrap(),
            frame_max: krabka_client_core::ClientFrameMax::try_from(krabka_units::kibibytes(32))
                .unwrap(),
            security: Some(Box::new(ClientSecurity {
                protocol: ListenerProtocol::SaslPlaintext,
                tls: None,
                sasl: Some(SaslCredentials::Plain {
                    username: "u".into(),
                    password: "p".into(),
                }),
                sasl_host: Some("broker.example".into()),
            })),
            ..ConnectionOptions::default()
        }
    }

    async fn metadata_times_out_with_custom_request_timeout(admin: &mut AdminClient) {
        let result = timeout(Duration::from_secs(2), admin.metadata(&[]))
            .await
            .expect("custom request timeout fires");
        assert2::assert!(result.is_err());
    }

    fn assert_custom_connect_timeout_is_stored(admin: &AdminClient) {
        assert2::assert!(
            admin.options.socket_connection_setup_timeout == krabka_units::millis(100)
        );
        assert2::assert!(admin.options.dispatch_queue_capacity.get() == 7);
        assert2::assert!(admin.options.frame_max.size() == krabka_units::kibibytes(32));
    }

    fn metadata_mock_response(version: i16, broker_id: i32, port: u16) -> Vec<u8> {
        let response = MetadataResponse {
            brokers: vec![MetadataResponseBroker {
                node_id: broker_id,
                host: "127.0.0.1".into(),
                port: i32::from(port),
                ..Default::default()
            }],
            ..Default::default()
        };
        let mut bytes = BytesMut::new();
        if version >= METADATA_FLEXIBLE_MIN {
            bytes.put_u8(0);
        }
        response.encode(&mut bytes, version).unwrap();
        bytes.to_vec()
    }

    fn admin_mock_api_versions() -> Vec<u8> {
        let response = ApiVersionsResponse {
            api_keys: vec![
                ApiVersion {
                    api_key: api_versions_request::API_KEY,
                    min_version: 0,
                    max_version: 3,
                    ..Default::default()
                },
                ApiVersion {
                    api_key: metadata_request::API_KEY,
                    min_version: 0,
                    max_version: 13,
                    ..Default::default()
                },
            ],
            ..Default::default()
        };
        let mut bytes = BytesMut::new();
        response.encode(&mut bytes, 0).unwrap();
        bytes.to_vec()
    }

    fn admin_mock_api_versions_with_describe_cluster(max_version: i16) -> Vec<u8> {
        let response = ApiVersionsResponse {
            api_keys: vec![
                ApiVersion {
                    api_key: api_versions_request::API_KEY,
                    min_version: 0,
                    max_version: 3,
                    ..Default::default()
                },
                ApiVersion {
                    api_key: describe_cluster_request::API_KEY,
                    min_version: 0,
                    max_version,
                    ..Default::default()
                },
            ],
            ..Default::default()
        };
        let mut bytes = BytesMut::new();
        response.encode(&mut bytes, 0).unwrap();
        bytes.to_vec()
    }

    async fn wait_for_admin_mock_closed(addr: SocketAddr) {
        timeout(Duration::from_secs(10), async {
            while TcpStream::connect(addr).await.is_ok() {
                yield_now().await;
            }
        })
        .await
        .expect("mock broker listener closes");
    }

    #[test]
    fn error_names_match_the_codes_that_apache_kafka_defines() {
        // Every constant of Apache Kafka's `Errors` enum at
        // apache/kafka@3eb0b8c, in code order, as `Errors.values()` prints
        // `code()` and `name()`. The list is exhaustive on purpose: a sampled
        // table lets a deleted arm pass unnoticed, and a wrong name sends an
        // operator to the wrong cause.
        let cases: [(i16, &str); 138] = [
            (-1, "UNKNOWN_SERVER_ERROR"),
            (0, "NONE"),
            (1, "OFFSET_OUT_OF_RANGE"),
            (2, "CORRUPT_MESSAGE"),
            (3, "UNKNOWN_TOPIC_OR_PARTITION"),
            (4, "INVALID_FETCH_SIZE"),
            (5, "LEADER_NOT_AVAILABLE"),
            (6, "NOT_LEADER_OR_FOLLOWER"),
            (7, "REQUEST_TIMED_OUT"),
            (8, "BROKER_NOT_AVAILABLE"),
            (9, "REPLICA_NOT_AVAILABLE"),
            (10, "MESSAGE_TOO_LARGE"),
            (11, "STALE_CONTROLLER_EPOCH"),
            (12, "OFFSET_METADATA_TOO_LARGE"),
            (13, "NETWORK_EXCEPTION"),
            (14, "COORDINATOR_LOAD_IN_PROGRESS"),
            (15, "COORDINATOR_NOT_AVAILABLE"),
            (16, "NOT_COORDINATOR"),
            (17, "INVALID_TOPIC_EXCEPTION"),
            (18, "RECORD_LIST_TOO_LARGE"),
            (19, "NOT_ENOUGH_REPLICAS"),
            (20, "NOT_ENOUGH_REPLICAS_AFTER_APPEND"),
            (21, "INVALID_REQUIRED_ACKS"),
            (22, "ILLEGAL_GENERATION"),
            (23, "INCONSISTENT_GROUP_PROTOCOL"),
            (24, "INVALID_GROUP_ID"),
            (25, "UNKNOWN_MEMBER_ID"),
            (26, "INVALID_SESSION_TIMEOUT"),
            (27, "REBALANCE_IN_PROGRESS"),
            (28, "INVALID_COMMIT_OFFSET_SIZE"),
            (29, "TOPIC_AUTHORIZATION_FAILED"),
            (30, "GROUP_AUTHORIZATION_FAILED"),
            (31, "CLUSTER_AUTHORIZATION_FAILED"),
            (32, "INVALID_TIMESTAMP"),
            (33, "UNSUPPORTED_SASL_MECHANISM"),
            (34, "ILLEGAL_SASL_STATE"),
            (35, "UNSUPPORTED_VERSION"),
            (36, "TOPIC_ALREADY_EXISTS"),
            (37, "INVALID_PARTITIONS"),
            (38, "INVALID_REPLICATION_FACTOR"),
            (39, "INVALID_REPLICA_ASSIGNMENT"),
            (40, "INVALID_CONFIG"),
            (41, "NOT_CONTROLLER"),
            (42, "INVALID_REQUEST"),
            (43, "UNSUPPORTED_FOR_MESSAGE_FORMAT"),
            (44, "POLICY_VIOLATION"),
            (45, "OUT_OF_ORDER_SEQUENCE_NUMBER"),
            (46, "DUPLICATE_SEQUENCE_NUMBER"),
            (47, "INVALID_PRODUCER_EPOCH"),
            (48, "INVALID_TXN_STATE"),
            (49, "INVALID_PRODUCER_ID_MAPPING"),
            (50, "INVALID_TRANSACTION_TIMEOUT"),
            (51, "CONCURRENT_TRANSACTIONS"),
            (52, "TRANSACTION_COORDINATOR_FENCED"),
            (53, "TRANSACTIONAL_ID_AUTHORIZATION_FAILED"),
            (54, "SECURITY_DISABLED"),
            (55, "OPERATION_NOT_ATTEMPTED"),
            (56, "KAFKA_STORAGE_ERROR"),
            (57, "LOG_DIR_NOT_FOUND"),
            (58, "SASL_AUTHENTICATION_FAILED"),
            (59, "UNKNOWN_PRODUCER_ID"),
            (60, "REASSIGNMENT_IN_PROGRESS"),
            (61, "DELEGATION_TOKEN_AUTH_DISABLED"),
            (62, "DELEGATION_TOKEN_NOT_FOUND"),
            (63, "DELEGATION_TOKEN_OWNER_MISMATCH"),
            (64, "DELEGATION_TOKEN_REQUEST_NOT_ALLOWED"),
            (65, "DELEGATION_TOKEN_AUTHORIZATION_FAILED"),
            (66, "DELEGATION_TOKEN_EXPIRED"),
            (67, "INVALID_PRINCIPAL_TYPE"),
            (68, "NON_EMPTY_GROUP"),
            (69, "GROUP_ID_NOT_FOUND"),
            (70, "FETCH_SESSION_ID_NOT_FOUND"),
            (71, "INVALID_FETCH_SESSION_EPOCH"),
            (72, "LISTENER_NOT_FOUND"),
            (73, "TOPIC_DELETION_DISABLED"),
            (74, "FENCED_LEADER_EPOCH"),
            (75, "UNKNOWN_LEADER_EPOCH"),
            (76, "UNSUPPORTED_COMPRESSION_TYPE"),
            (77, "STALE_BROKER_EPOCH"),
            (78, "OFFSET_NOT_AVAILABLE"),
            (79, "MEMBER_ID_REQUIRED"),
            (80, "PREFERRED_LEADER_NOT_AVAILABLE"),
            (81, "GROUP_MAX_SIZE_REACHED"),
            (82, "FENCED_INSTANCE_ID"),
            (83, "ELIGIBLE_LEADERS_NOT_AVAILABLE"),
            (84, "ELECTION_NOT_NEEDED"),
            (85, "NO_REASSIGNMENT_IN_PROGRESS"),
            (86, "GROUP_SUBSCRIBED_TO_TOPIC"),
            (87, "INVALID_RECORD"),
            (88, "UNSTABLE_OFFSET_COMMIT"),
            (89, "THROTTLING_QUOTA_EXCEEDED"),
            (90, "PRODUCER_FENCED"),
            (91, "RESOURCE_NOT_FOUND"),
            (92, "DUPLICATE_RESOURCE"),
            (93, "UNACCEPTABLE_CREDENTIAL"),
            (94, "INCONSISTENT_VOTER_SET"),
            (95, "INVALID_UPDATE_VERSION"),
            (96, "FEATURE_UPDATE_FAILED"),
            (97, "PRINCIPAL_DESERIALIZATION_FAILURE"),
            (98, "SNAPSHOT_NOT_FOUND"),
            (99, "POSITION_OUT_OF_RANGE"),
            (100, "UNKNOWN_TOPIC_ID"),
            (101, "DUPLICATE_BROKER_REGISTRATION"),
            (102, "BROKER_ID_NOT_REGISTERED"),
            (103, "INCONSISTENT_TOPIC_ID"),
            (104, "INCONSISTENT_CLUSTER_ID"),
            (105, "TRANSACTIONAL_ID_NOT_FOUND"),
            (106, "FETCH_SESSION_TOPIC_ID_ERROR"),
            (107, "INELIGIBLE_REPLICA"),
            (108, "NEW_LEADER_ELECTED"),
            (109, "OFFSET_MOVED_TO_TIERED_STORAGE"),
            (110, "FENCED_MEMBER_EPOCH"),
            (111, "UNRELEASED_INSTANCE_ID"),
            (112, "UNSUPPORTED_ASSIGNOR"),
            (113, "STALE_MEMBER_EPOCH"),
            (114, "MISMATCHED_ENDPOINT_TYPE"),
            (115, "UNSUPPORTED_ENDPOINT_TYPE"),
            (116, "UNKNOWN_CONTROLLER_ID"),
            (117, "UNKNOWN_SUBSCRIPTION_ID"),
            (118, "TELEMETRY_TOO_LARGE"),
            (119, "INVALID_REGISTRATION"),
            (120, "TRANSACTION_ABORTABLE"),
            (121, "INVALID_RECORD_STATE"),
            (122, "SHARE_SESSION_NOT_FOUND"),
            (123, "INVALID_SHARE_SESSION_EPOCH"),
            (124, "FENCED_STATE_EPOCH"),
            (125, "INVALID_VOTER_KEY"),
            (126, "DUPLICATE_VOTER"),
            (127, "VOTER_NOT_FOUND"),
            (128, "INVALID_REGULAR_EXPRESSION"),
            (129, "REBOOTSTRAP_REQUIRED"),
            (130, "STREAMS_INVALID_TOPOLOGY"),
            (131, "STREAMS_INVALID_TOPOLOGY_EPOCH"),
            (132, "STREAMS_TOPOLOGY_FENCED"),
            (133, "SHARE_SESSION_LIMIT_REACHED"),
            (134, "GROUP_DELETION_FAILED"),
            (135, "STREAMS_TOPOLOGY_DESCRIPTION_UPDATE_FAILED"),
            (136, "CONTROLLER_ID_NOT_REGISTERED"),
        ];

        let names = cases
            .iter()
            .map(|(code, _)| (*code, kafka_error_name(*code)))
            .collect::<Vec<_>>();
        assert2::assert!(names == cases.to_vec());
        // Kafka's codes are contiguous from -1, so every code outside the
        // table is one Kafka does not define.
    }

    #[yare::parameterized(
        minimum = { i16::MIN },
        negative = { -2 },
        beyond_table = { 137 },
        maximum = { i16::MAX },
    )]
    fn kafka_error_name_unknown_codes_are_unknown(code: i16) {
        assert2::assert!(kafka_error_name(code) == "UNKNOWN", "code {code}");
    }

    #[yare::parameterized(
        success = { 0, "NONE" },
        topic_exists = { 36, "TOPIC_ALREADY_EXISTS" },
        not_controller = { 41, "NOT_CONTROLLER" },
    )]
    fn kafka_error_name_known_codes(code: i16, want: &str) {
        assert2::assert!(kafka_error_name(code) == want);
    }

    #[yare::parameterized(
        cluster_authorization = { 31, "CLUSTER_AUTHORIZATION_FAILED" },
        unsupported_sasl = { 33, "UNSUPPORTED_SASL_MECHANISM" },
        unsupported_version = { 35, "UNSUPPORTED_VERSION" },
        delegation_token_expired = { 66, "DELEGATION_TOKEN_EXPIRED" },
        eligible_leaders_unavailable = { 83, "ELIGIBLE_LEADERS_NOT_AVAILABLE" },
        election_unnecessary = { 84, "ELECTION_NOT_NEEDED" },
        resource_not_found = { 91, "RESOURCE_NOT_FOUND" },
        duplicate_resource = { 92, "DUPLICATE_RESOURCE" },
        unacceptable_credential = { 93, "UNACCEPTABLE_CREDENTIAL" },
    )]
    fn users_kafka_error_name_includes_scram_describe_codes(code: i16, want: &str) {
        assert2::assert!(kafka_error_name(code) == want);
    }

    #[test]
    fn users_kafka_error_name_includes_ineligible_replica() {
        assert2::assert!(kafka_error_name(107) == "INELIGIBLE_REPLICA");
    }

    #[test]
    fn kafka_error_name_unknown_returns_unknown() {
        assert2::assert!(kafka_error_name(9999) == "UNKNOWN");
    }

    #[test]
    fn kafka_error_if_zero_code_is_none() {
        assert2::assert!(kafka_error_if(0, None).is_none());
    }

    #[test]
    fn kafka_error_if_nonzero_carries_name() {
        let e = kafka_error_if(36, Some("dup".into())).unwrap();
        assert2::assert!(
            e == KafkaError {
                code: 36,
                name: "TOPIC_ALREADY_EXISTS",
                message: Some("dup".to_owned()),
            }
        );
    }

    #[tokio::test]
    async fn connect_secured_threads_security_and_fails_to_closed_port() {
        use krabka_security::ListenerProtocol;
        use security::{ClientSecurity, SaslCredentials};

        let security = ClientSecurity {
            protocol: ListenerProtocol::SaslPlaintext,
            tls: None,
            sasl: Some(SaslCredentials::Plain {
                username: "u".into(),
                password: "p".into(),
            }),
            sasl_host: None,
        };
        // 127.0.0.1:1 has no listener; the secured connect must fail —
        // proving the security arg is threaded (not a type error).
        let res = AdminClient::connect_secured(&["127.0.0.1:1".to_owned()], Some(security)).await;
        assert2::assert!(res.is_err());
    }

    #[tokio::test(start_paused = true)]
    async fn dns_lookup_stops_at_connection_option_deadline() {
        let timeout = ClientDnsTimeout::new(Time::from_millis(37)).expect("positive timeout");
        let started = Instant::now();
        let pending = pending::<io::Result<vec::IntoIter<SocketAddr>>>();

        let result = lookup_first("broker.invalid:9092", timeout, pending).await;

        assert2::assert!(result.is_err());
        assert2::assert!(started.elapsed() == Duration::from_millis(37));
    }

    #[tokio::test]
    async fn custom_options_are_observable_on_initial_dial() {
        let live = ObservedAdminBroker::start(Duration::ZERO).await;
        let mut admin =
            AdminClient::connect_with_options(&[live.addr.to_string()], custom_admin_options())
                .await
                .unwrap();

        assert2::assert!(live.observed_custom_security_and_id());
        assert_custom_connect_timeout_is_stored(&admin);
        metadata_times_out_with_custom_request_timeout(&mut admin).await;

        let slow = ObservedAdminBroker::start(Duration::from_millis(300)).await;
        let result = timeout(
            Duration::from_secs(1),
            AdminClient::connect_with_options(&[slow.addr.to_string()], custom_admin_options()),
        )
        .await
        .expect("ApiVersions obeys the stored connect timeout");
        assert2::assert!(result.is_err());

        live.stop();
        slow.stop();
    }

    /// `connect_with_config` and `connect_controller_with_config` take the
    /// settings of `AdminClientConfig` onto the connection and the retry
    /// policy of the client, and an invalid setting stops the connect.
    #[tokio::test]
    async fn connect_with_config_carries_the_settings_and_refuses_an_invalid_one() {
        let live = ObservedAdminBroker::start(Duration::ZERO).await;
        let config = || AdminClientConfig {
            client_id: Some("tool-a".to_owned()),
            request_timeout: secs(45),
            default_api_timeout: Some(secs(90)),
            retries: 4,
            retry_backoff: Time::from_millis(30),
            retry_backoff_max: Time::from_millis(300),
            ..AdminClientConfig::default()
        };

        let admin = AdminClient::connect_with_config(&[live.addr.to_string()], config())
            .await
            .expect("admin connects");

        assert2::assert!(admin.options.client_id == "tool-a");
        assert2::assert!(admin.options.request_timeout == secs(45));
        assert2::assert!(
            admin.retry
                == retry::RetryPolicy {
                    timeout: Duration::from_secs(90),
                    initial_backoff: Duration::from_millis(30),
                    max_backoff: Duration::from_millis(300),
                    jitter: admin.retry.jitter,
                    max_retries: 4,
                }
        );
        live.stop();

        let controller = ObservedController::start(2).await;
        let on_controller =
            AdminClient::connect_controller_with_config(&[controller.addr.to_string()], config())
                .await
                .expect("controller admin connects");

        assert2::assert!(on_controller.conn.uses_controller_bootstrap());
        assert2::assert!(on_controller.options.client_id == "tool-a");
        assert2::assert!(on_controller.retry.max_retries == 4);

        let refused = AdminClient::connect_with_config(
            &["127.0.0.1:1".to_owned()],
            AdminClientConfig {
                request_timeout: secs(90),
                default_api_timeout: Some(secs(60)),
                ..AdminClientConfig::default()
            },
        )
        .await;

        assert2::assert!(matches!(refused, Err(AdminError::InvalidConfig(_))));
    }

    #[tokio::test]
    async fn connect_with_dns_timeout_preserves_admin_defaults() {
        let live = ObservedAdminBroker::start(Duration::ZERO).await;
        let timeout = ClientDnsTimeout::new(Time::from_millis(37)).expect("positive timeout");
        let admin = AdminClient::connect_with_dns_timeout(&[live.addr.to_string()], timeout)
            .await
            .expect("admin connects");

        assert2::assert!(admin.options.dns_timeout == timeout);
        assert2::assert!(admin.options.client_id.starts_with("adminclient-"));
        assert2::assert!(admin.options.socket_connection_setup_timeout == secs(10));
        assert2::assert!(admin.options.request_timeout == secs(30));
        live.stop();
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn admin_metadata_uses_a_learned_broker_after_the_only_seed_retires() {
        let healthy_port = Arc::new(AtomicU16::new(0));
        let healthy_calls = Arc::new(AtomicUsize::new(0));
        let h_port = Arc::clone(&healthy_port);
        let h_calls = Arc::clone(&healthy_calls);
        let healthy = MockBroker::start(move |api_key, version, _correlation_id, _body| {
            if api_key == api_versions_request::API_KEY {
                return Some(admin_mock_api_versions());
            }
            if api_key == metadata_request::API_KEY {
                h_calls.fetch_add(1, Ordering::SeqCst);
                return Some(metadata_mock_response(
                    version,
                    7,
                    h_port.load(Ordering::SeqCst),
                ));
            }
            None
        })
        .await
        .unwrap();
        healthy_port.store(healthy.addr.port(), Ordering::SeqCst);

        let advertised_port = Arc::clone(&healthy_port);
        let seed = MockBroker::start(move |api_key, version, _correlation_id, _body| {
            if api_key == api_versions_request::API_KEY {
                return Some(admin_mock_api_versions());
            }
            (api_key == metadata_request::API_KEY)
                .then(|| metadata_mock_response(version, 7, advertised_port.load(Ordering::SeqCst)))
        })
        .await
        .unwrap();
        let admin = AdminClient::connect(&[seed.addr.to_string()])
            .await
            .expect("admin connects to its seed");
        let first = admin
            .conn
            .send(metadata_request::MetadataRequest::default())
            .await
            .expect("seed provides last-known broker metadata");
        assert2::assert!(first.brokers[0].node_id == 7);

        let seed_addr = seed.addr;
        seed.stop();
        wait_for_admin_mock_closed(seed_addr).await;

        let refreshed = admin
            .conn
            .send(metadata_request::MetadataRequest::default())
            .await
            .expect("admin refreshes through the learned broker");
        assert2::assert!(refreshed.brokers[0].node_id == 7);
        assert2::assert!(
            healthy_calls.load(Ordering::SeqCst) >= 1,
            "recovery must issue metadata through the learned broker"
        );
    }

    #[tokio::test]
    async fn recovering_connection_enforces_required_api_version() {
        let broker = MockBroker::start(|api_key, _version, _correlation_id, _body| {
            (api_key == api_versions_request::API_KEY)
                .then(|| admin_mock_api_versions_with_describe_cluster(1))
        })
        .await
        .unwrap();
        let admin = AdminClient::connect(&[broker.addr.to_string()])
            .await
            .expect("admin connects");

        let result = admin
            .conn
            .send_at_least(DescribeClusterRequest::default(), 2)
            .await;

        assert2::assert!(matches!(
            result,
            Err(AdminError::Transport(ClientError::IncompatibleVersion {
                api_key: describe_cluster_request::API_KEY,
                broker_max: 1,
                client_min: 2,
                ..
            }))
        ));
        broker.stop();
    }

    #[tokio::test]
    async fn controller_bootstrap_discovers_and_connects_active_controller() {
        let controller = ObservedController::start(2).await;

        let admin = AdminClient::connect_controller(&[controller.addr.to_string()])
            .await
            .unwrap_or_else(|error| {
                panic!(
                    "controller bootstrap: {error}; connections={}; endpoint_types={:?}",
                    controller.connections.load(Ordering::SeqCst),
                    controller.endpoint_types.lock().unwrap(),
                )
            });

        assert2::assert!(admin.conn.uses_controller_bootstrap());
        assert2::assert!(controller.connections.load(Ordering::SeqCst) == 2);
        assert2::assert!(*controller.endpoint_types.lock().unwrap() == vec![2]);
    }

    #[tokio::test]
    async fn controller_bootstrap_requires_endpoint_type_version() {
        let controller = ObservedController::start(0).await;

        let result = AdminClient::connect_controller(&[controller.addr.to_string()]).await;

        assert2::assert!(matches!(
            result,
            Err(AdminError::Transport(ClientError::IncompatibleVersion {
                api_key: describe_cluster_request::API_KEY,
                broker_max: 0,
                client_min: 1,
                ..
            }))
        ));
        assert2::assert!(controller.endpoint_types.lock().unwrap().is_empty());
    }

    #[tokio::test]
    async fn secured_dns_timeout_preserves_security_and_admin_defaults() {
        let live = ObservedAdminBroker::start(Duration::ZERO).await;
        let timeout = ClientDnsTimeout::new(Time::from_millis(37)).expect("positive timeout");
        let security = ClientSecurity {
            protocol: ListenerProtocol::SaslPlaintext,
            tls: None,
            sasl: Some(SaslCredentials::Plain {
                username: "u".into(),
                password: "p".into(),
            }),
            sasl_host: Some("broker.example".into()),
        };
        let admin = AdminClient::connect_secured_with_dns_timeout(
            &[live.addr.to_string()],
            Some(security),
            timeout,
        )
        .await
        .expect("secured admin connects");

        assert2::assert!(admin.options.dns_timeout == timeout);
        assert2::assert!(admin.options.security.is_some());
        assert2::assert!(admin.options.client_id.starts_with("adminclient-"));
        assert2::assert!(admin.options.socket_connection_setup_timeout == secs(10));
        assert2::assert!(admin.options.request_timeout == secs(30));
        live.stop();
    }

    #[tokio::test]
    async fn custom_options_are_observable_on_controller_reconnect() {
        let bootstrap = ObservedAdminBroker::start(Duration::ZERO).await;
        let controller = ObservedAdminBroker::start(Duration::ZERO).await;
        let slow = ObservedAdminBroker::start(Duration::from_millis(300)).await;
        let mut admin = AdminClient::connect_with_options(
            &[bootstrap.addr.to_string()],
            custom_admin_options(),
        )
        .await
        .unwrap();

        admin.reconnect(&controller.addr.to_string()).await.unwrap();
        assert2::assert!(controller.observed_custom_security_and_id());
        assert_custom_connect_timeout_is_stored(&admin);
        metadata_times_out_with_custom_request_timeout(&mut admin).await;
        let result = timeout(
            Duration::from_secs(1),
            admin.reconnect(&slow.addr.to_string()),
        )
        .await
        .expect("reconnected ApiVersions obeys the stored request timeout");
        assert2::assert!(result.is_err());

        bootstrap.stop();
        controller.stop();
        slow.stop();
    }

    #[tokio::test]
    async fn custom_options_are_observable_on_bootstrap_reconnect() {
        let slow = ObservedAdminBroker::start(Duration::from_millis(300)).await;
        let live = ObservedAdminBroker::start(Duration::ZERO).await;
        let bootstrap = [slow.addr.to_string(), live.addr.to_string()];
        let mut admin = timeout(
            Duration::from_secs(1),
            AdminClient::connect_with_options(&bootstrap, custom_admin_options()),
        )
        .await
        .expect("custom initial timeout advances to next bootstrap")
        .unwrap();

        timeout(Duration::from_secs(1), admin.reconnect_bootstrap())
            .await
            .expect("custom reconnect timeout advances to next bootstrap")
            .unwrap();
        assert2::assert!(live.connections.load(Ordering::SeqCst) == 2);
        assert2::assert!(live.observed_custom_security_and_id());
        assert_custom_connect_timeout_is_stored(&admin);
        metadata_times_out_with_custom_request_timeout(&mut admin).await;

        slow.stop();
        live.stop();
    }

    #[test]
    fn existing_connectors_keep_admin_defaults() {
        let options = AdminClient::opts(None).unwrap();

        assert2::assert!(options.client_id.starts_with("adminclient-"));
        assert2::assert!(options.socket_connection_setup_timeout == secs(10));
        assert2::assert!(options.connections_max_idle == krabka_units::minutes(5));
        assert2::assert!(options.request_timeout == secs(30));
        assert2::assert!(options.security.is_none());
    }

    #[test]
    fn evacuation_methods_are_available_on_trait_objects() {
        fn compile_calls(client: &mut dyn AdminClientLike) {
            drop(client.describe_partition_assignments(&["orders"]));
            drop(client.list_partition_reassignments(&BTreeMap::new(), secs(30)));
            drop(client.alter_partition_assignments(&BTreeMap::new(), secs(30)));
            drop(client.unregister_broker(7));
        }

        hint::black_box(compile_calls as fn(&mut dyn AdminClientLike));
    }
}