datum-mq 0.10.8

use std::{
    collections::{BTreeMap, BTreeSet, HashMap, VecDeque},
    sync::{
        Arc, Condvar, Mutex, OnceLock,
        atomic::{AtomicBool, AtomicU64, Ordering},
        mpsc as std_mpsc,
    },
    time::Duration,
};

use tokio::{
    runtime::{Builder, Runtime},
    sync::{mpsc, oneshot},
    time::{Instant, sleep, sleep_until},
};

use crate::{KafkaMetrics, KafkaProducerSettings, MqError, MqResult, ProducerRecord};

use super::{
    KafkaClientError,
    client::{NativeKafkaClient, NativeKafkaClientConfig},
    connection::BrokerResponse,
    protocol::{
        API_KEY_PRODUCE, API_VERSION_PRODUCE, BatchIdentity, CompressionCodec,
        ProducePartitionRequest, ProduceResult, ProduceTopicRequest, ZstdEncoderContext,
        decode_produce_response, encode_produce_record_batch, encode_produce_request,
        request_header_version, response_header_version,
    },
};

const STREAM_GROUP_SIZE: usize = 256;

/// Kafka error codes that participate in the idempotent-producer protocol.
const DUPLICATE_SEQUENCE_NUMBER: i16 = 46;
const OUT_OF_ORDER_SEQUENCE_NUMBER: i16 = 45;
const UNKNOWN_PRODUCER_ID: i16 = 59;
/// Kafka bounds in-flight requests per connection to 5 when idempotence is on,
/// so a partition's dedup window on the broker cannot be exceeded by reordering.
const MAX_IDEMPOTENT_IN_FLIGHT: usize = 5;

static PRODUCER_RUNTIME: OnceLock<Result<Runtime, String>> = OnceLock::new();

#[derive(Clone)]
pub(crate) struct NativeKafkaProducerHandle {
    state: Arc<ProducerState>,
}

#[derive(Clone)]
pub(crate) struct NativeKafkaProducerControl {
    state: Arc<ProducerState>,
}

struct ProducerState {
    data: mpsc::Sender<ProduceCommand>,
    control: mpsc::UnboundedSender<ControlCommand>,
    metrics: KafkaMetrics,
    in_flight: AtomicU64,
    closed: AtomicBool,
    drain_timeout: Duration,
    capacity_lock: Mutex<()>,
    capacity_ready: Condvar,
    terminal_error: Mutex<Option<String>>,
}

struct ProduceCommand {
    records: Vec<ProducerRecord>,
    respond: oneshot::Sender<MqResult<()>>,
}

enum ControlCommand {
    Flush(std_mpsc::SyncSender<MqResult<()>>),
    Shutdown(std_mpsc::SyncSender<MqResult<()>>),
}

#[derive(Debug, Clone)]
struct NativeProducerConfig {
    client: NativeKafkaClientConfig,
    acks: i16,
    timeout_ms: i32,
    linger: Duration,
    max_batch_bytes: usize,
    max_retries: usize,
    retry_backoff: Duration,
    compression: CompressionCodec,
    enable_idempotence: bool,
}

/// Producer-id/epoch/sequence state for the idempotent producer.
///
/// Established once by `InitProducerId` and reset on re-initialization. Every
/// `(topic, partition)` owns a monotonic Kafka sequence space; `next_sequence`
/// tracks the base sequence to assign to the next batch for that partition.
#[derive(Debug)]
struct IdempotentState {
    producer_id: i64,
    producer_epoch: i16,
    next_sequence: HashMap<(String, i32), i32>,
}

impl IdempotentState {
    fn new(producer_id: i64, producer_epoch: i16) -> Self {
        Self {
            producer_id,
            producer_epoch,
            next_sequence: HashMap::new(),
        }
    }

    /// Assigns the base sequence for a batch of `count` records on a partition
    /// and advances the partition's sequence counter, wrapping at `i32::MAX`
    /// exactly as Kafka's `DefaultRecordBatch.incrementSequence` does.
    fn assign(&mut self, topic: &str, partition: i32, count: usize) -> i32 {
        let increment = i32::try_from(count).unwrap_or(i32::MAX);
        let entry = self
            .next_sequence
            .entry((topic.to_owned(), partition))
            .or_insert(0);
        let base = *entry;
        *entry = increment_sequence(base, increment);
        base
    }
}

/// Per-call idempotent header assignment: the negotiated producer id/epoch plus
/// the base sequence chosen for each `(topic, partition)` batch in one produce
/// call. Reused verbatim across retries so a resent batch carries the same
/// sequence and the broker deduplicates it.
#[derive(Debug)]
struct IdentityPlan {
    producer_id: i64,
    producer_epoch: i16,
    sequences: HashMap<(String, i32), i32>,
}

impl IdentityPlan {
    fn identity_for(&self, topic: &str, partition: i32) -> BatchIdentity {
        let base_sequence = self
            .sequences
            .get(&(topic.to_owned(), partition))
            .copied()
            .expect("identity plan covers every produced partition");
        BatchIdentity {
            producer_id: self.producer_id,
            producer_epoch: self.producer_epoch,
            base_sequence,
        }
    }
}

fn increment_sequence(sequence: i32, increment: i32) -> i32 {
    if sequence > i32::MAX - increment {
        increment - (i32::MAX - sequence) - 1
    } else {
        sequence + increment
    }
}

/// A non-zero produce error code that idempotent producers recover from by
/// re-initializing the producer id and resetting the per-partition sequences.
fn is_reinit_error(error: &KafkaClientError) -> bool {
    matches!(
        error,
        KafkaClientError::Broker { code, .. }
            if *code == UNKNOWN_PRODUCER_ID || *code == OUT_OF_ORDER_SEQUENCE_NUMBER
    )
}

/// Assigns a base sequence to each `(topic, partition)` in `pending`, advancing
/// `state`'s per-partition counters. Every partition in `pending` gets exactly
/// one batch (the serial path combines a partition's records into one), so this
/// is called only over records not yet acknowledged by the broker.
fn build_identity_plan_for(
    state: &mut IdempotentState,
    pending: &[&AssignedRecord],
) -> IdentityPlan {
    let mut counts = BTreeMap::<(String, i32), usize>::new();
    for record in pending {
        *counts
            .entry((record.record.topic.clone(), record.partition))
            .or_insert(0) += 1;
    }
    let mut sequences = HashMap::with_capacity(counts.len());
    for ((topic, partition), count) in counts {
        let base = state.assign(&topic, partition, count);
        sequences.insert((topic, partition), base);
    }
    IdentityPlan {
        producer_id: state.producer_id,
        producer_epoch: state.producer_epoch,
        sequences,
    }
}

/// Folds a broker's combined Produce response into the running `committed` set
/// and the first failure. Every partition the broker accepted (`error=0`, or a
/// deduplicated `DUPLICATE_SEQUENCE_NUMBER` when idempotent) is recorded as
/// committed **before** any failure is returned, so a partial commit never
/// re-sends an already-committed partition — including across a re-init that
/// resets the producer id and sequences.
fn apply_produce_results(
    sent: &[(String, i32)],
    results: &[ProduceResult],
    idempotent: bool,
    broker: i32,
    committed: &mut BTreeSet<(String, i32)>,
) -> Option<ProduceFailure> {
    let mut first_failure: Option<ProduceFailure> = None;
    for (topic, partition) in sent {
        let Some(result) = results
            .iter()
            .find(|result| &result.topic == topic && result.partition == *partition)
        else {
            first_failure.get_or_insert_with(|| ProduceFailure {
                error: KafkaClientError::protocol(format!(
                    "Produce response omitted {topic}:{partition}"
                )),
                topic: Some(topic.clone()),
                partition: Some(*partition),
                broker: Some(broker),
            });
            continue;
        };
        let duplicate = idempotent && result.error_code == DUPLICATE_SEQUENCE_NUMBER;
        if result.error_code == 0 || duplicate {
            committed.insert((topic.clone(), *partition));
        } else {
            let code = result.error_code;
            first_failure.get_or_insert_with(|| ProduceFailure {
                error: KafkaClientError::broker("Produce", code),
                topic: Some(topic.clone()),
                partition: Some(*partition),
                broker: Some(broker),
            });
        }
    }
    first_failure
}

#[derive(Debug, Clone)]
enum TerminalFailure {
    Tls {
        broker: String,
        message: String,
    },
    Sasl {
        broker: String,
        mechanism: String,
        message: String,
    },
    Delivery(String),
}

impl TerminalFailure {
    fn from_error(error: &MqError) -> Self {
        match error {
            MqError::NativeTls { broker, message } => Self::Tls {
                broker: broker.clone(),
                message: message.clone(),
            },
            MqError::NativeSasl {
                broker,
                mechanism,
                message,
            } => Self::Sasl {
                broker: broker.clone(),
                mechanism: mechanism.clone(),
                message: message.clone(),
            },
            other => Self::Delivery(other.to_string()),
        }
    }

