kafkit-client 0.1.4

//! Configuration types used to build clients.
//!
//! The direct config structs are useful when you want full control. For day to
//! day code, [`KafkaClient`](crate::KafkaClient) and its builders are a little
//! more convenient.
//!
//! ```no_run
//! use std::time::Duration;
//! use kafkit_client::{ProducerConfig, SecurityProtocol};
//!
//! let config = ProducerConfig::new("localhost:9092")
//!     .with_client_id("orders-writer")
//!     .with_security_protocol(SecurityProtocol::Plaintext)
//!     .with_linger(Duration::from_millis(5));
//! ```
//!
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;

use kafka_protocol::records::Compression;

use crate::constants::{READ_COMMITTED, READ_UNCOMMITTED};
use crate::network::{TcpConnector, TokioTcpConnector};
use crate::types::ConsumerRebalanceListener;

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
/// Security Protocol.
pub enum SecurityProtocol {
    /// Plaintext.
    Plaintext,
    /// Ssl.
    Ssl,
    /// Sasl plaintext.
    SaslPlaintext,
    /// Sasl ssl.
    SaslSsl,
}

impl SecurityProtocol {
    /// Returns whether tls.
    pub fn uses_tls(self) -> bool {
        matches!(self, Self::Ssl | Self::SaslSsl)
    }

    /// Returns whether sasl.
    pub fn uses_sasl(self) -> bool {
        matches!(self, Self::SaslPlaintext | Self::SaslSsl)
    }

    fn with_sasl(self) -> Self {
        if self.uses_tls() {
            Self::SaslSsl
        } else {
            Self::SaslPlaintext
        }
    }
}

#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
/// Sasl Mechanism.
pub enum SaslMechanism {
    #[default]
    /// Plain.
    Plain,
    /// Scram sha256.
    ScramSha256,
    /// Scram sha512.
    ScramSha512,
}

impl SaslMechanism {
    /// Returns the Kafka-style name for this value.
    pub fn as_str(self) -> &'static str {
        match self {
            Self::Plain => "PLAIN",
            Self::ScramSha256 => "SCRAM-SHA-256",
            Self::ScramSha512 => "SCRAM-SHA-512",
        }
    }

    /// Returns whether scram.
    pub fn is_scram(self) -> bool {
        matches!(self, Self::ScramSha256 | Self::ScramSha512)
    }

    /// Scram Type.
    pub fn scram_type(self) -> Option<i8> {
        match self {
            Self::Plain => None,
            Self::ScramSha256 => Some(1),
            Self::ScramSha512 => Some(2),
        }
    }
}

#[derive(Debug, Clone, Default, PartialEq, Eq)]
/// Sasl Config.
pub struct SaslConfig {
    /// Mechanism.
    pub mechanism: SaslMechanism,
    /// Username.
    pub username: Option<String>,
    /// Password.
    pub password: Option<String>,
    /// Authorization Id.
    pub authorization_id: Option<String>,
}

impl SaslConfig {
    /// Plain.
    pub fn plain(username: impl Into<String>, password: impl Into<String>) -> Self {
        Self {
            mechanism: SaslMechanism::Plain,
            username: Some(username.into()),
            password: Some(password.into()),
            authorization_id: None,
        }
    }

    /// Scram Sha 256.
    pub fn scram_sha_256(username: impl Into<String>, password: impl Into<String>) -> Self {
        Self {
            mechanism: SaslMechanism::ScramSha256,
            username: Some(username.into()),
            password: Some(password.into()),
            authorization_id: None,
        }
    }

    /// Scram Sha 512.
    pub fn scram_sha_512(username: impl Into<String>, password: impl Into<String>) -> Self {
        Self {
            mechanism: SaslMechanism::ScramSha512,
            username: Some(username.into()),
            password: Some(password.into()),
            authorization_id: None,
        }
    }

    /// Sets authorization id and returns the updated value.
    pub fn with_authorization_id(mut self, authorization_id: impl Into<String>) -> Self {
        self.authorization_id = Some(authorization_id.into());
        self
    }
}

#[derive(Debug, Clone, Default, PartialEq, Eq)]
/// Tls Config.
pub struct TlsConfig {
    /// Ca Cert Path.
    pub ca_cert_path: Option<PathBuf>,
    /// Client Cert Path.
    pub client_cert_path: Option<PathBuf>,
    /// Client Key Path.
    pub client_key_path: Option<PathBuf>,
    /// Server Name.
    pub server_name: Option<String>,
}

impl TlsConfig {
    /// Creates a new value.
    pub fn new() -> Self {
        Self::default()
    }

    /// Sets ca cert path and returns the updated value.
    pub fn with_ca_cert_path(mut self, path: impl Into<PathBuf>) -> Self {
        self.ca_cert_path = Some(path.into());
        self
    }

    /// Sets client cert path and returns the updated value.
    pub fn with_client_cert_path(mut self, path: impl Into<PathBuf>) -> Self {
        self.client_cert_path = Some(path.into());
        self
    }

    /// Sets client key path and returns the updated value.
    pub fn with_client_key_path(mut self, path: impl Into<PathBuf>) -> Self {
        self.client_key_path = Some(path.into());
        self
    }