    fn to_error(&self) -> MqError {
        match self {
            Self::Tls { broker, message } => MqError::NativeTls {
                broker: broker.clone(),
                message: message.clone(),
            },
            Self::Sasl {
                broker,
                mechanism,
                message,
            } => MqError::NativeSasl {
                broker: broker.clone(),
                mechanism: mechanism.clone(),
                message: message.clone(),
            },
            Self::Delivery(message) => MqError::Delivery(message.clone()),
        }
    }

    fn message(&self) -> String {
        self.to_error().to_string()
    }
}

#[derive(Debug, Clone, Copy)]
struct PartitionRoute {
    partition: i32,
    leader: i32,
}

#[derive(Debug)]
struct AssignedRecord {
    partition: i32,
    record: ProducerRecord,
}

#[derive(Debug)]
struct ProduceFailure {
    error: KafkaClientError,
    topic: Option<String>,
    partition: Option<i32>,
    broker: Option<i32>,
}

#[derive(Debug)]
struct ProduceUnit {
    sequence: usize,
    partitions: Vec<ProducePartitionBatch>,
    attempts: usize,
}

#[derive(Debug)]
struct ProducePartitionBatch {
    topic: String,
    partition: i32,
    records: Vec<ProducerRecord>,
}

#[derive(Debug)]
struct IssuedProduce {
    unit: ProduceUnit,
    broker: i32,
    response: Option<BrokerResponse>,
}

#[derive(Debug)]
struct FailedProduce {
    unit: ProduceUnit,
    failure: ProduceFailure,
}

struct ProducerEngine {
    config: NativeProducerConfig,
    metrics: KafkaMetrics,
    client: Option<NativeKafkaClient>,
    routes: HashMap<String, Vec<PartitionRoute>>,
    sticky_cursor: HashMap<String, usize>,
    // Reused across produce batches to avoid re-allocating Zstd's internal
    // compression tables per call; the produce loop is single-owner.
    zstd_context: ZstdEncoderContext,
    idempotent: Option<IdempotentState>,
}

impl NativeKafkaProducerControl {
    pub(crate) fn start(
        settings: &KafkaProducerSettings,
        metrics: KafkaMetrics,
    ) -> MqResult<(Self, NativeKafkaProducerHandle)> {
        let config = native_config(settings)?;
        let command_capacity = settings
            .in_flight_limit
            .div_ceil(STREAM_GROUP_SIZE)
            .clamp(1, 256);
        let (data, data_rx) = mpsc::channel(command_capacity);
        let (control, control_rx) = mpsc::unbounded_channel();
        let task_metrics = metrics.clone();
        let state = Arc::new(ProducerState {
            data,
            control,
            metrics,
            in_flight: AtomicU64::new(0),
            closed: AtomicBool::new(false),
            drain_timeout: settings.drain_timeout,
            capacity_lock: Mutex::new(()),
            capacity_ready: Condvar::new(),
            terminal_error: Mutex::new(None),
        });
        producer_runtime()?.spawn(producer_task(
            config,
            task_metrics,
            Arc::clone(&state),
            data_rx,
            control_rx,
        ));
        Ok((
            Self {
                state: Arc::clone(&state),
            },
            NativeKafkaProducerHandle {
                state: Arc::clone(&state),
            },
        ))
    }

    pub(crate) fn flush(&self) -> MqResult<()> {
        self.control(false)
    }

    pub(crate) fn finish(&self) -> MqResult<()> {
        if self.state.closed.swap(true, Ordering::SeqCst) {
            return Ok(());
        }
        self.control(true)
    }

    fn control(&self, shutdown: bool) -> MqResult<()> {
        let (respond, receiver) = std_mpsc::sync_channel(1);
        let command = if shutdown {
            ControlCommand::Shutdown(respond)
        } else {
            ControlCommand::Flush(respond)
        };
        self.state
            .control
            .send(command)
            .map_err(|_| MqError::Failed("native Kafka producer task closed".to_owned()))?;
        receiver
            .recv_timeout(self.state.drain_timeout)
            .map_err(|_| MqError::DrainTimeout)?
    }
}

impl NativeKafkaProducerHandle {
    pub(crate) fn send(&self, records: Vec<ProducerRecord>) -> MqResult<()> {
        if self.state.closed.load(Ordering::SeqCst) {
            return Err(MqError::Failed(
                "native Kafka producer is already shut down".to_owned(),
            ));
        }
        let count = records.len() as u64;
        let in_flight = self.state.in_flight.fetch_add(count, Ordering::Relaxed) + count;
        self.state.metrics.producer_in_flight(in_flight);
        let (respond, receiver) = oneshot::channel();
        let mut command = ProduceCommand { records, respond };
        let deadline = std::time::Instant::now() + self.state.drain_timeout;
        loop {
            if let Some(error) = self.state.terminal_error() {
                self.state.complete(count);
                return Err(MqError::Delivery(error));
            }
            match self.state.data.try_send(command) {
                Ok(()) => {
                    drop(receiver);
                    return Ok(());
                }
                Err(mpsc::error::TrySendError::Full(returned)) => {
                    self.state.metrics.queue_full();
                    command = returned;
                    let guard = self.state.capacity_lock.lock().map_err(|_| {
                        MqError::Failed("native Kafka producer capacity lock poisoned".to_owned())
                    })?;
                    if std::time::Instant::now() >= deadline {
                        self.state.complete(count);
                        return Err(MqError::DrainTimeout);
                    }
                    drop(
                        self.state
                            .capacity_ready
                            .wait_timeout(guard, Duration::from_millis(1))
                            .map_err(|_| {
                                MqError::Failed(
                                    "native Kafka producer capacity lock poisoned".to_owned(),
                                )
                            })?,
                    );
                }
                Err(mpsc::error::TrySendError::Closed(_)) => {
                    self.state.complete(count);
                    return Err(MqError::Failed(
                        "native Kafka producer task closed".to_owned(),
                    ));
                }
            }
        }
    }
}

impl ProducerState {
    fn terminal_error(&self) -> Option<String> {
        self.terminal_error
            .lock()
            .ok()
            .and_then(|error| error.clone())
    }

    fn set_terminal_error(&self, message: String) {
        if let Ok(mut error) = self.terminal_error.lock()
            && error.is_none()
        {
            *error = Some(message);
        }
        self.capacity_ready.notify_all();
    }

    fn complete(&self, count: u64) {
        let previous = self.in_flight.fetch_sub(count, Ordering::Relaxed);
        self.metrics
            .producer_in_flight(previous.saturating_sub(count));
        self.capacity_ready.notify_all();
    }
}

async fn producer_task(
    config: NativeProducerConfig,
    metrics: KafkaMetrics,
    state: Arc<ProducerState>,
    mut data: mpsc::Receiver<ProduceCommand>,
    mut control: mpsc::UnboundedReceiver<ControlCommand>,
) {
    let mut engine = ProducerEngine {
        config,
        metrics,
        client: None,
        routes: HashMap::new(),
        sticky_cursor: HashMap::new(),
        zstd_context: ZstdEncoderContext::new().expect("zstd encoder context initializes"),
        idempotent: None,
    };
    let mut terminal_error: Option<TerminalFailure> = None;
    loop {
        tokio::select! {
            biased;
            command = control.recv() => {
                let Some(command) = command else { break };
                let shutdown = matches!(command, ControlCommand::Shutdown(_));
                let result = drain_ready(&mut engine, &state, &mut data, &mut terminal_error).await;
                match command {
                    ControlCommand::Flush(respond) | ControlCommand::Shutdown(respond) => {
                        let _ = respond.send(result);
                    }
                }
                if shutdown { break; }
            }
            command = data.recv() => {
                let Some(first) = command else { break };
                state.capacity_ready.notify_all();
                process_group(&mut engine, &state, &mut data, first, &mut terminal_error).await;
            }
        }
    }
    while let Ok(command) = data.try_recv() {
        state.complete(command.records.len() as u64);
        let _ = command.respond.send(Err(MqError::Failed(
            terminal_error
                .as_ref()
                .map(TerminalFailure::message)
                .unwrap_or_else(|| "native Kafka producer shut down".to_owned()),
        )));
    }
}

async fn drain_ready(
    engine: &mut ProducerEngine,
    state: &ProducerState,
    data: &mut mpsc::Receiver<ProduceCommand>,
    terminal_error: &mut Option<TerminalFailure>,
) -> MqResult<()> {
    while let Ok(first) = data.try_recv() {
        state.capacity_ready.notify_all();
        process_group(engine, state, data, first, terminal_error).await;
    }
    terminal_error
        .as_ref()
        .map_or(Ok(()), |failure| Err(failure.to_error()))
}

async fn process_group(
    engine: &mut ProducerEngine,
    state: &ProducerState,
    data: &mut mpsc::Receiver<ProduceCommand>,
    first: ProduceCommand,
    terminal_error: &mut Option<TerminalFailure>,
) {
    if let Some(failure) = terminal_error.as_ref() {
        let count = first.records.len() as u64;
        engine.metrics.delivery_failed_batch(count);
        state.complete(count);
        let _ = first.respond.send(Err(failure.to_error()));
        return;
    }
    let mut commands = vec![first];
    let mut estimated = estimated_command_bytes(&commands[0]);
    let first_topics = commands[0]
        .records
        .iter()
        .map(|record| record.topic.clone())
        .collect::<BTreeSet<_>>()
        .into_iter()
        .collect::<Vec<_>>();
    if let Err(error) = engine.ensure_routes_with_retries(&first_topics).await {
        let count = commands[0].records.len() as u64;
        engine.metrics.delivery_failed_batch(count);
        state.complete(count);
        *terminal_error = Some(TerminalFailure::from_error(&error));
        state.set_terminal_error(error.to_string());
        let _ = commands
            .pop()
            .expect("first command exists")
            .respond
            .send(Err(error));
        return;
    }
    let mut partition_bytes = BTreeMap::<(String, i32), usize>::new();
    observe_command_partitions(engine, &commands[0], &mut partition_bytes);
    let deadline = Instant::now() + engine.config.linger;
    while partition_bytes.values().copied().max().unwrap_or(estimated)
        < engine.config.max_batch_bytes
    {
        tokio::select! {
            _ = sleep_until(deadline) => break,
            command = data.recv() => {
                let Some(command) = command else { break };
                state.capacity_ready.notify_all();
                estimated = estimated.saturating_add(estimated_command_bytes(&command));
                observe_command_partitions(engine, &command, &mut partition_bytes);
                commands.push(command);
            }
        }
    }
    let record_count = commands
        .iter()
        .map(|command| command.records.len())
        .sum::<usize>();
    let records = commands
        .iter_mut()
        .flat_map(|command| std::mem::take(&mut command.records))
        .collect::<Vec<_>>();
    match engine.produce(records).await {
        Ok(()) => {
            engine.metrics.produced_batch(record_count as u64);
            state.complete(record_count as u64);
            for command in commands {
                let _ = command.respond.send(Ok(()));
            }
        }
        Err(error) => {
            let message = error.to_string();
            let failure = TerminalFailure::from_error(&error);
            engine.metrics.delivery_failed_batch(record_count as u64);
            state.complete(record_count as u64);
            let mut commands = commands.into_iter();
            if let Some(command) = commands.next() {
                let _ = command.respond.send(Err(error));
            }
            for command in commands {
                let _ = command
                    .respond
                    .send(Err(MqError::Delivery(message.clone())));
            }
            *terminal_error = Some(failure);
            state.set_terminal_error(message);
        }
    }
}

fn observe_command_partitions(
    engine: &ProducerEngine,
    command: &ProduceCommand,
    partition_bytes: &mut BTreeMap<(String, i32), usize>,
) {
    for record in &command.records {
        let Some(partition) = engine.preview_partition(record) else {
            continue;
        };
        let bytes = estimated_record_bytes(record);
        partition_bytes
            .entry((record.topic.clone(), partition))
            .and_modify(|total| *total = total.saturating_add(bytes))
            .or_insert(bytes);
    }
}

fn estimated_command_bytes(command: &ProduceCommand) -> usize {
    command.records.iter().fold(0_usize, |total, record| {
        total.saturating_add(estimated_record_bytes(record))
    })
}

fn estimated_record_bytes(record: &ProducerRecord) -> usize {
    record
        .topic
        .len()
        .saturating_add(record.key.as_ref().map_or(0, bytes::Bytes::len))
        .saturating_add(record.payload.as_ref().map_or(0, bytes::Bytes::len))
        .saturating_add(64)
}

impl ProducerEngine {
    async fn produce(&mut self, records: Vec<ProducerRecord>) -> MqResult<()> {
        let topics = records
            .iter()
            .map(|record| record.topic.clone())
            .collect::<BTreeSet<_>>()
            .into_iter()
            .collect::<Vec<_>>();
        self.ensure_routes_with_retries(&topics).await?;
        let assigned = self.assign(records).map_err(produce_error)?;
        // Idempotent produce always uses the combined per-broker request path
        // (one request per broker, one batch per partition): it preserves a
        // single in-flight request per topic-partition, which the broker's
        // sequence dedup window requires. Cross-batch pipelining (KC-5.5 N>1)
        // stays non-idempotent this cycle.
        if self.config.enable_idempotence || self.config.client.max_in_flight == 1 {
            self.produce_serial(&assigned, &topics).await?;
            self.rotate_sticky(&assigned);
            return Ok(());
        }
        let sticky_topics = assigned
            .iter()
            .filter(|record| record.record.key.is_none() && record.record.partition.is_none())
            .map(|record| record.record.topic.clone())
            .collect::<BTreeSet<_>>();
        let units = partition_units(assigned, &self.routes, self.config.client.max_in_flight)
            .map_err(produce_error)?;
        self.send_pipelined(units, &topics).await?;
        self.rotate_sticky_topics(sticky_topics);
        Ok(())
    }

    async fn ensure_routes_with_retries(&mut self, topics: &[String]) -> MqResult<()> {
        let mut last_error = None;
        for attempt in 0..=self.config.max_retries {
            match self.ensure_routes(topics).await {
                Ok(()) => return Ok(()),
                Err(failure) => {
                    let retriable = failure.error.retriable();
                    if matches!(
                        &failure.error,
                        KafkaClientError::Io(_) | KafkaClientError::ChannelClosed
                    ) {
                        self.client = None;
                    }
                    last_error = Some(failure);
                    if !retriable || attempt == self.config.max_retries {
                        break;
                    }
                    self.retry_sleep(attempt).await;
                }
            }
        }
        Err(produce_error(last_error.unwrap_or_else(|| {
            ProduceFailure {
                error: KafkaClientError::protocol("native Metadata retry loop exhausted"),
                topic: None,
                partition: None,
                broker: None,
            }
        })))
    }

    async fn retry_sleep(&self, attempt: usize) {
        sleep(
            self.config
                .retry_backoff
                .saturating_mul((attempt + 1) as u32),
        )
        .await;
    }

    async fn ensure_routes(&mut self, topics: &[String]) -> Result<(), ProduceFailure> {
        if topics.iter().all(|topic| self.routes.contains_key(topic)) {
            return Ok(());
        }
        if self.client.is_none() {
            self.client = Some(
                NativeKafkaClient::connect(self.config.client.clone())
                    .await
                    .map_err(|error| ProduceFailure {
                        error,
                        topic: None,
                        partition: None,
                        broker: None,
                    })?,
            );
        }
        let metadata = self
            .client
            .as_mut()
            .expect("native producer client initialized")
            .refresh_metadata(topics)
            .await
            .map_err(|error| ProduceFailure {
                error,
                topic: None,
                partition: None,
                broker: None,
            })?;
        for topic in topics {
            let metadata_topic = metadata
                .topics
                .iter()
                .find(|candidate| candidate.name == *topic)
                .ok_or_else(|| ProduceFailure {
                    error: KafkaClientError::protocol(format!(
                        "Metadata response omitted produce topic {topic}"
                    )),
                    topic: Some(topic.clone()),
                    partition: None,
                    broker: None,
                })?;
            if metadata_topic.error_code != 0 {
                return Err(ProduceFailure {
                    error: KafkaClientError::broker("Metadata", metadata_topic.error_code),
                    topic: Some(topic.clone()),
                    partition: None,
                    broker: None,
                });
            }
            let mut routes = metadata_topic
                .partitions
                .iter()
                .map(|partition| {
                    if partition.error_code != 0 {
                        return Err(ProduceFailure {
                            error: KafkaClientError::broker(
                                "Metadata partition",
                                partition.error_code,
                            ),
                            topic: Some(topic.clone()),
                            partition: Some(partition.partition_index),
                            broker: None,
                        });
                    }
                    Ok(PartitionRoute {
                        partition: partition.partition_index,
                        leader: partition.leader_id,
                    })
                })
                .collect::<Result<Vec<_>, _>>()?;
            routes.sort_by_key(|route| route.partition);
            if routes.is_empty() {
                return Err(ProduceFailure {
                    error: KafkaClientError::protocol(format!(
                        "produce topic {topic} has no partitions"
                    )),
                    topic: Some(topic.clone()),
                    partition: None,
                    broker: None,
                });
            }
            self.routes.insert(topic.clone(), routes);
        }
        Ok(())
    }

    fn assign(
        &mut self,
        records: Vec<ProducerRecord>,
    ) -> Result<Vec<AssignedRecord>, ProduceFailure> {
        records
            .into_iter()
            .map(|record| {
                let routes = self
                    .routes
                    .get(&record.topic)
                    .ok_or_else(|| ProduceFailure {
                        error: KafkaClientError::protocol(format!(
                            "no metadata routes for produce topic {}",
                            record.topic
                        )),
                        topic: Some(record.topic.clone()),
                        partition: record.partition,
                        broker: None,
                    })?;
                let partition = if let Some(partition) = record.partition {
                    if routes.iter().any(|route| route.partition == partition) {
                        partition
                    } else {
                        return Err(ProduceFailure {
                            error: KafkaClientError::InvalidConfig(format!(
                                "produce partition {}:{} does not exist",
                                record.topic, partition
                            )),
                            topic: Some(record.topic.clone()),
                            partition: Some(partition),
                            broker: None,
                        });
                    }
                } else if let Some(key) = record.key.as_deref() {
                    let index = (murmur2(key) & 0x7fff_ffff) as usize % routes.len();
                    routes[index].partition
                } else {
                    let cursor = self.sticky_cursor.entry(record.topic.clone()).or_insert(0);
                    routes[*cursor % routes.len()].partition
                };
                Ok(AssignedRecord { partition, record })
            })
            .collect()
    }