    /// Sets server name and returns the updated value.
    pub fn with_server_name(mut self, server_name: impl Into<String>) -> Self {
        self.server_name = Some(server_name.into());
        self
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
/// Producer Partitioner.
pub enum ProducerPartitioner {
    /// Default.
    Default,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
/// Producer Compression.
pub enum ProducerCompression {
    /// None.
    None,
    /// Gzip.
    Gzip,
    /// Snappy.
    Snappy,
    /// Lz4.
    Lz4,
    /// Zstd.
    Zstd,
}

impl From<ProducerCompression> for Compression {
    fn from(value: ProducerCompression) -> Self {
        match value {
            ProducerCompression::None => Compression::None,
            ProducerCompression::Gzip => Compression::Gzip,
            ProducerCompression::Snappy => Compression::Snappy,
            ProducerCompression::Lz4 => Compression::Lz4,
            ProducerCompression::Zstd => Compression::Zstd,
        }
    }
}

#[derive(Debug, Clone)]
/// Producer Config.
pub struct ProducerConfig {
    /// Bootstrap Servers.
    pub bootstrap_servers: Vec<String>,
    /// Client Id.
    pub client_id: String,
    /// Security Protocol.
    pub security_protocol: SecurityProtocol,
    /// Tls.
    pub tls: TlsConfig,
    /// Sasl.
    pub sasl: SaslConfig,
    /// Acks.
    pub acks: i16,
    /// Enable Idempotence.
    pub enable_idempotence: bool,
    /// Partitioner.
    pub partitioner: ProducerPartitioner,
    /// Compression.
    pub compression: ProducerCompression,
    /// Batch Size.
    pub batch_size: usize,
    /// Linger.
    pub linger: Duration,
    /// Delivery Timeout.
    pub delivery_timeout: Duration,
    /// Request Timeout.
    pub request_timeout: Duration,
    /// Metadata Max Age.
    pub metadata_max_age: Duration,
    /// Retry Backoff.
    pub retry_backoff: Duration,
    /// Max Retries.
    pub max_retries: usize,
    /// Max In Flight Requests Per Connection.
    pub max_in_flight_requests_per_connection: usize,
    /// Transactional Id.
    pub transactional_id: Option<String>,
    /// Transaction Timeout.
    pub transaction_timeout: Duration,
    /// TCP connector.
    pub tcp_connector: Arc<dyn TcpConnector>,
}

impl ProducerConfig {
    /// Creates a new value.
    pub fn new(bootstrap_server: impl Into<String>) -> Self {
        Self {
            bootstrap_servers: vec![bootstrap_server.into()],
            client_id: "rust-producer".to_owned(),
            security_protocol: SecurityProtocol::Plaintext,
            tls: TlsConfig::default(),
            sasl: SaslConfig::default(),
            acks: 1,
            enable_idempotence: false,
            partitioner: ProducerPartitioner::Default,
            compression: ProducerCompression::None,
            batch_size: 16 * 1024,
            linger: Duration::ZERO,
            delivery_timeout: Duration::from_secs(120),
            request_timeout: Duration::from_secs(5),
            metadata_max_age: Duration::from_secs(30),
            retry_backoff: Duration::from_millis(250),
            max_retries: 3,
            max_in_flight_requests_per_connection: 5,
            transactional_id: None,
            transaction_timeout: Duration::from_secs(30),
            tcp_connector: Arc::new(TokioTcpConnector),
        }
    }

    /// Sets client id and returns the updated value.
    pub fn with_client_id(mut self, client_id: impl Into<String>) -> Self {
        self.client_id = client_id.into();
        self
    }

    /// Sets bootstrap servers and returns the updated value.
    pub fn with_bootstrap_servers(
        mut self,
        servers: impl IntoIterator<Item = impl Into<String>>,
    ) -> Self {
        self.bootstrap_servers = servers.into_iter().map(Into::into).collect();
        self
    }

    /// Sets security protocol and returns the updated value.
    pub fn with_security_protocol(mut self, security_protocol: SecurityProtocol) -> Self {
        self.security_protocol = security_protocol;
        self
    }

    /// Sets tls and returns the updated value.
    pub fn with_tls(mut self, tls: TlsConfig) -> Self {
        self.security_protocol = if self.security_protocol.uses_sasl() {
            SecurityProtocol::SaslSsl
        } else {
            SecurityProtocol::Ssl
        };
        self.tls = tls;
        self
    }

    /// Sets tls ca cert path and returns the updated value.
    pub fn with_tls_ca_cert_path(mut self, path: impl Into<PathBuf>) -> Self {
        self.security_protocol = if self.security_protocol.uses_sasl() {
            SecurityProtocol::SaslSsl
        } else {
            SecurityProtocol::Ssl
        };
        self.tls = self.tls.with_ca_cert_path(path);
        self
    }

    /// Sets tls client auth paths and returns the updated value.
    pub fn with_tls_client_auth_paths(
        mut self,
        cert_path: impl Into<PathBuf>,
        key_path: impl Into<PathBuf>,
    ) -> Self {
        self.security_protocol = if self.security_protocol.uses_sasl() {
            SecurityProtocol::SaslSsl
        } else {
            SecurityProtocol::Ssl
        };
        self.tls = self
            .tls
            .with_client_cert_path(cert_path)
            .with_client_key_path(key_path);
        self
    }

    /// Sets tls server name and returns the updated value.
    pub fn with_tls_server_name(mut self, server_name: impl Into<String>) -> Self {
        self.security_protocol = if self.security_protocol.uses_sasl() {
            SecurityProtocol::SaslSsl
        } else {
            SecurityProtocol::Ssl
        };
        self.tls = self.tls.with_server_name(server_name);
        self
    }

    /// Sets sasl and returns the updated value.
    pub fn with_sasl(mut self, sasl: SaslConfig) -> Self {
        self.security_protocol = self.security_protocol.with_sasl();
        self.sasl = sasl;
        self
    }

    /// Sets sasl plain and returns the updated value.
    pub fn with_sasl_plain(self, username: impl Into<String>, password: impl Into<String>) -> Self {
        self.with_sasl(SaslConfig::plain(username, password))
    }

    /// Sets sasl scram sha 256 and returns the updated value.
    pub fn with_sasl_scram_sha_256(
        self,
        username: impl Into<String>,
        password: impl Into<String>,
    ) -> Self {
        self.with_sasl(SaslConfig::scram_sha_256(username, password))
    }

    /// Sets sasl scram sha 512 and returns the updated value.
    pub fn with_sasl_scram_sha_512(
        self,
        username: impl Into<String>,
        password: impl Into<String>,
    ) -> Self {
        self.with_sasl(SaslConfig::scram_sha_512(username, password))
    }

    /// Sets acks and returns the updated value.
    pub fn with_acks(mut self, acks: i16) -> Self {
        self.acks = acks;
        self
    }

    /// Sets enable idempotence and returns the updated value.
    pub fn with_enable_idempotence(mut self, enable_idempotence: bool) -> Self {
        self.enable_idempotence = enable_idempotence;
        if enable_idempotence {
            self.acks = -1;
            self.max_retries = self.max_retries.max(1);
        }
        self
    }

    /// Sets partitioner and returns the updated value.
    pub fn with_partitioner(mut self, partitioner: ProducerPartitioner) -> Self {
        self.partitioner = partitioner;
        self
    }

    /// Sets compression and returns the updated value.
    pub fn with_compression(mut self, compression: ProducerCompression) -> Self {
        self.compression = compression;
        self
    }

    /// Sets batch size and returns the updated value.
    pub fn with_batch_size(mut self, batch_size: usize) -> Self {
        self.batch_size = batch_size.max(1);
        self
    }

    /// Sets linger and returns the updated value.
    pub fn with_linger(mut self, linger: Duration) -> Self {
        self.linger = linger;
        self
    }

    /// Sets delivery timeout and returns the updated value.
    pub fn with_delivery_timeout(mut self, delivery_timeout: Duration) -> Self {
        self.delivery_timeout = delivery_timeout;
        self
    }

    /// Sets request timeout and returns the updated value.
    pub fn with_request_timeout(mut self, request_timeout: Duration) -> Self {
        self.request_timeout = request_timeout;
        self
    }

    /// Sets metadata max age and returns the updated value.
    pub fn with_metadata_max_age(mut self, metadata_max_age: Duration) -> Self {
        self.metadata_max_age = metadata_max_age;
        self
    }

    /// Sets retry backoff and returns the updated value.
    pub fn with_retry_backoff(mut self, retry_backoff: Duration) -> Self {
        self.retry_backoff = retry_backoff;
        self
    }

    /// Sets max retries and returns the updated value.
    pub fn with_max_retries(mut self, max_retries: usize) -> Self {
        self.max_retries = max_retries;
        self
    }

    /// Sets max in-flight requests per broker connection and returns the updated value.
    pub fn with_max_in_flight_requests_per_connection(mut self, max_in_flight: usize) -> Self {
        self.max_in_flight_requests_per_connection = max_in_flight.max(1);
        self
    }

    /// Sets transactional id and returns the updated value.
    pub fn with_transactional_id(mut self, transactional_id: impl Into<String>) -> Self {
        self.transactional_id = Some(transactional_id.into());
        self.acks = -1;
        self.enable_idempotence = true;
        self
    }

    /// Sets transaction timeout and returns the updated value.
    pub fn with_transaction_timeout(mut self, transaction_timeout: Duration) -> Self {
        self.transaction_timeout = transaction_timeout;
        self
    }

    /// Sets TCP connector and returns the updated value.
    pub fn with_tcp_connector(mut self, tcp_connector: Arc<dyn TcpConnector>) -> Self {
        self.tcp_connector = tcp_connector;
        self
    }

    /// Returns whether transactional.
    pub fn is_transactional(&self) -> bool {
        self.transactional_id.is_some()
    }

    /// Returns whether idempotent.
    pub fn is_idempotent(&self) -> bool {
        self.enable_idempotence
    }
}

/// Alias for client config.
pub type ClientConfig = ProducerConfig;

#[derive(Debug, Clone)]
/// Admin Config.
pub struct AdminConfig {
    /// Bootstrap Servers.
    pub bootstrap_servers: Vec<String>,
    /// Client Id.
    pub client_id: String,
    /// Security Protocol.
    pub security_protocol: SecurityProtocol,
    /// Tls.
    pub tls: TlsConfig,
    /// Sasl.
    pub sasl: SaslConfig,
    /// Request Timeout.
    pub request_timeout: Duration,
    /// TCP connector.
    pub tcp_connector: Arc<dyn TcpConnector>,
}

impl AdminConfig {
    /// Creates a new value.
    pub fn new(bootstrap_server: impl Into<String>) -> Self {
        Self {
            bootstrap_servers: vec![bootstrap_server.into()],
            client_id: "rust-admin".to_owned(),
            security_protocol: SecurityProtocol::Plaintext,
            tls: TlsConfig::default(),
            sasl: SaslConfig::default(),
            request_timeout: Duration::from_secs(5),
            tcp_connector: Arc::new(TokioTcpConnector),
        }
    }

    /// Sets client id and returns the updated value.
    pub fn with_client_id(mut self, client_id: impl Into<String>) -> Self {
        self.client_id = client_id.into();
        self
    }

    /// Sets bootstrap servers and returns the updated value.
    pub fn with_bootstrap_servers(
        mut self,
        servers: impl IntoIterator<Item = impl Into<String>>,
    ) -> Self {
        self.bootstrap_servers = servers.into_iter().map(Into::into).collect();
        self
    }

    /// Sets security protocol and returns the updated value.
    pub fn with_security_protocol(mut self, security_protocol: SecurityProtocol) -> Self {
        self.security_protocol = security_protocol;
        self
    }

    /// Sets tls and returns the updated value.
    pub fn with_tls(mut self, tls: TlsConfig) -> Self {
        self.security_protocol = if self.security_protocol.uses_sasl() {
            SecurityProtocol::SaslSsl
        } else {
            SecurityProtocol::Ssl
        };
        self.tls = tls;
        self
    }

    /// Sets tls ca cert path and returns the updated value.
    pub fn with_tls_ca_cert_path(mut self, path: impl Into<PathBuf>) -> Self {
        self.security_protocol = if self.security_protocol.uses_sasl() {
            SecurityProtocol::SaslSsl
        } else {
            SecurityProtocol::Ssl
        };
        self.tls = self.tls.with_ca_cert_path(path);
        self
    }

    /// Sets tls client auth paths and returns the updated value.
    pub fn with_tls_client_auth_paths(
        mut self,
        cert_path: impl Into<PathBuf>,
        key_path: impl Into<PathBuf>,
    ) -> Self {
        self.security_protocol = if self.security_protocol.uses_sasl() {
            SecurityProtocol::SaslSsl
        } else {
            SecurityProtocol::Ssl
        };
        self.tls = self
            .tls
            .with_client_cert_path(cert_path)
            .with_client_key_path(key_path);
        self
    }

    /// Sets tls server name and returns the updated value.
    pub fn with_tls_server_name(mut self, server_name: impl Into<String>) -> Self {
        self.security_protocol = if self.security_protocol.uses_sasl() {
            SecurityProtocol::SaslSsl
        } else {
            SecurityProtocol::Ssl
        };
        self.tls = self.tls.with_server_name(server_name);
        self
    }

    /// Sets sasl and returns the updated value.
    pub fn with_sasl(mut self, sasl: SaslConfig) -> Self {
        self.security_protocol = self.security_protocol.with_sasl();
        self.sasl = sasl;
        self
    }

    /// Sets sasl plain and returns the updated value.
    pub fn with_sasl_plain(self, username: impl Into<String>, password: impl Into<String>) -> Self {
        self.with_sasl(SaslConfig::plain(username, password))
    }

    /// Sets sasl scram sha 256 and returns the updated value.
    pub fn with_sasl_scram_sha_256(
        self,
        username: impl Into<String>,
        password: impl Into<String>,
    ) -> Self {
        self.with_sasl(SaslConfig::scram_sha_256(username, password))
    }

    /// Sets sasl scram sha 512 and returns the updated value.
    pub fn with_sasl_scram_sha_512(
        self,
        username: impl Into<String>,
        password: impl Into<String>,
    ) -> Self {
        self.with_sasl(SaslConfig::scram_sha_512(username, password))
    }

    /// Sets request timeout and returns the updated value.
    pub fn with_request_timeout(mut self, request_timeout: Duration) -> Self {
        self.request_timeout = request_timeout;
        self
    }

    /// Sets TCP connector and returns the updated value.
    pub fn with_tcp_connector(mut self, tcp_connector: Arc<dyn TcpConnector>) -> Self {
        self.tcp_connector = tcp_connector;
        self
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
/// Auto Offset Reset.
pub enum AutoOffsetReset {
    /// Earliest.
    Earliest,
    /// Latest.
    Latest,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
/// Isolation Level.
pub enum IsolationLevel {
    /// Read uncommitted.
    ReadUncommitted,
    /// Read committed.
    ReadCommitted,
}

impl IsolationLevel {
    /// As Protocol Value.
    pub fn as_protocol_value(self) -> i8 {
        match self {
            Self::ReadUncommitted => READ_UNCOMMITTED,
            Self::ReadCommitted => READ_COMMITTED,
        }
    }
}

#[derive(Debug, Clone)]
/// Consumer Config.
pub struct ConsumerConfig {
    /// Bootstrap Servers.
    pub bootstrap_servers: Vec<String>,
    /// Client Id.
    pub client_id: String,
    /// Group Id.
    pub group_id: String,
    /// Security Protocol.
    pub security_protocol: SecurityProtocol,
    /// Tls.
    pub tls: TlsConfig,
    /// Sasl.
    pub sasl: SaslConfig,
    /// Request Timeout.
    pub request_timeout: Duration,
    /// Metadata Max Age.
    pub metadata_max_age: Duration,
    /// Retry Backoff.
    pub retry_backoff: Duration,
    /// Max Retries.
    pub max_retries: usize,
    /// Rebalance Timeout.
    pub rebalance_timeout: Duration,
    /// Fetch Max Wait.
    pub fetch_max_wait: Duration,
    /// Fetch Min Bytes.
    pub fetch_min_bytes: i32,
    /// Fetch Max Bytes.
    pub fetch_max_bytes: i32,
    /// Partition Max Bytes.
    pub partition_max_bytes: i32,
    /// Auto Offset Reset.
    pub auto_offset_reset: AutoOffsetReset,
    /// Isolation Level.
    pub isolation_level: IsolationLevel,
    /// Enable Auto Commit.
    pub enable_auto_commit: bool,
    /// Auto Commit Interval.
    pub auto_commit_interval: Duration,
    /// Server Assignor.
    pub server_assignor: Option<String>,
    /// Rack Id.
    pub rack_id: Option<String>,
    /// Instance Id.
    pub instance_id: Option<String>,
    /// Rebalance Listener.
    pub rebalance_listener: Option<ConsumerRebalanceListener>,
    /// TCP connector.
    pub tcp_connector: Arc<dyn TcpConnector>,
}

impl ConsumerConfig {
    /// Creates a new value.
    pub fn new(bootstrap_server: impl Into<String>, group_id: impl Into<String>) -> Self {
        Self {
            bootstrap_servers: vec![bootstrap_server.into()],
            client_id: "rust-consumer".to_owned(),
            group_id: group_id.into(),
            security_protocol: SecurityProtocol::Plaintext,
            tls: TlsConfig::default(),
            sasl: SaslConfig::default(),
            request_timeout: Duration::from_secs(5),
            metadata_max_age: Duration::from_secs(30),
            retry_backoff: Duration::from_millis(250),
            max_retries: 3,
            rebalance_timeout: Duration::from_secs(30),
            fetch_max_wait: Duration::from_millis(500),
            fetch_min_bytes: 1,
            fetch_max_bytes: 50 * 1024 * 1024,
            partition_max_bytes: 1024 * 1024,
            auto_offset_reset: AutoOffsetReset::Earliest,
            isolation_level: IsolationLevel::ReadUncommitted,
            enable_auto_commit: false,
            auto_commit_interval: Duration::from_secs(5),
            server_assignor: None,
            rack_id: None,
            instance_id: None,
            rebalance_listener: None,
            tcp_connector: Arc::new(TokioTcpConnector),
        }
    }

    /// Sets client id and returns the updated value.
    pub fn with_client_id(mut self, client_id: impl Into<String>) -> Self {
        self.client_id = client_id.into();
        self
    }

    /// Sets bootstrap servers and returns the updated value.
    pub fn with_bootstrap_servers(
        mut self,
        servers: impl IntoIterator<Item = impl Into<String>>,
    ) -> Self {
        self.bootstrap_servers = servers.into_iter().map(Into::into).collect();
        self
    }

    /// Sets security protocol and returns the updated value.
    pub fn with_security_protocol(mut self, security_protocol: SecurityProtocol) -> Self {
        self.security_protocol = security_protocol;
        self
    }

    /// Sets tls and returns the updated value.
    pub fn with_tls(mut self, tls: TlsConfig) -> Self {
        self.security_protocol = if self.security_protocol.uses_sasl() {
            SecurityProtocol::SaslSsl
        } else {
            SecurityProtocol::Ssl
        };
        self.tls = tls;
        self
    }

    /// Sets tls ca cert path and returns the updated value.
    pub fn with_tls_ca_cert_path(mut self, path: impl Into<PathBuf>) -> Self {
        self.security_protocol = if self.security_protocol.uses_sasl() {
            SecurityProtocol::SaslSsl
        } else {
            SecurityProtocol::Ssl
        };
        self.tls = self.tls.with_ca_cert_path(path);
        self
    }

    /// Sets tls client auth paths and returns the updated value.
    pub fn with_tls_client_auth_paths(
        mut self,
        cert_path: impl Into<PathBuf>,
        key_path: impl Into<PathBuf>,
    ) -> Self {
        self.security_protocol = if self.security_protocol.uses_sasl() {
            SecurityProtocol::SaslSsl
        } else {
            SecurityProtocol::Ssl
        };
        self.tls = self
            .tls
            .with_client_cert_path(cert_path)
            .with_client_key_path(key_path);
        self
    }

    /// Sets tls server name and returns the updated value.
    pub fn with_tls_server_name(mut self, server_name: impl Into<String>) -> Self {
        self.security_protocol = if self.security_protocol.uses_sasl() {
            SecurityProtocol::SaslSsl
        } else {
            SecurityProtocol::Ssl
        };
        self.tls = self.tls.with_server_name(server_name);
        self
    }

    /// Sets sasl and returns the updated value.
    pub fn with_sasl(mut self, sasl: SaslConfig) -> Self {
        self.security_protocol = self.security_protocol.with_sasl();
        self.sasl = sasl;
        self
    }

    /// Sets sasl plain and returns the updated value.
    pub fn with_sasl_plain(self, username: impl Into<String>, password: impl Into<String>) -> Self {
        self.with_sasl(SaslConfig::plain(username, password))
    }

    /// Sets sasl scram sha 256 and returns the updated value.
    pub fn with_sasl_scram_sha_256(
        self,
        username: impl Into<String>,
        password: impl Into<String>,
    ) -> Self {
        self.with_sasl(SaslConfig::scram_sha_256(username, password))
    }

    /// Sets sasl scram sha 512 and returns the updated value.
    pub fn with_sasl_scram_sha_512(
        self,
        username: impl Into<String>,
        password: impl Into<String>,
    ) -> Self {
        self.with_sasl(SaslConfig::scram_sha_512(username, password))
    }

    /// Sets request timeout and returns the updated value.
    pub fn with_request_timeout(mut self, request_timeout: Duration) -> Self {
        self.request_timeout = request_timeout;
        self
    }

    /// Sets metadata max age and returns the updated value.
    pub fn with_metadata_max_age(mut self, metadata_max_age: Duration) -> Self {
        self.metadata_max_age = metadata_max_age;
        self
    }

    /// Sets retry backoff and returns the updated value.
    pub fn with_retry_backoff(mut self, retry_backoff: Duration) -> Self {
        self.retry_backoff = retry_backoff;
        self
    }

    /// Sets max retries and returns the updated value.
    pub fn with_max_retries(mut self, max_retries: usize) -> Self {
        self.max_retries = max_retries;
        self
    }

    /// Sets rebalance timeout and returns the updated value.
    pub fn with_rebalance_timeout(mut self, rebalance_timeout: Duration) -> Self {
        self.rebalance_timeout = rebalance_timeout;
        self
    }

    /// Sets fetch max wait and returns the updated value.
    pub fn with_fetch_max_wait(mut self, fetch_max_wait: Duration) -> Self {
        self.fetch_max_wait = fetch_max_wait;
        self
    }

    /// Sets fetch min bytes and returns the updated value.
    pub fn with_fetch_min_bytes(mut self, fetch_min_bytes: i32) -> Self {
        self.fetch_min_bytes = fetch_min_bytes;
        self
    }

    /// Sets fetch max bytes and returns the updated value.
    pub fn with_fetch_max_bytes(mut self, fetch_max_bytes: i32) -> Self {
        self.fetch_max_bytes = fetch_max_bytes;
        self
    }

    /// Sets partition max bytes and returns the updated value.
    pub fn with_partition_max_bytes(mut self, partition_max_bytes: i32) -> Self {
        self.partition_max_bytes = partition_max_bytes;
        self
    }

    /// Sets auto offset reset and returns the updated value.
    pub fn with_auto_offset_reset(mut self, auto_offset_reset: AutoOffsetReset) -> Self {
        self.auto_offset_reset = auto_offset_reset;
        self
    }

    /// Sets isolation level and returns the updated value.
    pub fn with_isolation_level(mut self, isolation_level: IsolationLevel) -> Self {
        self.isolation_level = isolation_level;
        self
    }

    /// Sets enable auto commit and returns the updated value.
    pub fn with_enable_auto_commit(mut self, enable_auto_commit: bool) -> Self {
        self.enable_auto_commit = enable_auto_commit;
        self
    }

    /// Sets auto commit interval and returns the updated value.
    pub fn with_auto_commit_interval(mut self, auto_commit_interval: Duration) -> Self {
        self.auto_commit_interval = auto_commit_interval;
        self
    }

    /// Sets server assignor and returns the updated value.
    pub fn with_server_assignor(mut self, server_assignor: impl Into<String>) -> Self {
        self.server_assignor = Some(server_assignor.into());
        self
    }

    /// Sets rack id and returns the updated value.
    pub fn with_rack_id(mut self, rack_id: impl Into<String>) -> Self {
        self.rack_id = Some(rack_id.into());
        self
    }

    /// Sets instance id and returns the updated value.
    pub fn with_instance_id(mut self, instance_id: impl Into<String>) -> Self {
        self.instance_id = Some(instance_id.into());
        self
    }

    /// Sets rebalance listener and returns the updated value.
    pub fn with_rebalance_listener(mut self, listener: ConsumerRebalanceListener) -> Self {
        self.rebalance_listener = Some(listener);
        self
    }

    /// Sets rebalance callback and returns the updated value.
    pub fn with_rebalance_callback(
        self,
        callback: impl Fn(crate::types::ConsumerRebalanceEvent) + Send + Sync + 'static,
    ) -> Self {
        self.with_rebalance_listener(ConsumerRebalanceListener::new(callback))
    }

    /// Sets TCP connector and returns the updated value.
    pub fn with_tcp_connector(mut self, tcp_connector: Arc<dyn TcpConnector>) -> Self {
        self.tcp_connector = tcp_connector;
        self
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn producer_tls_builder_enables_ssl() {
        let config =
            ProducerConfig::new("localhost:9093").with_tls_ca_cert_path("/tmp/cluster-ca.pem");
        assert_eq!(config.security_protocol, SecurityProtocol::Ssl);
        assert_eq!(
            config.tls.ca_cert_path.as_deref(),
            Some(std::path::Path::new("/tmp/cluster-ca.pem"))
        );
    }

    #[test]
    fn producer_sasl_plain_builder_enables_sasl_plaintext() {
        let config = ProducerConfig::new("localhost:9092").with_sasl_plain("user-a", "secret-a");
        assert_eq!(config.security_protocol, SecurityProtocol::SaslPlaintext);
        assert_eq!(config.sasl.mechanism, SaslMechanism::Plain);
        assert_eq!(config.sasl.username.as_deref(), Some("user-a"));
        assert_eq!(config.sasl.password.as_deref(), Some("secret-a"));
    }

    #[test]
    fn producer_sasl_scram_builder_enables_sasl_plaintext() {
        let config =
            ProducerConfig::new("localhost:9092").with_sasl_scram_sha_256("user-a", "secret-a");
        assert_eq!(config.security_protocol, SecurityProtocol::SaslPlaintext);
        assert_eq!(config.sasl.mechanism, SaslMechanism::ScramSha256);
        assert_eq!(config.sasl.username.as_deref(), Some("user-a"));
        assert_eq!(config.sasl.password.as_deref(), Some("secret-a"));
    }

    #[test]
    fn producer_tls_and_sasl_builders_enable_sasl_ssl() {
        let config = ProducerConfig::new("localhost:9093")
            .with_sasl_plain("user-a", "secret-a")
            .with_tls_ca_cert_path("/tmp/cluster-ca.pem");
        assert_eq!(config.security_protocol, SecurityProtocol::SaslSsl);

        let config = ProducerConfig::new("localhost:9093")
            .with_tls_ca_cert_path("/tmp/cluster-ca.pem")
            .with_sasl_plain("user-a", "secret-a");
        assert_eq!(config.security_protocol, SecurityProtocol::SaslSsl);
    }

    #[test]
    fn producer_multi_broker_with_bootstrap_servers() {
        let config = ProducerConfig::new("host1:9092").with_bootstrap_servers([
            "host1:9092",
            "host2:9092",
            "host3:9092",
        ]);
        assert_eq!(
            config.bootstrap_servers,
            vec!["host1:9092", "host2:9092", "host3:9092"]
        );
    }

    #[test]
    fn producer_builder_records_tuning_and_transaction_options() {
        let config = ProducerConfig::new("localhost:9092")
            .with_client_id("producer-a")
            .with_security_protocol(SecurityProtocol::Ssl)
            .with_tls_client_auth_paths("/tmp/client.crt", "/tmp/client.key")
            .with_tls_server_name("kafka.internal")
            .with_acks(-1)
            .with_enable_idempotence(true)
            .with_partitioner(ProducerPartitioner::Default)
            .with_compression(ProducerCompression::Zstd)
            .with_batch_size(0)
            .with_linger(Duration::from_millis(50))
            .with_delivery_timeout(Duration::from_secs(10))
            .with_request_timeout(Duration::from_secs(2))
            .with_metadata_max_age(Duration::from_secs(60))
            .with_retry_backoff(Duration::from_millis(75))
            .with_max_retries(7)
            .with_transactional_id("tx-a")
            .with_transaction_timeout(Duration::from_secs(45));

        assert_eq!(config.client_id, "producer-a");
        assert_eq!(config.security_protocol, SecurityProtocol::Ssl);
        assert_eq!(
            config.tls.client_cert_path.as_deref(),
            Some(std::path::Path::new("/tmp/client.crt"))
        );
        assert_eq!(
            config.tls.client_key_path.as_deref(),
            Some(std::path::Path::new("/tmp/client.key"))
        );
        assert_eq!(config.tls.server_name.as_deref(), Some("kafka.internal"));
        assert_eq!(config.acks, -1);
        assert!(config.is_idempotent());
        assert_eq!(config.compression, ProducerCompression::Zstd);
        assert_eq!(config.batch_size, 1);
        assert_eq!(config.linger, Duration::from_millis(50));
        assert_eq!(config.delivery_timeout, Duration::from_secs(10));
        assert_eq!(config.request_timeout, Duration::from_secs(2));
        assert_eq!(config.metadata_max_age, Duration::from_secs(60));
        assert_eq!(config.retry_backoff, Duration::from_millis(75));
        assert_eq!(config.max_retries, 7);
        assert_eq!(config.transactional_id.as_deref(), Some("tx-a"));
        assert_eq!(config.transaction_timeout, Duration::from_secs(45));
        assert!(config.is_transactional());
    }

    #[test]
    fn consumer_tls_builder_keeps_override_server_name() {
        let config = ConsumerConfig::new("localhost:9093", "group-a")
            .with_tls(TlsConfig::new().with_server_name("kafka.internal"));
        assert_eq!(config.security_protocol, SecurityProtocol::Ssl);
        assert_eq!(config.tls.server_name.as_deref(), Some("kafka.internal"));
    }

    #[test]
    fn consumer_multi_broker_with_bootstrap_servers() {
        let config = ConsumerConfig::new("host1:9092", "group-a")
            .with_bootstrap_servers(["host1:9092", "host2:9092"]);
        assert_eq!(config.bootstrap_servers, vec!["host1:9092", "host2:9092"]);
    }

    #[test]
    fn admin_tls_builder_records_client_auth_paths() {
        let config = AdminConfig::new("localhost:9093")
            .with_tls_client_auth_paths("/tmp/client.crt", "/tmp/client.key");
        assert_eq!(config.security_protocol, SecurityProtocol::Ssl);
        assert_eq!(
            config.tls.client_cert_path.as_deref(),
            Some(std::path::Path::new("/tmp/client.crt"))
        );
        assert_eq!(
            config.tls.client_key_path.as_deref(),
            Some(std::path::Path::new("/tmp/client.key"))
        );
    }

    #[test]
    fn admin_and_consumer_builders_record_remaining_options() {
        let admin = AdminConfig::new("localhost:9092")
            .with_client_id("admin-a")
            .with_bootstrap_servers(["host-a:9092", "host-b:9092"])
            .with_security_protocol(SecurityProtocol::Plaintext)
            .with_tls(TlsConfig::new().with_ca_cert_path("/tmp/ca.pem"))
            .with_sasl_scram_sha_512("user-a", "secret-a")
            .with_request_timeout(Duration::from_secs(8));

        assert_eq!(admin.bootstrap_servers, vec!["host-a:9092", "host-b:9092"]);
        assert_eq!(admin.client_id, "admin-a");
        assert_eq!(admin.security_protocol, SecurityProtocol::SaslSsl);
        assert_eq!(
            admin.tls.ca_cert_path.as_deref(),
            Some(std::path::Path::new("/tmp/ca.pem"))
        );
        assert_eq!(admin.sasl.mechanism, SaslMechanism::ScramSha512);
        assert_eq!(admin.request_timeout, Duration::from_secs(8));

        let consumer = ConsumerConfig::new("localhost:9092", "group-a")
            .with_client_id("consumer-a")
            .with_security_protocol(SecurityProtocol::Ssl)
            .with_tls_ca_cert_path("/tmp/ca.pem")
            .with_tls_client_auth_paths("/tmp/client.crt", "/tmp/client.key")
            .with_sasl_scram_sha_512("user-b", "secret-b")
            .with_request_timeout(Duration::from_secs(3))
            .with_metadata_max_age(Duration::from_secs(4))
            .with_retry_backoff(Duration::from_millis(5))
            .with_max_retries(6)
            .with_rebalance_timeout(Duration::from_secs(7))
            .with_fetch_max_wait(Duration::from_millis(8))
            .with_fetch_min_bytes(9)
            .with_fetch_max_bytes(10)
            .with_partition_max_bytes(11)
            .with_auto_offset_reset(AutoOffsetReset::Latest)
            .with_isolation_level(IsolationLevel::ReadCommitted)
            .with_enable_auto_commit(true)
            .with_auto_commit_interval(Duration::from_secs(12))
            .with_server_assignor("range")
            .with_rack_id("rack-a")
            .with_instance_id("instance-a");

        assert_eq!(consumer.client_id, "consumer-a");
        assert_eq!(consumer.security_protocol, SecurityProtocol::SaslSsl);
        assert_eq!(consumer.sasl.mechanism, SaslMechanism::ScramSha512);
        assert_eq!(consumer.request_timeout, Duration::from_secs(3));
        assert_eq!(consumer.metadata_max_age, Duration::from_secs(4));
        assert_eq!(consumer.retry_backoff, Duration::from_millis(5));
        assert_eq!(consumer.max_retries, 6);
        assert_eq!(consumer.rebalance_timeout, Duration::from_secs(7));
        assert_eq!(consumer.fetch_max_wait, Duration::from_millis(8));
        assert_eq!(consumer.fetch_min_bytes, 9);
        assert_eq!(consumer.fetch_max_bytes, 10);
        assert_eq!(consumer.partition_max_bytes, 11);
        assert_eq!(consumer.auto_offset_reset, AutoOffsetReset::Latest);
        assert_eq!(consumer.isolation_level, IsolationLevel::ReadCommitted);
        assert!(consumer.enable_auto_commit);
        assert_eq!(consumer.auto_commit_interval, Duration::from_secs(12));
        assert_eq!(consumer.server_assignor.as_deref(), Some("range"));
        assert_eq!(consumer.rack_id.as_deref(), Some("rack-a"));
        assert_eq!(consumer.instance_id.as_deref(), Some("instance-a"));
    }
}