    fn preview_partition(&self, record: &ProducerRecord) -> Option<i32> {
        let routes = self.routes.get(&record.topic)?;
        if let Some(partition) = record.partition {
            return routes
                .iter()
                .any(|route| route.partition == partition)
                .then_some(partition);
        }
        if let Some(key) = record.key.as_deref() {
            let index = (murmur2(key) & 0x7fff_ffff) as usize % routes.len();
            return Some(routes[index].partition);
        }
        let cursor = self.sticky_cursor.get(&record.topic).copied().unwrap_or(0);
        Some(routes[cursor % routes.len()].partition)
    }

    async fn send_pipelined(
        &mut self,
        units: VecDeque<ProduceUnit>,
        topics: &[String],
    ) -> MqResult<()> {
        let mut ready = units;
        while !ready.is_empty() {
            if !topics.iter().all(|topic| self.routes.contains_key(topic)) {
                self.ensure_routes_with_retries(topics).await?;
            }
            let (selected, deferred) =
                select_wave(ready, &self.routes, self.config.client.max_in_flight)
                    .map_err(produce_error)?;
            ready = deferred;

            // Once any issue/response fails, no newer request is submitted. All
            // requests already issued in this wave are settled first. Because
            // select_wave admits at most one request per topic-partition, a
            // retry can never be written behind a newer request for the same
            // partition.
            let mut issued = Vec::with_capacity(selected.len());
            let mut selected = selected.into_iter();
            let mut failures = Vec::new();
            while let Some(unit) = selected.next() {
                match self.begin_unit(unit).await {
                    Ok(pending) => issued.push(pending),
                    Err(failed) => {
                        failures.push(failed);
                        ready.extend(selected);
                        break;
                    }
                }
            }
            for pending in issued {
                if let Err(failed) = finish_unit(pending).await {
                    failures.push(failed);
                }
            }
            if failures.is_empty() {
                continue;
            }

            let recovery =
                prepare_retry(failures, self.config.max_retries).map_err(produce_error)?;
            if let Some(client) = self.client.as_mut() {
                for broker in &recovery.brokers {
                    client.invalidate_broker_connection(*broker);
                }
            }
            self.routes.clear();
            if recovery.reset_client {
                self.client = None;
            }
            let mut remaining = ready.into_iter().collect::<Vec<_>>();
            remaining.extend(recovery.units);
            remaining.sort_by_key(|unit| unit.sequence);
            ready = remaining.into();
            self.retry_sleep(recovery.backoff_attempt).await;
        }
        Ok(())
    }

    /// Negotiates the producer id/epoch once for the idempotent producer.
    ///
    /// Requires a connected client (guaranteed by `ensure_routes`). A no-op once
    /// the id is established; cleared by re-initialization.
    async fn ensure_producer_id(&mut self) -> Result<(), ProduceFailure> {
        if self.idempotent.is_some() {
            return Ok(());
        }
        let client = self.client.as_mut().ok_or_else(|| ProduceFailure {
            error: KafkaClientError::protocol(
                "native idempotent producer requires a connected client before InitProducerId",
            ),
            topic: None,
            partition: None,
            broker: None,
        })?;
        let (producer_id, producer_epoch) =
            client
                .init_producer_id()
                .await
                .map_err(|error| ProduceFailure {
                    error,
                    topic: None,
                    partition: None,
                    broker: None,
                })?;
        self.idempotent = Some(IdempotentState::new(producer_id, producer_epoch));
        Ok(())
    }

    /// Assigns a base sequence to each `(topic, partition)` in `pending` and
    /// advances the engine's per-partition sequence counters. The returned plan
    /// is reused verbatim across retries so a resent batch keeps its sequence; a
    /// re-initialization clears the counters and rebuilds from 0 over only the
    /// still-unacknowledged records, so already-committed partitions are never
    /// re-sequenced under the new producer id.
    fn build_identity_plan(&mut self, pending: &[&AssignedRecord]) -> IdentityPlan {
        let state = self
            .idempotent
            .as_mut()
            .expect("idempotent state is initialized before building the identity plan");
        build_identity_plan_for(state, pending)
    }

    async fn produce_serial(
        &mut self,
        assigned: &[AssignedRecord],
        topics: &[String],
    ) -> MqResult<()> {
        let idempotent = self.config.enable_idempotence;
        let mut plan: Option<IdentityPlan> = None;
        // Partitions the broker has durably committed. A committed partition is
        // never re-sent — not on a normal retry and not after a re-init — so a
        // partial commit (one partition succeeds while another returns a re-init
        // code in the same combined response) cannot duplicate the committed
        // partition under a fresh producer id.
        let mut committed = BTreeSet::<(String, i32)>::new();
        let mut last_error = None;
        for attempt in 0..=self.config.max_retries {
            let pending = assigned
                .iter()
                .filter(|record| {
                    !committed.contains(&(record.record.topic.clone(), record.partition))
                })
                .collect::<Vec<_>>();
            if pending.is_empty() {
                return Ok(());
            }
            if !topics.iter().all(|topic| self.routes.contains_key(topic)) {
                match self.ensure_routes(topics).await {
                    Ok(()) => {}
                    Err(failure) => {
                        let retriable = failure.error.retriable();
                        if matches!(
                            &failure.error,
                            KafkaClientError::Io(_) | KafkaClientError::ChannelClosed
                        ) {
                            self.client = None;
                        }
                        last_error = Some(failure);
                        if !retriable || attempt == self.config.max_retries {
                            break;
                        }
                        self.retry_sleep(attempt).await;
                        continue;
                    }
                }
            }
            if idempotent && plan.is_none() {
                match self.ensure_producer_id().await {
                    Ok(()) => plan = Some(self.build_identity_plan(&pending)),
                    Err(failure) => {
                        let retriable = failure.error.retriable();
                        if matches!(
                            &failure.error,
                            KafkaClientError::Io(_) | KafkaClientError::ChannelClosed
                        ) {
                            self.client = None;
                        }
                        last_error = Some(failure);
                        if !retriable || attempt == self.config.max_retries {
                            break;
                        }
                        self.retry_sleep(attempt).await;
                        continue;
                    }
                }
            }
            match self
                .send_combined_once(&pending, plan.as_ref(), &mut committed)
                .await
            {
                Ok(()) => return Ok(()),
                Err(failure) => {
                    let retriable = failure.error.retriable();
                    let reinit = idempotent && is_reinit_error(&failure.error);
                    if let Some(broker) = failure.broker
                        && let Some(client) = self.client.as_mut()
                    {
                        client.invalidate_broker_connection(broker);
                    }
                    let reset_client = matches!(
                        &failure.error,
                        KafkaClientError::Io(_) | KafkaClientError::ChannelClosed
                    );
                    last_error = Some(failure);
                    if (!retriable && !reinit) || attempt == self.config.max_retries {
                        break;
                    }
                    self.routes.clear();
                    if reset_client {
                        self.client = None;
                    }
                    if reinit {
                        // Drop the producer id and per-partition sequences: the
                        // next iteration re-initializes and rebuilds the plan
                        // with fresh sequences from 0 under the new producer id.
                        self.idempotent = None;
                        plan = None;
                    }
                    self.retry_sleep(attempt).await;
                }
            }
        }
        Err(produce_error(last_error.unwrap_or_else(|| {
            ProduceFailure {
                error: KafkaClientError::protocol("native Produce retry loop exhausted"),
                topic: None,
                partition: None,
                broker: None,
            }
        })))
    }

    async fn send_combined_once(
        &mut self,
        records: &[&AssignedRecord],
        plan: Option<&IdentityPlan>,
        committed: &mut BTreeSet<(String, i32)>,
    ) -> Result<(), ProduceFailure> {
        let mut by_broker = BTreeMap::<i32, BTreeMap<(String, i32), Vec<&ProducerRecord>>>::new();
        for assigned in records {
            let leader = leader_for(&self.routes, &assigned.record.topic, assigned.partition)?;
            by_broker
                .entry(leader)
                .or_default()
                .entry((assigned.record.topic.clone(), assigned.partition))
                .or_default()
                .push(&assigned.record);
        }
        for (broker, batches) in by_broker {
            self.send_combined_broker(broker, batches, plan, committed)
                .await?;
        }
        Ok(())
    }

    async fn send_combined_broker(
        &mut self,
        broker: i32,
        batches: BTreeMap<(String, i32), Vec<&ProducerRecord>>,
        plan: Option<&IdentityPlan>,
        committed: &mut BTreeSet<(String, i32)>,
    ) -> Result<(), ProduceFailure> {
        let mut topics = BTreeMap::<String, Vec<ProducePartitionRequest>>::new();
        for ((topic, partition), records) in &batches {
            let identity = plan.map(|plan| plan.identity_for(topic, *partition));
            topics
                .entry(topic.clone())
                .or_default()
                .push(ProducePartitionRequest {
                    index: *partition,
                    records: encode_produce_record_batch(
                        records,
                        self.config.compression,
                        &mut self.zstd_context,
                        identity,
                    )
                    .map_err(|error| ProduceFailure {
                        error,
                        topic: Some(topic.clone()),
                        partition: Some(*partition),
                        broker: Some(broker),
                    })?,
                });
        }
        let topics = topics
            .into_iter()
            .map(|(name, partitions)| ProduceTopicRequest { name, partitions })
            .collect::<Vec<_>>();
        let body = encode_produce_request(None, self.config.acks, self.config.timeout_ms, &topics)
            .map_err(|error| ProduceFailure {
                error,
                topic: None,
                partition: None,
                broker: Some(broker),
            })?;
        let connection = self
            .client
            .as_mut()
            .expect("native producer client initialized")
            .connection_for_broker(broker)
            .await
            .map_err(|error| ProduceFailure {
                error,
                topic: None,
                partition: None,
                broker: Some(broker),
            })?;
        if self.config.acks == 0 {
            return connection
                .send_one_way(
                    API_KEY_PRODUCE,
                    API_VERSION_PRODUCE,
                    request_header_version(API_VERSION_PRODUCE),
                    body,
                )
                .await
                .map_err(|error| ProduceFailure {
                    error,
                    topic: None,
                    partition: None,
                    broker: Some(broker),
                });
        }
        let response = connection
            .request(
                API_KEY_PRODUCE,
                API_VERSION_PRODUCE,
                request_header_version(API_VERSION_PRODUCE),
                response_header_version(API_KEY_PRODUCE, API_VERSION_PRODUCE),
                body,
            )
            .await
            .map_err(|error| ProduceFailure {
                error,
                topic: None,
                partition: None,
                broker: Some(broker),
            })?;
        let results = decode_produce_response(API_VERSION_PRODUCE, &response).map_err(|error| {
            ProduceFailure {
                error,
                topic: None,
                partition: None,
                broker: Some(broker),
            }
        })?;
        let sent = batches.keys().cloned().collect::<Vec<_>>();
        match apply_produce_results(&sent, &results, plan.is_some(), broker, committed) {
            Some(failure) => Err(failure),
            None => Ok(()),
        }
    }

    async fn begin_unit(&mut self, unit: ProduceUnit) -> Result<IssuedProduce, FailedProduce> {
        let (context_topic, context_partition) = unit_context(&unit);
        let broker = match unit_leader(&self.routes, &unit) {
            Ok(broker) => broker,
            Err(failure) => return Err(FailedProduce { unit, failure }),
        };
        let mut topics = BTreeMap::<String, Vec<ProducePartitionRequest>>::new();
        for partition in &unit.partitions {
            let record_refs = partition.records.iter().collect::<Vec<_>>();
            // The pipelined (N>1) path stays non-idempotent this cycle; idempotent
            // produce always routes through the combined per-broker serial path.
            let batch = match encode_produce_record_batch(
                &record_refs,
                self.config.compression,
                &mut self.zstd_context,
                None,
            ) {
                Ok(batch) => batch,
                Err(error) => {
                    return Err(FailedProduce {
                        failure: ProduceFailure {
                            error,
                            topic: Some(partition.topic.clone()),
                            partition: Some(partition.partition),
                            broker: Some(broker),
                        },
                        unit,
                    });
                }
            };
            topics
                .entry(partition.topic.clone())
                .or_default()
                .push(ProducePartitionRequest {
                    index: partition.partition,
                    records: batch,
                });
        }
        let topics = topics
            .into_iter()
            .map(|(name, partitions)| ProduceTopicRequest { name, partitions })
            .collect::<Vec<_>>();
        let body =
            match encode_produce_request(None, self.config.acks, self.config.timeout_ms, &topics) {
                Ok(body) => body,
                Err(error) => {
                    return Err(FailedProduce {
                        failure: ProduceFailure {
                            error,
                            topic: Some(context_topic.clone()),
                            partition: Some(context_partition),
                            broker: Some(broker),
                        },
                        unit,
                    });
                }
            };
        let connection = match self
            .client
            .as_mut()
            .expect("native producer client initialized")
            .connection_for_broker(broker)
            .await
        {
            Ok(connection) => connection,
            Err(error) => {
                return Err(FailedProduce {
                    failure: ProduceFailure {
                        error,
                        topic: Some(context_topic.clone()),
                        partition: Some(context_partition),
                        broker: Some(broker),
                    },
                    unit,
                });
            }
        };
        let response = if self.config.acks == 0 {
            if let Err(error) = connection
                .send_one_way(
                    API_KEY_PRODUCE,
                    API_VERSION_PRODUCE,
                    request_header_version(API_VERSION_PRODUCE),
                    body,
                )
                .await
            {
                return Err(FailedProduce {
                    failure: ProduceFailure {
                        error,
                        topic: Some(context_topic.clone()),
                        partition: Some(context_partition),
                        broker: Some(broker),
                    },
                    unit,
                });
            }
            None
        } else {
            match connection
                .begin_request(
                    API_KEY_PRODUCE,
                    API_VERSION_PRODUCE,
                    request_header_version(API_VERSION_PRODUCE),
                    response_header_version(API_KEY_PRODUCE, API_VERSION_PRODUCE),
                    body,
                )
                .await
            {
                Ok(response) => Some(response),
                Err(error) => {
                    return Err(FailedProduce {
                        failure: ProduceFailure {
                            error,
                            topic: Some(context_topic.clone()),
                            partition: Some(context_partition),
                            broker: Some(broker),
                        },
                        unit,
                    });
                }
            }
        };
        Ok(IssuedProduce {
            unit,
            broker,
            response,
        })
    }

    fn rotate_sticky(&mut self, records: &[AssignedRecord]) {
        let topics = records
            .iter()
            .filter(|record| record.record.key.is_none() && record.record.partition.is_none())
            .map(|record| record.record.topic.clone())
            .collect::<BTreeSet<_>>();
        self.rotate_sticky_topics(topics);
    }

    fn rotate_sticky_topics(&mut self, topics: BTreeSet<String>) {
        for topic in topics {
            let partition_count = self.routes.get(&topic).map_or(1, Vec::len);
            let cursor = self.sticky_cursor.entry(topic).or_insert(0);
            *cursor = (*cursor + 1) % partition_count;
        }
    }
}

#[derive(Debug)]
struct RetryRecovery {
    units: Vec<ProduceUnit>,
    brokers: BTreeSet<i32>,
    reset_client: bool,
    backoff_attempt: usize,
}

fn partition_units(
    assigned: Vec<AssignedRecord>,
    routes: &HashMap<String, Vec<PartitionRoute>>,
    max_in_flight: usize,
) -> Result<VecDeque<ProduceUnit>, ProduceFailure> {
    let mut partitions = BTreeMap::<(String, i32), Vec<ProducerRecord>>::new();
    for assigned in assigned {
        partitions
            .entry((assigned.record.topic.clone(), assigned.partition))
            .or_default()
            .push(assigned.record);
    }
    let mut by_broker = BTreeMap::<i32, Vec<ProducePartitionBatch>>::new();
    for ((topic, partition), records) in partitions {
        let broker = leader_for(routes, &topic, partition)?;
        by_broker
            .entry(broker)
            .or_default()
            .push(ProducePartitionBatch {
                topic,
                partition,
                records,
            });
    }
    let mut units = VecDeque::new();
    for batches in by_broker.into_values() {
        let width = batches.len().min(max_in_flight.max(1));
        let mut stripes = (0..width).map(|_| Vec::new()).collect::<Vec<_>>();
        for (index, batch) in batches.into_iter().enumerate() {
            stripes[index % width].push(batch);
        }
        for partitions in stripes {
            units.push_back(ProduceUnit {
                sequence: units.len(),
                partitions,
                attempts: 0,
            });
        }
    }
    Ok(units)
}

fn leader_for(
    routes: &HashMap<String, Vec<PartitionRoute>>,
    topic: &str,
    partition: i32,
) -> Result<i32, ProduceFailure> {
    routes
        .get(topic)
        .and_then(|routes| routes.iter().find(|route| route.partition == partition))
        .map(|route| route.leader)
        .ok_or_else(|| ProduceFailure {
            error: KafkaClientError::protocol(format!("no leader route for {topic}:{partition}")),
            topic: Some(topic.to_owned()),
            partition: Some(partition),
            broker: None,
        })
}

fn unit_context(unit: &ProduceUnit) -> (String, i32) {
    let partition = unit
        .partitions
        .first()
        .expect("Produce units always contain a partition");
    (partition.topic.clone(), partition.partition)
}

fn unit_leader(
    routes: &HashMap<String, Vec<PartitionRoute>>,
    unit: &ProduceUnit,
) -> Result<i32, ProduceFailure> {
    let (topic, partition) = unit_context(unit);
    let broker = leader_for(routes, &topic, partition)?;
    for candidate in unit.partitions.iter().skip(1) {
        let candidate_broker = leader_for(routes, &candidate.topic, candidate.partition)?;
        if candidate_broker != broker {
            return Err(ProduceFailure {
                error: KafkaClientError::protocol(
                    "Produce pipeline unit spans multiple broker connections",
                ),
                topic: Some(candidate.topic.clone()),
                partition: Some(candidate.partition),
                broker: Some(candidate_broker),
            });
        }
    }
    Ok(broker)
}

fn select_wave(
    ready: VecDeque<ProduceUnit>,
    routes: &HashMap<String, Vec<PartitionRoute>>,
    max_in_flight: usize,
) -> Result<(Vec<ProduceUnit>, VecDeque<ProduceUnit>), ProduceFailure> {
    let mut selected = Vec::new();
    let mut deferred = VecDeque::new();
    let mut broker_counts = HashMap::<i32, usize>::new();
    let mut seen_partitions = BTreeSet::<(String, i32)>::new();
    for unit in ready {
        let keys = unit
            .partitions
            .iter()
            .map(|partition| (partition.topic.clone(), partition.partition))
            .collect::<Vec<_>>();
        let overlaps_earlier = keys.iter().any(|key| seen_partitions.contains(key));
        seen_partitions.extend(keys);
        if overlaps_earlier {
            deferred.push_back(unit);
            continue;
        }
        let broker = unit_leader(routes, &unit)?;
        let count = broker_counts.entry(broker).or_insert(0);
        if *count < max_in_flight.max(1) {
            *count += 1;
            selected.push(unit);
        } else {
            deferred.push_back(unit);
        }
    }
    Ok((selected, deferred))
}

async fn finish_unit(pending: IssuedProduce) -> Result<(), FailedProduce> {
    let IssuedProduce {
        unit,
        broker,
        response,
    } = pending;
    let Some(response) = response else {
        return Ok(());
    };
    let (context_topic, context_partition) = unit_context(&unit);
    let response = match response.receive().await {
        Ok(response) => response,
        Err(error) => {
            return Err(FailedProduce {
                failure: ProduceFailure {
                    error,
                    topic: Some(context_topic.clone()),
                    partition: Some(context_partition),
                    broker: Some(broker),
                },
                unit,
            });
        }
    };
    let results = match decode_produce_response(API_VERSION_PRODUCE, &response) {
        Ok(results) => results,
        Err(error) => {
            return Err(FailedProduce {
                failure: ProduceFailure {
                    error,
                    topic: Some(context_topic.clone()),
                    partition: Some(context_partition),
                    broker: Some(broker),
                },
                unit,
            });
        }
    };
    for partition in &unit.partitions {
        let Some(result) = results.iter().find(|result| {
            result.topic == partition.topic && result.partition == partition.partition
        }) else {
            return Err(FailedProduce {
                failure: ProduceFailure {
                    error: KafkaClientError::protocol(format!(
                        "Produce response omitted {}:{}",
                        partition.topic, partition.partition
                    )),
                    topic: Some(partition.topic.clone()),
                    partition: Some(partition.partition),
                    broker: Some(broker),
                },
                unit,
            });
        };
        if result.error_code != 0 {
            return Err(FailedProduce {
                failure: ProduceFailure {
                    error: KafkaClientError::broker("Produce", result.error_code),
                    topic: Some(partition.topic.clone()),
                    partition: Some(partition.partition),
                    broker: Some(broker),
                },
                unit,
            });
        }
    }
    Ok(())
}

fn prepare_retry(
    mut failures: Vec<FailedProduce>,
    max_retries: usize,
) -> Result<RetryRecovery, ProduceFailure> {
    failures.sort_by_key(|failed| failed.unit.sequence);
    let mut recovery = RetryRecovery {
        units: Vec::with_capacity(failures.len()),
        brokers: BTreeSet::new(),
        reset_client: false,
        backoff_attempt: 0,
    };
    for mut failed in failures {
        if !failed.failure.error.retriable() || failed.unit.attempts == max_retries {
            return Err(failed.failure);
        }
        recovery.reset_client |= matches!(
            failed.failure.error,
            KafkaClientError::Io(_) | KafkaClientError::ChannelClosed
        );
        if let Some(broker) = failed.failure.broker {
            recovery.brokers.insert(broker);
        }
        recovery.backoff_attempt = recovery.backoff_attempt.max(failed.unit.attempts);
        failed.unit.attempts += 1;
        recovery.units.push(failed.unit);
    }
    Ok(recovery)
}

impl std::fmt::Display for ProduceFailure {
    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        self.error.fmt(formatter)
    }
}

fn produce_error(failure: ProduceFailure) -> MqError {
    match (&failure.error, failure.topic, failure.partition) {
        (KafkaClientError::Broker { code, message, .. }, Some(topic), Some(partition)) => {
            MqError::NativeDelivery {
                topic,
                partition,
                code: *code,
                message: message.clone(),
            }
        }
        (KafkaClientError::Tls { broker, message }, _, _) => MqError::NativeTls {
            broker: broker.clone(),
            message: message.clone(),
        },
        (
            KafkaClientError::Sasl {
                broker,
                mechanism,
                message,
            },
            _,
            _,
        ) => MqError::NativeSasl {
            broker: broker.clone(),
            mechanism: mechanism.clone(),
            message: message.clone(),
        },
        (error, _, _) => MqError::Delivery(error.to_string()),
    }
}

fn native_config(settings: &KafkaProducerSettings) -> MqResult<NativeProducerConfig> {
    validate_native_boundary(settings)?;
    let bootstrap = required(settings, "bootstrap.servers")?;
    let mut client = NativeKafkaClientConfig::new(bootstrap);
    client.security = crate::native::native_security_config(&settings.config)?;
    if let Some(client_id) = settings.config.get("client.id") {
        client.client_id = client_id.to_owned();
    }
    if let Some(timeout) = parse_u64(settings, "request.timeout.ms")? {
        client.request_timeout = Duration::from_millis(timeout);
    }
    let max_in_flight =
        parse_usize(settings, "max.in.flight.requests.per.connection")?.unwrap_or(1);
    if max_in_flight == 0 {
        return Err(MqError::InvalidConfig(
            "native Kafka producer max.in.flight.requests.per.connection must be greater than zero"
                .to_owned(),
        ));
    }
    client.max_in_flight = max_in_flight;
    let acks = match settings.config.get("acks").unwrap_or("all").trim() {
        "all" | "-1" => -1,
        "1" => 1,
        "0" => 0,
        other => {
            return Err(MqError::InvalidConfig(format!(
                "native Kafka producer supports acks=all/-1, 1, or 0; got {other}"
            )));
        }
    };
    let enable_idempotence = idempotence_enabled(settings);
    if enable_idempotence {
        // Kafka's idempotent-producer invariants. Fail fast naming the offending
        // setting rather than silently downgrading delivery guarantees.
        if acks != -1 {
            return Err(native_idempotence_conflict("acks", "acks=all is required"));
        }
        if max_in_flight > MAX_IDEMPOTENT_IN_FLIGHT {
            return Err(native_idempotence_conflict(
                "max.in.flight.requests.per.connection",
                "at most 5 in-flight requests per connection are allowed",
            ));
        }
    }
    let timeout_ms = parse_i32(settings, "request.timeout.ms")?.unwrap_or(30_000);
    if timeout_ms < 0 {
        return Err(MqError::InvalidConfig(
            "native Kafka producer request.timeout.ms must be non-negative".to_owned(),
        ));
    }
    Ok(NativeProducerConfig {
        client,
        acks,
        timeout_ms,
        linger: Duration::from_millis(parse_u64(settings, "linger.ms")?.unwrap_or(5)),
        max_batch_bytes: parse_usize(settings, "batch.size")?
            .unwrap_or(131_072)
            .max(1),
        max_retries: parse_usize(settings, "retries")?.unwrap_or(3),
        retry_backoff: Duration::from_millis(
            parse_u64(settings, "retry.backoff.ms")?.unwrap_or(100),
        ),
        compression: native_compression(settings)?,
        enable_idempotence,
    })
}

/// Reads `enable.idempotence`, defaulting to off when callers construct raw
/// settings without the public defaults. Accepts the librdkafka-style
/// truthy spellings used across the config surface.
fn idempotence_enabled(settings: &KafkaProducerSettings) -> bool {
    settings
        .config
        .get("enable.idempotence")
        .map(|value| value.trim().to_ascii_lowercase())
        .is_some_and(|value| matches!(value.as_str(), "1" | "true" | "yes"))
}

fn native_idempotence_conflict(setting: &str, requirement: &str) -> MqError {
    MqError::InvalidConfig(format!(
        "native Kafka idempotent producer requires {requirement}; incompatible {setting}. \
         Disable enable.idempotence for bounded at-least-once delivery, or {}",
        crate::producer_rdkafka_hint()
    ))
}

fn native_compression(settings: &KafkaProducerSettings) -> MqResult<CompressionCodec> {
    let value = settings
        .config
        .get("compression.type")
        .or_else(|| settings.config.get("compression.codec"))
        .unwrap_or("none")
        .trim()
        .to_ascii_lowercase();
    match value.as_str() {
        "" | "none" | "uncompressed" => Ok(CompressionCodec::None),
        "lz4" => Ok(CompressionCodec::Lz4),
        "zstd" => Ok(CompressionCodec::Zstd),
        "gzip" | "snappy" => Err(native_unsupported(
            "compression.type supports none, lz4, and zstd; gzip/Snappy are not supported",
        )),
        other => Err(MqError::InvalidConfig(format!(
            "native Kafka producer compression.type must be none, lz4, or zstd; got {other}"
        ))),
    }
}

fn validate_native_boundary(settings: &KafkaProducerSettings) -> MqResult<()> {
    for (key, value) in settings.config.properties() {
        let key = key.trim().to_ascii_lowercase();
        let value = value.trim().to_ascii_lowercase();
        if key == "transactional.id" || key.starts_with("transaction.") {
            return Err(native_unsupported("transactions are not supported"));
        }
        if matches!(key.as_str(), "compression.type" | "compression.codec")
            && matches!(value.as_str(), "gzip" | "snappy")
        {
            return Err(native_unsupported(
                "compression.type supports none, lz4, and zstd; gzip/Snappy are not supported",
            ));
        }
    }
    Ok(())
}

fn native_unsupported(message: &str) -> MqError {
    MqError::InvalidConfig(format!(
        "native Kafka producer backend: {message}; {}",
        crate::producer_rdkafka_hint()
    ))
}

fn required(settings: &KafkaProducerSettings, key: &str) -> MqResult<String> {
    settings
        .config
        .get(key)
        .filter(|value| !value.trim().is_empty())
        .map(str::to_owned)
        .ok_or_else(|| MqError::InvalidConfig(format!("native Kafka producer requires {key}")))
}

fn parse_i32(settings: &KafkaProducerSettings, key: &str) -> MqResult<Option<i32>> {
    settings
        .config
        .get(key)
        .map(|value| {
            value.parse::<i32>().map_err(|_| {
                MqError::InvalidConfig(format!(
                    "native Kafka producer expected integer {key}, got {value}"
                ))
            })
        })
        .transpose()
}

fn parse_u64(settings: &KafkaProducerSettings, key: &str) -> MqResult<Option<u64>> {
    settings
        .config
        .get(key)
        .map(|value| {
            value.parse::<u64>().map_err(|_| {
                MqError::InvalidConfig(format!(
                    "native Kafka producer expected non-negative integer {key}, got {value}"
                ))
            })
        })
        .transpose()
}

fn parse_usize(settings: &KafkaProducerSettings, key: &str) -> MqResult<Option<usize>> {
    settings
        .config
        .get(key)
        .map(|value| {
            value.parse::<usize>().map_err(|_| {
                MqError::InvalidConfig(format!(
                    "native Kafka producer expected non-negative integer {key}, got {value}"
                ))
            })
        })
        .transpose()
}

fn producer_runtime() -> MqResult<&'static Runtime> {
    match PRODUCER_RUNTIME.get_or_init(|| {
        Builder::new_multi_thread()
            .thread_name("datum-kafka-native-producer")
            .worker_threads(1)
            .enable_all()
            .build()
            .map_err(|error| error.to_string())
    }) {
        Ok(runtime) => Ok(runtime),
        Err(message) => Err(MqError::Failed(format!(
            "failed to start native Kafka producer runtime: {message}"
        ))),
    }
}

/// Kafka's Java-compatible murmur2 hash used by the default key partitioner.
fn murmur2(data: &[u8]) -> u32 {
    const SEED: u32 = 0x9747_b28c;
    const M: u32 = 0x5bd1_e995;
    const R: u32 = 24;
    let mut hash = SEED ^ data.len() as u32;
    let mut chunks = data.chunks_exact(4);
    for chunk in &mut chunks {
        let mut k = u32::from_le_bytes(chunk.try_into().expect("four-byte chunk"));
        k = k.wrapping_mul(M);
        k ^= k >> R;
        k = k.wrapping_mul(M);
        hash = hash.wrapping_mul(M);
        hash ^= k;
    }
    let remainder = chunks.remainder();
    match remainder.len() {
        3 => {
            hash ^= (remainder[2] as u32) << 16;
            hash ^= (remainder[1] as u32) << 8;
            hash ^= remainder[0] as u32;
            hash = hash.wrapping_mul(M);
        }
        2 => {
            hash ^= (remainder[1] as u32) << 8;
            hash ^= remainder[0] as u32;
            hash = hash.wrapping_mul(M);
        }
        1 => {
            hash ^= remainder[0] as u32;
            hash = hash.wrapping_mul(M);
        }
        _ => {}
    }
    hash ^= hash >> 13;
    hash = hash.wrapping_mul(M);
    hash ^ (hash >> 15)
}

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

    #[test]
    fn murmur2_matches_java_client_vectors() {
        assert_eq!(murmur2(b""), 0x106e_08d9);
        assert_eq!(murmur2(b"kafka"), 0xd067_cf64);
    }

    #[test]
    fn native_config_defaults_to_idempotent() {
        let settings = KafkaProducerSettings::new("localhost:9092")
            .with_producer_backend(KafkaProducerBackend::Native);
        assert!(
            native_config(&settings)
                .expect("default config")
                .enable_idempotence
        );
    }

    #[test]
    fn native_config_accepts_idempotence_with_acks_all() {
        let settings = KafkaProducerSettings::new("localhost:9092")
            .with_producer_backend(KafkaProducerBackend::Native)
            .with("enable.idempotence", "true")
            .with("acks", "all");
        let config = native_config(&settings).expect("idempotent config with acks=all");
        assert!(config.enable_idempotence);
        assert_eq!(config.acks, -1);
    }

    #[test]
    fn native_config_idempotence_rejects_incompatible_acks() {
        for acks in ["1", "0"] {
            let settings = KafkaProducerSettings::new("localhost:9092")
                .with_producer_backend(KafkaProducerBackend::Native)
                .with("enable.idempotence", "true")
                .with("acks", acks);
            let error = native_config(&settings).expect_err("acks must be all for idempotence");
            assert!(
                matches!(&error, MqError::InvalidConfig(message) if message.contains("acks")),
                "acks={acks} error={error:?}"
            );
        }
    }

    #[test]
    fn native_config_idempotence_rejects_excess_in_flight() {
        let settings = KafkaProducerSettings::new("localhost:9092")
            .with_producer_backend(KafkaProducerBackend::Native)
            .with("enable.idempotence", "true")
            .with("acks", "all")
            .with_max_in_flight_requests_per_connection(6);
        let error = native_config(&settings).expect_err("idempotence caps in-flight at 5");
        assert!(matches!(
            &error,
            MqError::InvalidConfig(message)
                if message.contains("max.in.flight.requests.per.connection")
        ));

        let bounded = settings.with_max_in_flight_requests_per_connection(5);
        assert!(
            native_config(&bounded)
                .expect("in-flight of 5 is allowed")
                .enable_idempotence
        );
    }

    #[test]
    fn native_config_accepts_tls() {
        let tls = KafkaProducerSettings::new("localhost:9092")
            .with_producer_backend(KafkaProducerBackend::Native)
            .with("security.protocol", "ssl");
        assert!(native_config(&tls).is_ok());
    }

    #[test]
    fn increment_sequence_wraps_at_i32_max_like_kafka() {
        assert_eq!(increment_sequence(0, 5), 5);
        assert_eq!(increment_sequence(10, 0), 10);
        assert_eq!(increment_sequence(i32::MAX, 1), 0);
        assert_eq!(increment_sequence(i32::MAX - 2, 5), 2);
    }

    #[test]
    fn idempotent_state_assigns_monotonic_per_partition_sequences() {
        let mut state = IdempotentState::new(99, 3);
        // Each partition owns an independent sequence space starting at 0.
        assert_eq!(state.assign("topic", 0, 4), 0);
        assert_eq!(state.assign("topic", 0, 3), 4);
        assert_eq!(state.assign("topic", 0, 1), 7);
        assert_eq!(state.assign("topic", 1, 2), 0);
        assert_eq!(state.assign("topic", 1, 5), 2);
        assert_eq!(state.assign("other", 0, 1), 0);
    }

    #[test]
    fn is_reinit_error_selects_reinitializable_codes() {
        assert!(is_reinit_error(&KafkaClientError::broker(
            "Produce",
            UNKNOWN_PRODUCER_ID
        )));
        assert!(is_reinit_error(&KafkaClientError::broker(
            "Produce",
            OUT_OF_ORDER_SEQUENCE_NUMBER
        )));
        // Duplicate is handled as success, not a re-init trigger.
        assert!(!is_reinit_error(&KafkaClientError::broker(
            "Produce",
            DUPLICATE_SEQUENCE_NUMBER
        )));
        assert!(!is_reinit_error(&KafkaClientError::broker("Produce", 6)));
        assert!(!is_reinit_error(&KafkaClientError::ChannelClosed));
    }

    fn assigned(partition: i32, payload: &'static [u8]) -> AssignedRecord {
        AssignedRecord {
            partition,
            record: ProducerRecord::new("topic", Some(bytes::Bytes::from_static(payload))),
        }
    }

    fn produce_result(partition: i32, error_code: i16) -> ProduceResult {
        ProduceResult {
            topic: "topic".to_owned(),
            partition,
            error_code,
            base_offset: if error_code == 0 { 0 } else { -1 },
            error_message: None,
        }
    }

    #[test]
    fn apply_produce_results_records_committed_before_returning_failure() {
        // Mixed combined response: partition 0 commits, partition 1 fails with a
        // re-init code — both in the SAME response.
        let sent = [("topic".to_owned(), 0), ("topic".to_owned(), 1)];
        let results = [produce_result(0, 0), produce_result(1, UNKNOWN_PRODUCER_ID)];
        let mut committed = BTreeSet::new();
        let failure = apply_produce_results(&sent, &results, true, 7, &mut committed)
            .expect("partition 1 must surface a failure");
        assert!(is_reinit_error(&failure.error));
        assert_eq!(failure.partition, Some(1));
        // The committed partition is recorded even though the round failed.
        assert_eq!(committed, BTreeSet::from([("topic".to_owned(), 0)]));

        // Idempotent DUPLICATE_SEQUENCE_NUMBER counts as committed, not a failure.
        let mut committed = BTreeSet::new();
        assert!(
            apply_produce_results(
                &[("topic".to_owned(), 0)],
                &[produce_result(0, DUPLICATE_SEQUENCE_NUMBER)],
                true,
                7,
                &mut committed,
            )
            .is_none()
        );
        assert_eq!(committed, BTreeSet::from([("topic".to_owned(), 0)]));
    }

    #[test]
    fn partial_commit_reinit_never_resends_committed_partition() {
        // Reproduces the blocker: a combined response commits partition 0 while
        // partition 1 returns UNKNOWN_PRODUCER_ID. A full re-init must re-send
        // ONLY partition 1 under the new producer id; re-sending the committed
        // partition 0 with a fresh seq=0 would physically duplicate it.
        let records = [assigned(0, b"a"), assigned(1, b"b")];

        // Round 1: plan covers both partitions under producer id 100, seq 0 each.
        let mut state = IdempotentState::new(100, 0);
        let round1 = records.iter().collect::<Vec<_>>();
        let plan1 = build_identity_plan_for(&mut state, &round1);
        assert_eq!(plan1.identity_for("topic", 0).base_sequence, 0);
        assert_eq!(plan1.identity_for("topic", 1).base_sequence, 0);

        // Round 1 response marks partition 0 committed, partition 1 needs re-init.
        let mut committed = BTreeSet::new();
        let failure = apply_produce_results(
            &[("topic".to_owned(), 0), ("topic".to_owned(), 1)],
            &[produce_result(0, 0), produce_result(1, UNKNOWN_PRODUCER_ID)],
            true,
            7,
            &mut committed,
        )
        .expect("partition 1 failed");
        assert!(is_reinit_error(&failure.error));

        // Re-init: fresh producer id 200, sequences cleared. The next round's
        // pending set EXCLUDES the committed partition 0.
        let mut state = IdempotentState::new(200, 0);
        let round2 = records
            .iter()
            .filter(|record| !committed.contains(&(record.record.topic.clone(), record.partition)))
            .collect::<Vec<_>>();
        assert_eq!(round2.len(), 1, "only partition 1 remains unacknowledged");
        assert_eq!(round2[0].partition, 1);

        let plan2 = build_identity_plan_for(&mut state, &round2);
        assert_eq!(plan2.producer_id, 200);
        // Partition 1 is re-sequenced from 0 under the new producer id...
        assert_eq!(plan2.identity_for("topic", 1).base_sequence, 0);
        // ...and partition 0 is absent from the resend plan — never re-sent.
        assert!(!plan2.sequences.contains_key(&("topic".to_owned(), 0)));
    }

    #[test]
    fn identity_plan_reuses_producer_id_epoch_and_base_sequence() {
        let mut state = IdempotentState::new(4242, 7);
        let base = state.assign("topic", 2, 3);
        let plan = IdentityPlan {
            producer_id: state.producer_id,
            producer_epoch: state.producer_epoch,
            sequences: HashMap::from([(("topic".to_owned(), 2), base)]),
        };
        let identity = plan.identity_for("topic", 2);
        assert_eq!(identity.producer_id, 4242);
        assert_eq!(identity.producer_epoch, 7);
        assert_eq!(identity.base_sequence, 0);
        // The lookup is stable across repeated retries.
        assert_eq!(plan.identity_for("topic", 2), identity);
    }

    #[test]
    fn native_config_accepts_all_ack_modes_when_idempotence_is_disabled() {
        for acks in ["all", "-1", "1", "0"] {
            let settings = KafkaProducerSettings::new("localhost:9092")
                .with_producer_backend(KafkaProducerBackend::Native)
                .with("enable.idempotence", "false")
                .with("acks", acks);
            assert!(native_config(&settings).is_ok(), "acks={acks}");
        }
    }

    #[test]
    fn native_config_accepts_lz4_and_zstd_but_rejects_other_codecs() {
        for (codec, expected) in [
            ("none", Some(CompressionCodec::None)),
            ("lz4", Some(CompressionCodec::Lz4)),
            ("zstd", Some(CompressionCodec::Zstd)),
            ("gzip", None),
            ("snappy", None),
            ("brotli", None),
        ] {
            let settings = KafkaProducerSettings::new("localhost:9092")
                .with_producer_backend(KafkaProducerBackend::Native)
                .with("compression.type", codec);
            match expected {
                Some(expected) => assert_eq!(
                    native_config(&settings)
                        .expect("supported codec")
                        .compression,
                    expected
                ),
                None => assert!(native_config(&settings).is_err(), "codec={codec}"),
            }
        }
    }

    #[test]
    fn unsupported_native_producer_mode_names_exact_rdkafka_switch() {
        let settings =
            KafkaProducerSettings::new("localhost:9092").with("compression.type", "gzip");
        let error = native_config(&settings).expect_err("gzip requires rdkafka");
        assert!(
            error.to_string().contains("KafkaProducerBackend::Rdkafka"),
            "unsupported boundary must name the exact backend switch: {error}"
        );
    }

    fn unit(sequence: usize, partition: i32) -> ProduceUnit {
        ProduceUnit {
            sequence,
            partitions: vec![ProducePartitionBatch {
                topic: "topic".to_owned(),
                partition,
                records: vec![ProducerRecord::new("topic", None)],
            }],
            attempts: 0,
        }
    }

    #[test]
    fn halt_drain_retry_never_sends_behind_failed_partition() {
        let routes = HashMap::from([(
            "topic".to_owned(),
            vec![
                PartitionRoute {
                    partition: 0,
                    leader: 7,
                },
                PartitionRoute {
                    partition: 1,
                    leader: 7,
                },
            ],
        )]);
        let ready = VecDeque::from([unit(0, 0), unit(1, 1), unit(2, 0)]);
        let (wave, deferred) = select_wave(ready, &routes, 4).expect("select first wave");
        assert_eq!(
            wave.iter().map(|unit| unit.sequence).collect::<Vec<_>>(),
            vec![0, 1]
        );
        assert_eq!(
            deferred
                .iter()
                .map(|unit| unit.sequence)
                .collect::<Vec<_>>(),
            vec![2],
            "newer partition-0 work must remain unissued"
        );

        // Inject failures in reverse response-settlement order. Recovery is
        // computed only after both issued outcomes are present, modeling the
        // halt-and-drain boundary, and restores original request order.
        let failures = vec![
            FailedProduce {
                unit: unit(1, 1),
                failure: ProduceFailure {
                    error: KafkaClientError::broker("Produce", 7),
                    topic: Some("topic".to_owned()),
                    partition: Some(1),
                    broker: Some(7),
                },
            },
            FailedProduce {
                unit: unit(0, 0),
                failure: ProduceFailure {
                    error: KafkaClientError::ChannelClosed,
                    topic: Some("topic".to_owned()),
                    partition: Some(0),
                    broker: Some(7),
                },
            },
        ];
        let recovery = prepare_retry(failures, 3).expect("failures are retriable");
        assert_eq!(
            recovery
                .units
                .iter()
                .map(|unit| unit.sequence)
                .collect::<Vec<_>>(),
            vec![0, 1]
        );
        let mut remaining = deferred.into_iter().collect::<Vec<_>>();
        remaining.extend(recovery.units);
        remaining.sort_by_key(|unit| unit.sequence);
        let (retry_wave, still_deferred) =
            select_wave(remaining.into(), &routes, 4).expect("select retry wave");
        assert_eq!(
            retry_wave
                .iter()
                .map(|unit| unit.sequence)
                .collect::<Vec<_>>(),
            vec![0, 1]
        );
        assert_eq!(
            still_deferred
                .iter()
                .map(|unit| unit.sequence)
                .collect::<Vec<_>>(),
            vec![2],
            "the failed partition-0 unit must settle before sequence 2 can issue"
        );
    }
}