krabka_client_producer/

transactional.rs

//! Client-side transactional state machine. It drives the
//! `init_transactions` / `begin` / `commit` / `abort` / `send_offsets_to_transaction`
//! flow.

use std::{
    fmt,
    str::FromStr,
    sync::{
        Arc,
        atomic::{AtomicI32, Ordering},
    },
};

use crate::{error::ProducerError, producer::Producer};

/// The slot holds no error. An abortable error code fits in an `i16`, and
/// `-1` is `UNKNOWN_SERVER_ERROR`, so the empty value is outside that range.
const NO_TXN_ERROR: i32 = i32::MIN;

/// The sentinel that marks the slot as holding an abortable timeout rather
/// than a broker error code. It is one above `NO_TXN_ERROR` (`i32::MIN`), so
/// it stays outside the `i16` code range and cannot collide with a real,
/// sign-extended code.
const ABORTABLE_TIMEOUT: i32 = i32::MIN + 1;

/// The sentinel that marks the slot as holding a fence.
const FATAL_FENCED: i32 = i32::MIN + 2;

/// The offset that moves a fatal error code out of the abortable code range.
/// A fatal code `c` is stored as `c + FATAL_CODE_OFFSET`, which is in
/// `[0x1_0000, 0x1_ffff]`.
const FATAL_CODE_OFFSET: i32 = 0x1_8000;

/// The lowest stored value of a fatal error code: `i16::MIN +
/// FATAL_CODE_OFFSET`.
const FATAL_CODE_MIN: i32 = 0x1_0000;

/// The error that makes the current transaction abort-only.
///
/// Apache Kafka's `TransactionManager` moves to the `ABORTABLE_ERROR` state
/// when a batch of the transaction fails, or when a coordinator answers with a
/// code that only an abort can clear (`abortableError`). Every later
/// `commitTransaction` and `send` then fails with the stored error
/// (`maybeFailWithError`), and only `abortTransaction` clears the state. Kafka
/// stores the raised exception itself, which is either a broker error code
/// wrapped in a `KafkaException`, or a client-side `TimeoutException` that
/// carries no code (a batch that ran out of retries or its routing budget
/// with no broker answer).
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum AbortableError {
    /// A coordinator, or a Produce response, answered with this code.
    Server(i16),
    /// A batch of the transaction was never acknowledged, and carries no
    /// broker code. Kafka raises `TimeoutException` for the same case.
    Timeout,
}

/// The error that stops every later transactional operation.
///
/// Apache Kafka's `TransactionManager` moves to the `FATAL_ERROR` state when a
/// coordinator fences the producer, or answers with a code that its handler
/// passes to `fatalError`. `FATAL_ERROR` is never a valid start state of a
/// transition, and `maybeFailWithError` throws in `initTransactions`,
/// `beginTransaction`, `send`, `sendOffsetsToTransaction`,
/// `commitTransaction` and `abortTransaction`. The application can only close
/// the producer.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum FatalError {
    /// A newer epoch owns the transactional id (`PRODUCER_FENCED` or
    /// `INVALID_PRODUCER_EPOCH`).
    Fenced,
    /// A coordinator answered with this code.
    Server(i16),
}

impl FatalError {
    /// The transaction state after this error.
    pub(crate) const fn state(self) -> TxnState {
        match self {
            Self::Fenced => TxnState::Fenced,
            Self::Server(_) => TxnState::FatalError,
        }
    }

    /// The error that an operation reports in this state.
    pub(crate) const fn error(self) -> ProducerError {
        match self {
            Self::Fenced => ProducerError::FencedProducer,
            Self::Server(code) => ProducerError::FatalTransactionError(code),
        }
    }
}

/// The error state of the transaction: none, abortable, or fatal.
///
/// The sender task writes this slot from a synchronous context, so it holds an
/// atomic rather than a mutex. A fatal error is final: a later abortable error
/// does not replace it, and a clear does not remove it, as Kafka's
/// `TransactionManager` allows no transition out of `FATAL_ERROR`.
#[derive(Debug)]
pub struct TxnErrorSlot(AtomicI32);

impl Default for TxnErrorSlot {
    fn default() -> Self {
        Self(AtomicI32::new(NO_TXN_ERROR))
    }
}

impl TxnErrorSlot {
    /// Store `value` unless the slot holds a fatal error.
    fn store_abortable(&self, value: i32) {
        let _ = self
            .0
            .fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
                (!is_fatal(current)).then_some(value)
            });
    }

    /// Store `code` as the error that the application must abort. A later
    /// error replaces an earlier one, as Kafka's `transitionTo` replaces
    /// `lastError`. It has no effect after a fatal error.
    pub(crate) fn set_abortable(&self, code: i16) {
        self.store_abortable(i32::from(code));
    }

    /// Store that the transaction can no longer commit because a batch of it
    /// timed out with no broker code. A later error replaces an earlier one,
    /// same as [`Self::set_abortable`].
    pub(crate) fn set_abortable_timeout(&self) {
        self.store_abortable(ABORTABLE_TIMEOUT);
    }

    /// Store the fatal error `error`. It replaces any earlier error, as
    /// Kafka's `transitionToFatalError` replaces `lastError`.
    pub(crate) fn set_fatal(&self, error: FatalError) {
        let value = match error {
            FatalError::Fenced => FATAL_FENCED,
            FatalError::Server(code) => i32::from(code) + FATAL_CODE_OFFSET,
        };
        self.0.store(value, Ordering::Release);
    }

    /// The stored abortable error, or `None` when the transaction can still
    /// commit or the error is fatal.
    pub(crate) fn abortable(&self) -> Option<AbortableError> {
        match self.0.load(Ordering::Acquire) {
            ABORTABLE_TIMEOUT => Some(AbortableError::Timeout),
            value => i16::try_from(value).ok().map(AbortableError::Server),
        }
    }

    /// The stored fatal error, or `None` when there is none.
    pub(crate) fn fatal(&self) -> Option<FatalError> {
        match self.0.load(Ordering::Acquire) {
            FATAL_FENCED => Some(FatalError::Fenced),
            value if value >= FATAL_CODE_MIN => i16::try_from(value - FATAL_CODE_OFFSET)
                .ok()
                .map(FatalError::Server),
            _ => None,
        }
    }

    /// Forget the stored abortable error. An abort and a new producer identity
    /// both clear it, as Kafka's `resetTransactionState` and
    /// `InitProducerIdHandler` clear `lastError`. A fatal error stays.
    pub(crate) fn clear_abortable(&self) {
        self.store_abortable(NO_TXN_ERROR);
    }
}

/// Tell if a stored slot value is a fatal error.
const fn is_fatal(value: i32) -> bool {
    value == FATAL_FENCED || value >= FATAL_CODE_MIN
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[allow(dead_code)]
pub enum TxnState {
    /// The caller has not yet called `init_transactions`.
    Uninitialized,
    /// `init_transactions` succeeded, and no txn is in flight.
    Ready,
    /// Inside `begin_transaction` ... `commit/abort`.
    InTransaction,
    /// `prepare_transaction` has stopped new writes and is flushing records.
    Preparing,
    /// A 2PC transaction has been flushed and awaits its external decision.
    Prepared,
    /// `init_transactions_with_keep_prepared` is in flight.
    Initializing,
    /// A `commit` or an `abort` is in progress.
    CommittingOrAborting,
    /// A guard was dropped or `EndTxn` had an uncertain transport outcome.
    /// `init_transactions` must establish a new epoch before reuse.
    RecoveryRequired,
    /// A newer epoch owns the transactional id. Every later transactional
    /// operation fails with `FencedProducer`, as Kafka's `FATAL_ERROR` state
    /// does for `ProducerFencedException`.
    Fenced,
    /// A coordinator answered with a fatal code. Every later transactional
    /// operation fails with `FatalTransactionError`, as Kafka's `FATAL_ERROR`
    /// state does.
    FatalError,
}

/// Stable identity of a transaction prepared for external two-phase commit.
///
/// Its string form is Kafka-compatible: `"producer_id:producer_epoch"`.
/// The empty string represents no transaction and round-trips through
/// [`Default`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub struct PreparedTransactionState {
    producer_id: i64,
    producer_epoch: i16,
}

impl PreparedTransactionState {
    /// Creates a prepared transaction identity.
    ///
    /// # Errors
    ///
    /// Returns [`PreparedTransactionStateParseError`] when either identity
    /// component is negative.
    pub fn new(
        producer_id: i64,
        producer_epoch: i16,
    ) -> Result<Self, PreparedTransactionStateParseError> {
        if producer_id < 0 || producer_epoch < 0 {
            return Err(PreparedTransactionStateParseError);
        }
        Ok(Self {
            producer_id,
            producer_epoch,
        })
    }

    /// Producer ID of the prepared transaction.
    #[must_use]
    pub const fn producer_id(self) -> i64 {
        self.producer_id
    }

    /// Producer epoch of the prepared transaction.
    #[must_use]
    pub const fn producer_epoch(self) -> i16 {
        self.producer_epoch
    }

    /// Whether this token identifies a real transaction.
    #[must_use]
    pub const fn has_transaction(self) -> bool {
        self.producer_id >= 0
    }
}

impl Default for PreparedTransactionState {
    fn default() -> Self {
        Self {
            producer_id: -1,
            producer_epoch: -1,
        }
    }
}

impl fmt::Display for PreparedTransactionState {
    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
        if self.has_transaction() {
            write!(formatter, "{}:{}", self.producer_id, self.producer_epoch)
        } else {
            Ok(())
        }
    }
}

impl FromStr for PreparedTransactionState {
    type Err = PreparedTransactionStateParseError;

    fn from_str(serialized: &str) -> Result<Self, Self::Err> {
        if serialized.is_empty() {
            return Ok(Self::default());
        }
        let (producer_id, producer_epoch) = serialized
            .split_once(':')
            .ok_or(PreparedTransactionStateParseError)?;
        if producer_epoch.contains(':') {
            return Err(PreparedTransactionStateParseError);
        }
        Self::new(
            producer_id
                .parse()
                .map_err(|_| PreparedTransactionStateParseError)?,
            producer_epoch
                .parse()
                .map_err(|_| PreparedTransactionStateParseError)?,
        )
    }
}

/// Error returned when a prepared transaction token is malformed.
#[derive(Debug, Clone, Copy, PartialEq, Eq, thiserror::Error)]
#[error("invalid prepared transaction state; expected producer_id:producer_epoch")]
pub struct PreparedTransactionStateParseError;

/// An open transaction, borrowing the [`Producer`] that opened it.
///
/// [`Producer::begin_transaction`] returns it. [`commit`](Self::commit) and
/// [`abort`](Self::abort) each consume `self` on success, so a transaction
/// cannot be silently reused or finished twice.
///
/// On failure the call hands the guard back through
/// [`EndTransactionError::transaction`] instead of dropping it. Kafka's
/// `EndTxn` contract makes some failures, such as `CONCURRENT_TRANSACTIONS`,
/// retryable against the very same broker-side transaction, so the caller can
/// retry `commit()` on the returned guard, or switch to `abort()`. For
/// non-retryable failures the producer's transaction state has already moved
/// on, and the returned guard's next `commit` or `abort` attempt fails
/// immediately.
///
/// A dropped unresolved guard marks the producer as recovery-required. The
/// producer never guesses whether Kafka committed or aborted the transaction.
/// The caller must call `init_transactions` before the producer can send or
/// begin again.
#[derive(Debug)]
#[must_use = "a transaction must be finished with `commit()` or `abort()`"]
pub struct Transaction<'p> {
    pub(crate) producer: &'p Producer,
    pub(crate) finished: bool,
    pub(crate) guard_generation: u64,
}

impl Transaction<'_> {
    /// Flushes and prepares this transaction for external two-phase commit.
    ///
    /// # Errors
    ///
    /// See [`Producer::prepare_transaction`].
    pub async fn prepare(&self) -> Result<PreparedTransactionState, ProducerError> {
        self.producer.prepare_transaction().await
    }

    /// Commit this transaction.
    ///
    /// # Errors
    ///
    /// See [`Producer::begin_transaction`] for the shared error conditions.
    /// On failure, `self` is returned via [`EndTransactionError::transaction`]
    /// so a retryable failure can be retried or aborted.
    pub async fn commit(mut self) -> Result<(), EndTransactionError<Self>> {
        match self.producer.end_transaction(true).await {
            Ok(()) => {
                self.finished = true;
                Ok(())
            }
            Err(source) => Err(EndTransactionError {
                transaction: self,
                source,
            }),
        }
    }

    /// Abort this transaction.
    ///
    /// # Errors
    ///
    /// See [`Producer::begin_transaction`] for the shared error conditions.
    /// On failure, `self` is returned via [`EndTransactionError::transaction`]
    /// so a retryable failure can be retried or aborted.
    pub async fn abort(mut self) -> Result<(), EndTransactionError<Self>> {
        match self.producer.end_transaction(false).await {
            Ok(()) => {
                self.finished = true;
                Ok(())
            }
            Err(source) => Err(EndTransactionError {
                transaction: self,
                source,
            }),
        }
    }
}

impl Drop for Transaction<'_> {
    fn drop(&mut self) {
        if !self.finished {
            self.producer
                .abandon_transaction_guard(self.guard_generation);
        }
    }
}

/// Same contract as [`Transaction`], but owns an `Arc<Producer>` instead of
/// borrowing it.
///
/// Use it when the caller must hold the guard across an owned or `'static`
/// boundary that a borrow cannot survive, for example behind a `dyn Trait`
/// object stored in a struct field across many separate async calls.
/// [`Producer::begin_transaction_owned`] returns it. It mirrors
/// `tokio::sync::Mutex::{lock, lock_owned}` and `MutexGuard`/`OwnedMutexGuard`.
#[derive(Debug)]
#[must_use = "a transaction must be finished with `commit()` or `abort()`"]
pub struct OwnedTransaction {
    pub(crate) producer: Arc<Producer>,
    pub(crate) finished: bool,
    pub(crate) guard_generation: u64,
}

impl OwnedTransaction {
    /// Flushes and prepares this transaction for external two-phase commit.
    ///
    /// # Errors
    ///
    /// See [`Producer::prepare_transaction`].
    pub async fn prepare(&self) -> Result<PreparedTransactionState, ProducerError> {
        self.producer.prepare_transaction().await
    }

    /// Commit this transaction.
    ///
    /// # Errors
    ///
    /// See [`Producer::begin_transaction`] for the shared error conditions.
    /// On failure, `self` is returned via [`EndTransactionError::transaction`]
    /// so a retryable failure can be retried or aborted.
    pub async fn commit(mut self) -> Result<(), EndTransactionError<Self>> {
        match self.producer.end_transaction(true).await {
            Ok(()) => {
                self.finished = true;
                Ok(())
            }
            Err(source) => Err(EndTransactionError {
                transaction: self,
                source,
            }),
        }
    }

    /// Abort this transaction.
    ///
    /// # Errors
    ///
    /// See [`Producer::begin_transaction`] for the shared error conditions.
    /// On failure, `self` is returned via [`EndTransactionError::transaction`]
    /// so a retryable failure can be retried or aborted.
    pub async fn abort(mut self) -> Result<(), EndTransactionError<Self>> {
        match self.producer.end_transaction(false).await {
            Ok(()) => {
                self.finished = true;
                Ok(())
            }
            Err(source) => Err(EndTransactionError {
                transaction: self,
                source,
            }),
        }
    }
}

impl Drop for OwnedTransaction {
    fn drop(&mut self) {
        if !self.finished {
            self.producer
                .abandon_transaction_guard(self.guard_generation);
        }
    }
}

/// Error returned by [`Transaction::commit`], [`abort`](Transaction::abort),
/// and the [`OwnedTransaction`] equivalents.
///
/// It carries the guard back, so the caller can retry or abort a retryable
/// failure, such as `CONCURRENT_TRANSACTIONS`, on the same underlying
/// transaction instead of stranding it.
#[derive(Debug, thiserror::Error)]
#[error("{source}")]
pub struct EndTransactionError<T> {
    /// The guard the `commit` or `abort` call was made on, handed back so the
    /// caller can retry `commit()` or call `abort()` on the same
    /// transaction.
    pub transaction: T,
    /// The underlying failure.
    #[source]
    pub source: ProducerError,
}

#[cfg(test)]
mod tests {
    use std::{
        collections::VecDeque,
        iter,
        sync::{
            Arc, Mutex,
            atomic::{AtomicBool, AtomicI16, AtomicU16, AtomicUsize, Ordering},
        },
        time::Duration,
    };

    use bytes::BytesMut;
    use futures::future::join_all;
    use krabka_client_core::{ClientError, MockBroker};
    use krabka_protocol::{
        Decode, Encode,
        owned::{
            add_offsets_to_txn_request,
            add_offsets_to_txn_response::{self, AddOffsetsToTxnResponse},
            add_partitions_to_txn_request,
            add_partitions_to_txn_response::{self, AddPartitionsToTxnResponse},
            api_versions_request,
            api_versions_response::{ApiVersion, ApiVersionsResponse},
            common::add_partitions_to_txn_response::{
                add_partitions_to_txn_partition_result::AddPartitionsToTxnPartitionResult,
                add_partitions_to_txn_topic_result::AddPartitionsToTxnTopicResult,
            },
            end_txn_request,
            end_txn_response::{self, EndTxnResponse},
            find_coordinator_request::{self, FindCoordinatorRequest},
            find_coordinator_response::FindCoordinatorResponse,
            init_producer_id_request::{self, InitProducerIdRequest},
            init_producer_id_response::{self, InitProducerIdResponse},
            metadata_request,
            metadata_response::{
                self, MetadataResponse, MetadataResponseBroker, MetadataResponsePartition,
                MetadataResponseTopic,
            },
            produce_request,
            produce_response::{
                self, PartitionProduceResponse, ProduceResponse, TopicProduceResponse,
            },
            sasl_handshake_request,
            txn_offset_commit_request::{
                self, TxnOffsetCommitRequest, TxnOffsetCommitRequestPartition,
                TxnOffsetCommitRequestTopic,
            },
            txn_offset_commit_response::{
                self, TxnOffsetCommitResponse, TxnOffsetCommitResponsePartition,
                TxnOffsetCommitResponseTopic,
            },
        },
        primitives::uuid::Uuid,
    };
    use tokio::time;

    use super::{AbortableError, EndTransactionError, PreparedTransactionState, TxnState};
    use crate::{ProducerRecord, error::ProducerError, producer::Producer};

    #[test]
    fn prepared_transaction_state_has_stable_string_round_trip() {
        let state = PreparedTransactionState::new(42, 7).expect("valid transaction identity");

        assert2::assert!(state.producer_id() == 42);
        assert2::assert!(state.producer_epoch() == 7);
        assert2::assert!(state.has_transaction());
        assert2::assert!(state.to_string() == "42:7");
        assert2::assert!("42:7".parse::<PreparedTransactionState>() == Ok(state));
        assert2::assert!(PreparedTransactionState::default().to_string().is_empty());
        assert2::assert!(
            "".parse::<PreparedTransactionState>() == Ok(PreparedTransactionState::default())
        );
    }

    #[test]
    fn prepared_transaction_state_rejects_malformed_or_negative_identity() {
        for serialized in ["42", "-1:0", "1:-1", "1:2:3", "a:2", "1:b"] {
            assert2::assert!(serialized.parse::<PreparedTransactionState>().is_err());
        }
    }

    /// Tell if an `InitProducerId` request body carries a `transactional.id`.
    ///
    /// The builder of an idempotent producer sends one request without it, and
    /// `init_transactions` sends the requests with it. The body starts with the
    /// request header client id, and a flexible version adds a tagged-field
    /// byte after it.
    fn is_transactional_init(body: &[u8], version: i16) -> bool {
        decode_request::<InitProducerIdRequest>(
            body,
            version,
            init_producer_id_request::FLEXIBLE_MIN,
        )
        .transactional_id
        .is_some()
    }

    /// Decode the request of a request body at `version`. The body starts
    /// with the request header client id, and a flexible version adds a
    /// tagged-field byte after it.
    fn decode_request<R: for<'a> Decode<'a>>(body: &[u8], version: i16, flexible_min: i16) -> R {
        let client_id_len = usize::try_from(i16::from_be_bytes([body[0], body[1]]).max(0))
            .expect("non-negative client id length");
        let flexible = usize::from(version >= flexible_min);
        let mut request = &body[2 + client_id_len + flexible..];
        R::decode(&mut request, version).expect("decode request")
    }

    fn encode_v0(resp: &impl Encode) -> Vec<u8> {
        let mut buf = BytesMut::new();
        resp.encode(&mut buf, 0).unwrap();
        buf.to_vec()
    }

    /// Boots a mock broker that also answers as its own transaction
    /// coordinator, so `FindCoordinator` resolves back to the mock's own
    /// address. It returns a transactional `Producer` with `init_transactions`
    /// already completed against that broker.
    ///
    /// `end_txn_error` lets each test steer the `error_code` of the `EndTxn`
    /// response independently per call, where 0 means success. A test can
    /// therefore fail a `commit` or `abort`, and then flip the mock so that a
    /// retry on the same guard succeeds.
    async fn transactional_producer(end_txn_error: Arc<AtomicI16>) -> (MockBroker, Producer) {
        transactional_producer_with_end_txn_timeout(end_txn_error, Arc::new(AtomicBool::new(false)))
            .await
    }

    #[tokio::test]
    async fn init_transactions_retries_with_configured_policy() {
        let port_cell = Arc::new(AtomicU16::new(0));
        let handler_port = Arc::clone(&port_cell);
        let attempts = Arc::new(AtomicU16::new(0));
        let observed = Arc::clone(&attempts);
        let mock = MockBroker::start(move |api_key, version, _corr_id, body| {
            if api_key == api_versions_request::API_KEY {
                return Some(encode_v0(&ApiVersionsResponse::default()));
            }
            if api_key == find_coordinator_request::API_KEY {
                return Some(encode_v0(&FindCoordinatorResponse {
                    error_code: 0,
                    node_id: 1,
                    host: "127.0.0.1".into(),
                    port: i32::from(handler_port.load(Ordering::SeqCst)),
                    ..Default::default()
                }));
            }
            if api_key == init_producer_id_request::API_KEY {
                if !is_transactional_init(body, version) {
                    return Some(encode_v0(&InitProducerIdResponse {
                        producer_id: 1,
                        ..Default::default()
                    }));
                }
                let attempt = observed.fetch_add(1, Ordering::SeqCst);
                return Some(encode_v0(&InitProducerIdResponse {
                    error_code: if attempt == 0 { 14 } else { 0 },
                    producer_id: 7,
                    producer_epoch: 3,
                    ..Default::default()
                }));
            }
            None
        })
        .await
        .unwrap();
        port_cell.store(mock.addr.port(), Ordering::SeqCst);

        let producer = Producer::builder()
            .bootstrap(mock.addr.to_string())
            .transactional_id("test-txn")
            .request_timeout(Duration::from_millis(100))
            .retry_backoff(Duration::from_millis(1))
            .init_retry_timeout(Duration::from_millis(100))
            .retry_backoff_max(Duration::from_millis(1))
            .build()
            .await
            .expect("producer connects");

        producer
            .init_transactions()
            .await
            .expect("cold coordinator retry succeeds");
        drop(producer);
        assert2::assert!(attempts.load(Ordering::SeqCst) == 2);
        mock.stop();
    }

    async fn transactional_producer_with_end_txn_timeout(
        end_txn_error: Arc<AtomicI16>,
        end_txn_silent: Arc<AtomicBool>,
    ) -> (MockBroker, Producer) {
        transactional_producer_configured(end_txn_error, end_txn_silent, false).await
    }

    async fn transactional_producer_configured(
        end_txn_error: Arc<AtomicI16>,
        end_txn_silent: Arc<AtomicBool>,
        two_phase_commit_enabled: bool,
    ) -> (MockBroker, Producer) {
        let port_cell = Arc::new(AtomicU16::new(0));
        let handler_port = Arc::clone(&port_cell);
        let next_epoch = Arc::new(AtomicI16::new(3));
        let handler_epoch = Arc::clone(&next_epoch);
        let mock = MockBroker::start(move |api_key, version, _corr_id, body| {
            if api_key == api_versions_request::API_KEY {
                return Some(encode_v0(&ApiVersionsResponse {
                    api_keys: vec![ApiVersion {
                        api_key: init_producer_id_request::API_KEY,
                        min_version: 0,
                        max_version: if two_phase_commit_enabled { 6 } else { 0 },
                        ..Default::default()
                    }],
                    ..Default::default()
                }));
            }
            if api_key == find_coordinator_request::API_KEY {
                return Some(encode_v0(&FindCoordinatorResponse {
                    error_code: 0,
                    node_id: 1,
                    host: "127.0.0.1".into(),
                    port: i32::from(handler_port.load(Ordering::SeqCst)),
                    ..Default::default()
                }));
            }
            if api_key == init_producer_id_request::API_KEY {
                let response = if is_transactional_init(body, version) {
                    InitProducerIdResponse {
                        error_code: 0,
                        producer_id: 7,
                        producer_epoch: handler_epoch.fetch_add(1, Ordering::SeqCst),
                        ..Default::default()
                    }
                } else {
                    InitProducerIdResponse {
                        producer_id: 1,
                        ..Default::default()
                    }
                };
                let mut buf = BytesMut::new();
                if version >= init_producer_id_response::FLEXIBLE_MIN {
                    buf.extend_from_slice(&[0]);
                }
                response.encode(&mut buf, version).unwrap();
                return Some(buf.to_vec());
            }
            if api_key == end_txn_request::API_KEY {
                if end_txn_silent.load(Ordering::SeqCst) {
                    return None;
                }
                return Some(encode_v0(&EndTxnResponse {
                    error_code: end_txn_error.load(Ordering::SeqCst),
                    ..Default::default()
                }));
            }
            None
        })
        .await
        .unwrap();
        port_cell.store(mock.addr.port(), Ordering::SeqCst);

        let producer = Producer::builder()
            .bootstrap(mock.addr.to_string())
            .transactional_id("test-txn")
            .transaction_two_phase_commit_enable(two_phase_commit_enabled)
            .request_timeout(Duration::from_millis(100))
            .retry_backoff(Duration::from_millis(1))
            .init_retry_timeout(Duration::from_millis(400))
            .retry_backoff_max(Duration::from_millis(20))
            .build()
            .await
            .expect("producer connects to the mock");
        producer
            .init_transactions()
            .await
            .expect("init_transactions against the mock coordinator");
        (mock, producer)
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn prepared_guard_can_drop_without_poisoning_the_next_transaction() {
        let end_txn_error = Arc::new(AtomicI16::new(0));
        let (mock, producer) = transactional_producer_configured(
            end_txn_error,
            Arc::new(AtomicBool::new(false)),
            true,
        )
        .await;
        let transaction = producer
            .begin_transaction()
            .await
            .expect("begin transaction");
        let prepared = transaction.prepare().await.expect("prepare transaction");

        assert2::assert!(*producer.txn_state.lock().await == TxnState::Prepared);
        drop(transaction);
        assert2::assert!(!producer.txn_recovery_required.load(Ordering::Acquire));
        producer
            .complete_transaction(prepared)
            .await
            .expect("complete prepared transaction");
        producer
            .begin_transaction()
            .await
            .expect("begin next transaction")
            .abort()
            .await
            .expect("abort next transaction");
        mock.stop();
    }

    macro_rules! end_txn_retry_test {
        ($name:ident, borrowed, $finish:ident) => {
            #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
            async fn $name() {
                let end_txn_error = Arc::new(AtomicI16::new(51));
                let (mock, producer) = transactional_producer(end_txn_error.clone()).await;
                let txn = producer
                    .begin_transaction()
                    .await
                    .expect("begin_transaction");

                let EndTransactionError {
                    transaction,
                    source,
                } = txn
                    .$finish()
                    .await
                    .expect_err("broker reported CONCURRENT_TRANSACTIONS");
                assert2::assert!(matches!(source, ProducerError::ConcurrentTransactions));

                end_txn_error.store(0, Ordering::SeqCst);
                transaction.$finish().await.expect("retry succeeds");
                mock.stop();
            }
        };
        ($name:ident, owned, $finish:ident) => {
            #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
            async fn $name() {
                let end_txn_error = Arc::new(AtomicI16::new(51));
                let (mock, producer) = transactional_producer(end_txn_error.clone()).await;
                let producer = Arc::new(producer);
                let txn = producer
                    .clone()
                    .begin_transaction_owned()
                    .await
                    .expect("begin_transaction_owned");

                let EndTransactionError {
                    transaction,
                    source,
                } = txn
                    .$finish()
                    .await
                    .expect_err("broker reported CONCURRENT_TRANSACTIONS");
                assert2::assert!(matches!(source, ProducerError::ConcurrentTransactions));

                end_txn_error.store(0, Ordering::SeqCst);
                transaction.$finish().await.expect("retry succeeds");
                mock.stop();
            }
        };
    }

    // CONCURRENT_TRANSACTIONS (51) proves `commit`/`abort` drive the broker
    // round trip, and the returned guard remains usable once the broker clears
    // the condition.
    end_txn_retry_test!(
        transaction_commit_reports_broker_error_and_retries_on_the_same_guard,
        borrowed,
        commit
    );
    end_txn_retry_test!(
        transaction_abort_reports_broker_error_and_retries_on_the_same_guard,
        borrowed,
        abort
    );
    end_txn_retry_test!(
        owned_transaction_commit_reports_broker_error_and_retries_on_the_same_guard,
        owned,
        commit
    );
    end_txn_retry_test!(
        owned_transaction_abort_reports_broker_error_and_retries_on_the_same_guard,
        owned,
        abort
    );

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn uncertain_end_txn_requires_reinitialization_before_reuse() {
        let end_txn_error = Arc::new(AtomicI16::new(0));
        let end_txn_silent = Arc::new(AtomicBool::new(true));
        let (mock, producer) =
            transactional_producer_with_end_txn_timeout(end_txn_error, Arc::clone(&end_txn_silent))
                .await;

        let EndTransactionError {
            transaction,
            source,
        } = producer
            .begin_transaction()
            .await
            .expect("begin transaction")
            .commit()
            .await
            .expect_err(
                "EndTxn that stays silent past the retry deadline has an uncertain outcome",
            );
        assert2::assert!(let ProducerError::RecoveryRequired = source);
        drop(transaction);

        let begin = producer.begin_transaction().await.err();
        assert2::assert!(let Some(ProducerError::RecoveryRequired) = begin);
        let acknowledgement = producer.send(ProducerRecord::default()).await;
        assert2::assert!(
            let Err(ProducerError::RecoveryRequired) =
                acknowledgement
        );

        end_txn_silent.store(false, Ordering::SeqCst);
        producer.next_seq.insert(("topic".to_owned(), 0), 5);
        producer
            .init_transactions()
            .await
            .expect("reinitialization obtains a new epoch");
        assert2::assert!(*producer.txn_pid_epoch.lock().await == (7, 4));
        // Kafka's `TransactionManager` starts the sequences again at a new epoch.
        assert2::assert!(producer.next_seq.is_empty());
        assert2::assert!(producer.transactional_identity().await == Some((7, 4)));
        producer
            .begin_transaction()
            .await
            .expect("new epoch permits a transaction")
            .abort()
            .await
            .expect("abort after recovery");
        mock.stop();
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn dropping_an_open_transaction_requires_explicit_recovery() {
        let end_txn_error = Arc::new(AtomicI16::new(0));
        let (mock, producer) = transactional_producer(end_txn_error).await;
        drop(
            producer
                .begin_transaction()
                .await
                .expect("begin transaction before drop"),
        );

        assert!(matches!(
            producer.begin_transaction().await,
            Err(ProducerError::RecoveryRequired)
        ));
        producer
            .init_transactions()
            .await
            .expect("explicit initialization recovers dropped transaction");
        producer
            .begin_transaction()
            .await
            .expect("begin after recovery")
            .abort()
            .await
            .expect("abort after recovery");
        mock.stop();
    }

    /// One scripted answer from the mock coordinator.
    #[derive(Debug, Clone, Copy)]
    enum Reply {
        /// Send no response, so the request times out in transport.
        Silent,
        /// Answer with this error code.
        Code(i16),
    }

    /// The scripted answers for one API, and the number of requests it got.
    #[derive(Debug)]
    struct Script {
        replies: VecDeque<Reply>,
        exhausted: Reply,
        requests: usize,
    }

    impl Default for Script {
        fn default() -> Self {
            Self::new(&[], Reply::Code(0))
        }
    }

    impl Script {
        fn new(replies: &[Reply], exhausted: Reply) -> Self {
            Self {
                replies: replies.iter().copied().collect(),
                exhausted,
                requests: 0,
            }
        }

        fn next(&mut self) -> Reply {
            self.requests += 1;
            self.replies.pop_front().unwrap_or(self.exhausted)
        }
    }

    /// The scripts of a mock transaction coordinator. The same mock is the
    /// group coordinator and the partition leader.
    #[derive(Debug, Default)]
    struct Coordinator {
        end_txn: Script,
        add_partitions: Script,
        add_offsets: Script,
        txn_offset_commit: Script,
        produce: Script,
        /// The answers to `FindCoordinator` for the transactional id.
        txn_lookup: Script,
        /// The answers to `FindCoordinator` for a group.
        group_lookup: Script,
        find_coordinator_requests: usize,
        /// The number of transactional `InitProducerId` requests.
        init_producer_id_requests: usize,
        /// The number of `Metadata` requests.
        metadata_requests: usize,
        /// The negotiated version of each `AddPartitionsToTxn` request.
        add_partitions_versions: Vec<i16>,
        /// The `AddPartitionsToTxn` version range that the mock advertises.
        /// A Kafka broker advertises 0 to 5.
        add_partitions_range: Option<(i16, i16)>,
        /// Build the producer with `transaction_two_phase_commit_enable`. The
        /// mock then advertises `InitProducerId` v6.
        two_phase_commit: bool,
        /// The linger of the producer. Zero when `None`.
        linger: Option<Duration>,
        /// The finalized `transaction.version` that the mock reports. With a
        /// value, the mock also advertises `AddOffsetsToTxn` v0 to v4, `EndTxn`
        /// v0 to v5 and `TxnOffsetCommit` up to `txn_offset_commit_max`, as a
        /// Kafka 4 broker does.
        transaction_version: Option<i16>,
        /// The API key and version of each request other than `ApiVersions`,
        /// `Metadata`, `FindCoordinator` and `InitProducerId`.
        transaction_requests: Vec<(i16, i16)>,
        /// The highest `TxnOffsetCommit` version that the mock advertises
        /// with a transaction version. v5 when `None`, as Kafka 4.3 does.
        txn_offset_commit_max: Option<i16>,
        /// The topic id that the mock metadata gives the topic `topic`.
        topic_id: Uuid,
        /// The negotiated version and the decoded topics of each
        /// `TxnOffsetCommit` request.
        txn_offset_commit_topics: Vec<(i16, Vec<TxnOffsetCommitRequestTopic>)>,
    }

    type SharedCoordinator = Arc<Mutex<Coordinator>>;

    /// How a coordinator request ended, in a form that tests can compare.
    #[derive(Debug, PartialEq, Eq)]
    enum TxnResult {
        Ok,
        Fenced,
        ConcurrentTransactions,
        Server(i16),
        /// The producer is in the fatal error state after this code.
        Fatal(i16),
        OutcomeUnknown,
        /// The broker supports no version that the client sends. The fields
        /// are the broker range and the client range.
        IncompatibleVersion((i16, i16), (i16, i16)),
        Other(String),
    }

    impl From<Result<(), ProducerError>> for TxnResult {
        fn from(result: Result<(), ProducerError>) -> Self {
            match result {
                Ok(()) => Self::Ok,
                Err(ProducerError::FencedProducer) => Self::Fenced,
                Err(ProducerError::ConcurrentTransactions) => Self::ConcurrentTransactions,
                Err(ProducerError::Server(code)) => Self::Server(code),
                Err(ProducerError::FatalTransactionError(code)) => Self::Fatal(code),
                Err(ProducerError::RecoveryRequired) => Self::OutcomeUnknown,
                Err(ProducerError::Client(ClientError::IncompatibleVersion {
                    broker_min,
                    broker_max,
                    client_min,
                    client_max,
                    ..
                })) => {
                    Self::IncompatibleVersion((broker_min, broker_max), (client_min, client_max))
                }
                Err(other) => Self::Other(other.to_string()),
            }
        }
    }

    /// Encode `response` at `version`, after the tagged-fields byte of
    /// response header v1 for a flexible version.
    fn encode_at(response: &impl Encode, version: i16, flexible_min: i16) -> Vec<u8> {
        let mut buf = BytesMut::new();
        if version >= flexible_min {
            buf.extend_from_slice(&[0]);
        }
        response.encode(&mut buf, version).expect("encode response");
        buf.to_vec()
    }

    /// The `ApiVersions` answer of the scripted mock coordinator, encoded at
    /// the request `version` (response header v0 for every version).
    fn scripted_api_versions(
        add_partitions_range: (i16, i16),
        two_phase_commit: bool,
        transaction_version: Option<i16>,
        txn_offset_commit_max: i16,
        version: i16,
    ) -> Vec<u8> {
        use krabka_protocol::owned::api_versions_response::FinalizedFeatureKey;
        let range = |api_key, max_version| ApiVersion {
            api_key,
            min_version: 0,
            max_version,
            ..Default::default()
        };
        let mut api_keys = vec![
            ApiVersion {
                api_key: add_partitions_to_txn_request::API_KEY,
                min_version: add_partitions_range.0,
                max_version: add_partitions_range.1,
                ..Default::default()
            },
            range(metadata_request::API_KEY, 12),
            ApiVersion {
                api_key: produce_request::API_KEY,
                min_version: 3,
                max_version: 13,
                ..Default::default()
            },
            range(
                init_producer_id_request::API_KEY,
                if two_phase_commit { 6 } else { 0 },
            ),
        ];
        let mut finalized_features = Vec::new();
        if let Some(level) = transaction_version {
            api_keys.extend([
                range(add_offsets_to_txn_request::API_KEY, 4),
                range(end_txn_request::API_KEY, 5),
                range(txn_offset_commit_request::API_KEY, txn_offset_commit_max),
            ]);
            finalized_features.push(FinalizedFeatureKey {
                name: "transaction.version".into(),
                max_version_level: level,
                min_version_level: level,
                ..Default::default()
            });
        }
        let mut buf = BytesMut::new();
        ApiVersionsResponse {
            api_keys,
            finalized_features_epoch: if transaction_version.is_some() { 1 } else { -1 },
            finalized_features,
            ..Default::default()
        }
        .encode(&mut buf, version.min(5))
        .expect("encode ApiVersions");
        buf.to_vec()
    }

    /// The `Metadata` answer of the scripted mock coordinator: the topic
    /// `topic` with one partition that the mock leads.
    ///
    /// The producer waits for metadata that holds the topic, so the answer
    /// must decode at the negotiated version.
    fn scripted_metadata(port: u16, topic_id: Uuid, version: i16) -> Vec<u8> {
        let response = MetadataResponse {
            brokers: vec![MetadataResponseBroker {
                node_id: 1,
                host: "127.0.0.1".into(),
                port: i32::from(port),
                ..Default::default()
            }],
            topics: vec![MetadataResponseTopic {
                name: Some("topic".into()),
                topic_id,
                partitions: vec![MetadataResponsePartition {
                    partition_index: 0,
                    leader_id: 1,
                    ..Default::default()
                }],
                ..Default::default()
            }],
            ..Default::default()
        };
        let mut buf = BytesMut::new();
        if version >= metadata_response::FLEXIBLE_MIN {
            buf.extend_from_slice(&[0]);
        }
        response.encode(&mut buf, version).expect("encode Metadata");
        buf.to_vec()
    }

    /// Record one `TxnOffsetCommit` request, and answer it from the script.
    /// The answer names each topic as the request did, and gives each
    /// partition the scripted code.
    fn scripted_txn_offset_commit(
        coordinator: &mut Coordinator,
        body: &[u8],
        version: i16,
    ) -> Option<Vec<u8>> {
        let request = decode_request::<TxnOffsetCommitRequest>(
            body,
            version,
            txn_offset_commit_request::FLEXIBLE_MIN,
        );
        coordinator
            .txn_offset_commit_topics
            .push((version, request.topics.clone()));
        let Reply::Code(error_code) = coordinator.txn_offset_commit.next() else {
            return None;
        };
        let topics = request
            .topics
            .iter()
            .map(|topic| TxnOffsetCommitResponseTopic {
                name: topic.name.clone(),
                topic_id: topic.topic_id,
                partitions: topic
                    .partitions
                    .iter()
                    .map(|partition| TxnOffsetCommitResponsePartition {
                        partition_index: partition.partition_index,
                        error_code,
                        ..Default::default()
                    })
                    .collect(),
                ..Default::default()
            })
            .collect();
        Some(encode_at(
            &TxnOffsetCommitResponse {
                topics,
                ..Default::default()
            },
            version,
            txn_offset_commit_response::FLEXIBLE_MIN,
        ))
    }

    /// Boot a mock coordinator that answers `EndTxn` and `AddPartitionsToTxn`
    /// from `coordinator`, and connect a transactional producer to it without
    /// calling `init_transactions`. The producer is left `Uninitialized`.
    async fn scripted_producer_raw(
        coordinator: Coordinator,
    ) -> (MockBroker, Producer, SharedCoordinator) {
        let port_cell = Arc::new(AtomicU16::new(0));
        let handler_port = Arc::clone(&port_cell);
        let shared = Arc::new(Mutex::new(coordinator));
        let handler_shared = Arc::clone(&shared);
        let (
            add_partitions_range,
            two_phase_commit,
            transaction_version,
            linger,
            txn_offset_commit_max,
            topic_id,
        ) = {
            let coordinator = shared.lock().expect("scripted coordinator");
            (
                coordinator.add_partitions_range.unwrap_or((0, 5)),
                coordinator.two_phase_commit,
                coordinator.transaction_version,
                coordinator.linger.unwrap_or(Duration::ZERO),
                coordinator.txn_offset_commit_max.unwrap_or(5),
                coordinator.topic_id,
            )
        };
        let mock = MockBroker::start(move |api_key, version, _corr_id, body| {
            if api_key == api_versions_request::API_KEY {
                return Some(scripted_api_versions(
                    add_partitions_range,
                    two_phase_commit,
                    transaction_version,
                    txn_offset_commit_max,
                    version,
                ));
            }
            let mut coordinator = handler_shared.lock().expect("scripted coordinator");
            if ![
                metadata_request::API_KEY,
                find_coordinator_request::API_KEY,
                init_producer_id_request::API_KEY,
            ]
            .contains(&api_key)
            {
                coordinator.transaction_requests.push((api_key, version));
            }
            if api_key == metadata_request::API_KEY {
                coordinator.metadata_requests += 1;
                return Some(scripted_metadata(
                    handler_port.load(Ordering::SeqCst),
                    topic_id,
                    version,
                ));
            }
            if api_key == produce_request::API_KEY {
                let Reply::Code(error_code) = coordinator.produce.next() else {
                    return None;
                };
                let response = ProduceResponse {
                    responses: vec![TopicProduceResponse {
                        name: "topic".into(),
                        partition_responses: vec![PartitionProduceResponse {
                            index: 0,
                            error_code,
                            base_offset: 0,
                            log_start_offset: -1,
                            ..Default::default()
                        }],
                        ..Default::default()
                    }],
                    ..Default::default()
                };
                let mut buf = BytesMut::new();
                if version >= produce_response::FLEXIBLE_MIN {
                    buf.extend_from_slice(&[0]);
                }
                response.encode(&mut buf, version).expect("encode Produce");
                return Some(buf.to_vec());
            }
            if api_key == add_offsets_to_txn_request::API_KEY {
                return match coordinator.add_offsets.next() {
                    Reply::Silent => None,
                    Reply::Code(error_code) => Some(encode_at(
                        &AddOffsetsToTxnResponse {
                            error_code,
                            ..Default::default()
                        },
                        version,
                        add_offsets_to_txn_response::FLEXIBLE_MIN,
                    )),
                };
            }
            if api_key == txn_offset_commit_request::API_KEY {
                return scripted_txn_offset_commit(&mut coordinator, body, version);
            }
            if api_key == find_coordinator_request::API_KEY {
                coordinator.find_coordinator_requests += 1;
                let request: FindCoordinatorRequest =
                    decode_request(body, version, find_coordinator_request::FLEXIBLE_MIN);
                let script = if request.key == "test-txn" {
                    &mut coordinator.txn_lookup
                } else {
                    &mut coordinator.group_lookup
                };
                let Reply::Code(error_code) = script.next() else {
                    return None;
                };
                return Some(encode_v0(&FindCoordinatorResponse {
                    error_code,
                    node_id: 1,
                    host: "127.0.0.1".into(),
                    port: i32::from(handler_port.load(Ordering::SeqCst)),
                    ..Default::default()
                }));
            }
            if api_key == init_producer_id_request::API_KEY {
                if is_transactional_init(body, version) {
                    coordinator.init_producer_id_requests += 1;
                }
                let mut buf = BytesMut::new();
                if version >= init_producer_id_response::FLEXIBLE_MIN {
                    buf.extend_from_slice(&[0]);
                }
                InitProducerIdResponse {
                    error_code: 0,
                    producer_id: 7,
                    producer_epoch: 3,
                    ..Default::default()
                }
                .encode(&mut buf, version)
                .expect("encode InitProducerId");
                return Some(buf.to_vec());
            }
            if api_key == end_txn_request::API_KEY {
                return match coordinator.end_txn.next() {
                    Reply::Silent => None,
                    Reply::Code(error_code) => Some(encode_at(
                        &EndTxnResponse {
                            error_code,
                            ..Default::default()
                        },
                        version,
                        end_txn_response::FLEXIBLE_MIN,
                    )),
                };
            }
            if api_key != add_partitions_to_txn_request::API_KEY {
                return None;
            }
            coordinator.add_partitions_versions.push(version);
            match coordinator.add_partitions.next() {
                Reply::Silent => None,
                Reply::Code(partition_error_code) => {
                    let response = AddPartitionsToTxnResponse {
                        results_by_topic_v3_and_below: vec![AddPartitionsToTxnTopicResult {
                            name: "topic".into(),
                            results_by_partition: vec![AddPartitionsToTxnPartitionResult {
                                partition_index: 0,
                                partition_error_code,
                                ..Default::default()
                            }],
                            ..Default::default()
                        }],
                        ..Default::default()
                    };
                    let mut buf = BytesMut::new();
                    if version >= add_partitions_to_txn_response::FLEXIBLE_MIN {
                        buf.extend_from_slice(&[0]);
                    }
                    response
                        .encode(&mut buf, version)
                        .expect("encode AddPartitionsToTxn");
                    Some(buf.to_vec())
                }
            }
        })
        .await
        .unwrap();
        port_cell.store(mock.addr.port(), Ordering::SeqCst);
        let producer = Producer::builder()
            .bootstrap(mock.addr.to_string())
            .transactional_id("test-txn")
            .transaction_two_phase_commit_enable(two_phase_commit)
            .linger(linger)
            .request_timeout(Duration::from_millis(100))
            .retry_backoff(Duration::from_millis(1))
            .init_retry_timeout(Duration::from_millis(1500))
            .retry_backoff_max(Duration::from_millis(20))
            .build()
            .await
            .expect("producer connects to the mock");
        (mock, producer, shared)
    }

    /// [`scripted_producer_raw`], then `init_transactions` against the mock
    /// coordinator, leaving the producer `Ready`.
    async fn scripted_producer(
        coordinator: Coordinator,
    ) -> (MockBroker, Producer, SharedCoordinator) {
        let (mock, producer, shared) = scripted_producer_raw(coordinator).await;
        producer
            .init_transactions()
            .await
            .expect("init_transactions against the mock coordinator");
        {
            let mut coordinator = shared.lock().expect("scripted coordinator");
            coordinator.find_coordinator_requests = 0;
            coordinator.init_producer_id_requests = 0;
        }
        (mock, producer, shared)
    }

    /// Boot a mock coordinator that answers `EndTxn` from `script`, and then
    /// with `exhausted` when the script is empty.
    async fn scripted_end_txn_producer(
        script: &[Reply],
        exhausted: Reply,
    ) -> (MockBroker, Producer, SharedCoordinator) {
        scripted_producer(Coordinator {
            end_txn: Script::new(script, exhausted),
            ..Coordinator::default()
        })
        .await
    }

    /// The observable result of one scripted commit.
    #[derive(Debug, PartialEq, Eq)]
    struct ScriptedCommit {
        result: TxnResult,
        end_txn_requests: usize,
        state: TxnState,
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn commit_learns_the_outcome_of_a_lost_end_txn() {
        use Reply::{Code, Silent};
        use TxnResult::{Fenced, OutcomeUnknown};
        let cases = [
            ("answered", vec![Code(0)], TxnResult::Ok, 1, TxnState::Ready),
            (
                "lost then committed",
                vec![Silent, Code(0)],
                TxnResult::Ok,
                2,
                TxnState::Ready,
            ),
            (
                "lost, prepare in progress, then committed",
                vec![Silent, Code(51), Code(51), Code(0)],
                TxnResult::Ok,
                4,
                TxnState::Ready,
            ),
            (
                "lost, coordinator loading, then committed",
                vec![Silent, Code(14), Code(15), Code(16), Code(0)],
                TxnResult::Ok,
                5,
                TxnState::Ready,
            ),
            (
                "loading then committed",
                vec![Code(14), Code(0)],
                TxnResult::Ok,
                2,
                TxnState::Ready,
            ),
            (
                "concurrent without a loss, then committed",
                vec![Code(51), Code(0)],
                TxnResult::Ok,
                2,
                TxnState::Ready,
            ),
            (
                "request timed out, then committed",
                vec![Code(7), Code(0)],
                TxnResult::Ok,
                2,
                TxnState::Ready,
            ),
            ("fenced", vec![Code(47)], Fenced, 1, TxnState::Fenced),
            (
                "producer fenced",
                vec![Code(90)],
                Fenced,
                1,
                TxnState::Fenced,
            ),
            (
                "invalid state is fatal",
                vec![Code(48)],
                TxnResult::Fatal(48),
                1,
                TxnState::FatalError,
            ),
            (
                "lost then fenced",
                vec![Silent, Code(47)],
                OutcomeUnknown,
                2,
                TxnState::RecoveryRequired,
            ),
            (
                "lost then producer fenced",
                vec![Silent, Code(90)],
                OutcomeUnknown,
                2,
                TxnState::RecoveryRequired,
            ),
            (
                "lost then invalid state",
                vec![Silent, Code(48)],
                OutcomeUnknown,
                2,
                TxnState::RecoveryRequired,
            ),
        ];
        for (name, script, result, end_txn_requests, state) in cases {
            let (mock, producer, coordinator) = scripted_end_txn_producer(&script, Code(0)).await;
            let outcome = producer
                .begin_transaction()
                .await
                .expect("begin transaction")
                .commit()
                .await
                .map_err(|error| error.source);
            let state_after = *producer.txn_state.lock().await;
            let actual = ScriptedCommit {
                result: outcome.into(),
                end_txn_requests: coordinator
                    .lock()
                    .expect("scripted coordinator")
                    .end_txn
                    .requests,
                state: state_after,
            };
            let expected = ScriptedCommit {
                result,
                end_txn_requests,
                state,
            };
            assert2::assert!(actual == expected, "{name}");
            mock.stop();
        }
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn committed_after_a_lost_end_txn_permits_the_next_transaction() {
        let (mock, producer, coordinator) =
            scripted_end_txn_producer(&[Reply::Silent], Reply::Code(0)).await;
        producer
            .begin_transaction()
            .await
            .expect("begin first transaction")
            .commit()
            .await
            .expect("retry learns that the first transaction committed");
        producer
            .begin_transaction()
            .await
            .expect("no reinitialization is needed after a learned outcome")
            .commit()
            .await
            .expect("second transaction commits");
        assert2::assert!(
            coordinator
                .lock()
                .expect("scripted coordinator")
                .end_txn
                .requests
                == 3
        );
        mock.stop();
    }

    /// The observable result of a coordinator request that retried until the
    /// deadline.
    #[derive(Debug, PartialEq, Eq)]
    struct DeadlineResult {
        result: TxnResult,
        retried: bool,
        state: TxnState,
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn end_txn_that_stays_unanswered_reports_the_last_answer_at_the_deadline() {
        use Reply::{Code, Silent};
        let cases = [
            (
                "stays lost",
                Silent,
                TxnResult::OutcomeUnknown,
                TxnState::RecoveryRequired,
            ),
            (
                "stays concurrent",
                Code(51),
                TxnResult::ConcurrentTransactions,
                TxnState::InTransaction,
            ),
            (
                "stays loading",
                Code(14),
                TxnResult::Server(14),
                TxnState::InTransaction,
            ),
        ];
        for (name, exhausted, result, state) in cases {
            let (mock, producer, coordinator) = scripted_end_txn_producer(&[], exhausted).await;
            let outcome = producer
                .begin_transaction()
                .await
                .expect("begin transaction")
                .commit()
                .await
                .map_err(|error| error.source);
            let state_after = *producer.txn_state.lock().await;
            let end_txn_requests = coordinator
                .lock()
                .expect("scripted coordinator")
                .end_txn
                .requests;
            let actual = DeadlineResult {
                result: outcome.into(),
                retried: end_txn_requests > 1,
                state: state_after,
            };
            let expected = DeadlineResult {
                result,
                retried: true,
                state,
            };
            assert2::assert!(actual == expected, "{name}");
            mock.stop();
        }
    }

    /// The observable result of one scripted `AddPartitionsToTxn`.
    #[derive(Debug, PartialEq, Eq)]
    struct ScriptedAddPartitions {
        result: TxnResult,
        add_partitions_requests: usize,
        coordinator_lookups: usize,
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn add_partitions_retries_the_codes_that_kafka_retries() {
        use Reply::{Code, Silent};
        use TxnResult::{Fenced, Server};
        let cases = [
            ("added", vec![Code(0)], TxnResult::Ok, 1, 0),
            (
                "concurrent, then added",
                vec![Code(51), Code(0)],
                TxnResult::Ok,
                2,
                0,
            ),
            (
                "loading, then added",
                vec![Code(14), Code(0)],
                TxnResult::Ok,
                2,
                0,
            ),
            (
                "moved, then added",
                vec![Code(16), Code(0)],
                TxnResult::Ok,
                2,
                1,
            ),
            (
                "lost, then added",
                vec![Silent, Code(0)],
                TxnResult::Ok,
                2,
                1,
            ),
            ("fenced", vec![Code(47)], Fenced, 1, 0),
            ("producer fenced", vec![Code(90)], Fenced, 1, 0),
            ("topic authorization", vec![Code(29)], Server(29), 1, 0),
            ("operation not attempted", vec![Code(55)], Server(55), 1, 0),
            ("invalid state", vec![Code(48)], TxnResult::Fatal(48), 1, 0),
            (
                "transactional id authorization",
                vec![Code(53)],
                TxnResult::Fatal(53),
                1,
                0,
            ),
        ];
        for (name, script, result, add_partitions_requests, coordinator_lookups) in cases {
            let (mock, producer, coordinator) = scripted_producer(Coordinator {
                add_partitions: Script::new(&script, Code(0)),
                ..Coordinator::default()
            })
            .await;
            let outcome = producer.register_transaction_partition("topic", 0).await;
            let actual = {
                let coordinator = coordinator.lock().expect("scripted coordinator");
                ScriptedAddPartitions {
                    result: outcome.into(),
                    add_partitions_requests: coordinator.add_partitions.requests,
                    coordinator_lookups: coordinator.find_coordinator_requests,
                }
            };
            let expected = ScriptedAddPartitions {
                result,
                add_partitions_requests,
                coordinator_lookups,
            };
            assert2::assert!(actual == expected, "{name}");
            mock.stop();
        }
    }

    /// The observable result of one scripted `AddPartitionsToTxn` version
    /// negotiation.
    #[derive(Debug, PartialEq, Eq)]
    struct ScriptedAddPartitionsVersion {
        result: TxnResult,
        versions: Vec<i16>,
    }

    /// Apache Kafka's client builds `AddPartitionsToTxn` with
    /// `AddPartitionsToTxnRequest.Builder.forClient`, which allows v3 at most.
    /// v4 and later carry the broker-to-broker form, which a broker answers
    /// only for a principal with `CLUSTER_ACTION`.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn add_partitions_to_txn_stops_at_the_last_client_version() {
        let cases = [
            ("broker stops at v3", (0, 3), TxnResult::Ok, vec![3]),
            ("broker supports v5", (0, 5), TxnResult::Ok, vec![3]),
            (
                "broker supports only the broker versions",
                (4, 5),
                TxnResult::IncompatibleVersion((4, 5), (0, 3)),
                Vec::new(),
            ),
        ];
        for (name, range, result, versions) in cases {
            let (mock, producer, coordinator) = scripted_producer(Coordinator {
                add_partitions_range: Some(range),
                ..Coordinator::default()
            })
            .await;
            let outcome = producer.register_transaction_partition("topic", 0).await;
            let actual = {
                let coordinator = coordinator.lock().expect("scripted coordinator");
                ScriptedAddPartitionsVersion {
                    result: outcome.into(),
                    versions: coordinator.add_partitions_versions.clone(),
                }
            };
            assert2::assert!(
                actual == ScriptedAddPartitionsVersion { result, versions },
                "{name}"
            );
            mock.stop();
        }
    }

    /// The observable result of one scripted `send_offsets_to_transaction`.
    #[derive(Debug, PartialEq, Eq)]
    struct ScriptedSendOffsets {
        result: TxnResult,
        add_offsets_requests: usize,
        txn_offset_commit_requests: usize,
        coordinator_lookups: usize,
        abortable_error: Option<i16>,
        state: TxnState,
    }

    /// Kafka's `AddOffsetsToTxnHandler` and `TxnOffsetCommitHandler` send the
    /// request again after a transport loss and after every retriable code,
    /// and they find the coordinator again for 15, 16 and, for
    /// `TxnOffsetCommit`, `REQUEST_TIMED_OUT`. An abortable code stops the
    /// transaction from committing, and 47 and 90 fence the producer.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn send_offsets_to_transaction_retries_the_codes_that_kafka_retries() {
        use Reply::{Code, Silent};
        use TxnResult::{Fenced, Server};
        let group = krabka_client_consumer::ConsumerGroupMetadata {
            group_id: "group-a".into(),
            generation_id: 3,
            member_id: "member-a".into(),
            group_instance_id: None,
        };
        // (name, add offsets script, offset commit script, result, add offsets
        // requests, offset commit requests, coordinator lookups, abortable,
        // transaction state)
        let cases = [
            (
                "committed",
                vec![Code(0)],
                vec![Code(0)],
                TxnResult::Ok,
                1,
                1,
                1,
                None,
                TxnState::InTransaction,
            ),
            (
                "add offsets loading, then committed",
                vec![Code(14), Code(0)],
                vec![Code(0)],
                TxnResult::Ok,
                2,
                1,
                1,
                None,
                TxnState::InTransaction,
            ),
            (
                "add offsets moved, then committed",
                vec![Code(16), Code(0)],
                vec![Code(0)],
                TxnResult::Ok,
                2,
                1,
                2,
                None,
                TxnState::InTransaction,
            ),
            (
                "add offsets lost, then committed",
                vec![Silent, Code(0)],
                vec![Code(0)],
                TxnResult::Ok,
                2,
                1,
                2,
                None,
                TxnState::InTransaction,
            ),
            (
                "add offsets fenced",
                vec![Code(90)],
                vec![Code(0)],
                Fenced,
                1,
                0,
                0,
                None,
                TxnState::Fenced,
            ),
            (
                "add offsets abortable",
                vec![Code(120)],
                vec![Code(0)],
                Server(120),
                1,
                0,
                0,
                Some(120),
                TxnState::InTransaction,
            ),
            (
                "add offsets invalid state is fatal",
                vec![Code(48)],
                vec![Code(0)],
                TxnResult::Fatal(48),
                1,
                0,
                0,
                None,
                TxnState::FatalError,
            ),
            (
                "offset commit unknown topic, then committed",
                vec![Code(0)],
                vec![Code(3), Code(0)],
                TxnResult::Ok,
                1,
                2,
                1,
                None,
                TxnState::InTransaction,
            ),
            (
                "offset commit timed out, then committed",
                vec![Code(0)],
                vec![Code(7), Code(0)],
                TxnResult::Ok,
                1,
                2,
                2,
                None,
                TxnState::InTransaction,
            ),
            (
                "offset commit moved, then committed",
                vec![Code(0)],
                vec![Code(16), Code(0)],
                TxnResult::Ok,
                1,
                2,
                2,
                None,
                TxnState::InTransaction,
            ),
            (
                "offset commit group metadata mismatch",
                vec![Code(0)],
                vec![Code(22)],
                Server(22),
                1,
                1,
                1,
                Some(22),
                TxnState::InTransaction,
            ),
            (
                "offset commit fenced",
                vec![Code(0)],
                vec![Code(47)],
                Fenced,
                1,
                1,
                1,
                None,
                TxnState::Fenced,
            ),
        ];
        for (
            name,
            add_offsets,
            txn_offset_commit,
            result,
            add_offsets_requests,
            txn_offset_commit_requests,
            coordinator_lookups,
            abortable_error,
            expected_state,
        ) in cases
        {
            let (mock, producer, coordinator) = scripted_producer(Coordinator {
                add_offsets: Script::new(&add_offsets, Code(0)),
                txn_offset_commit: Script::new(&txn_offset_commit, Code(0)),
                ..Coordinator::default()
            })
            .await;
            let transaction = producer
                .begin_transaction()
                .await
                .expect("begin transaction");
            let outcome = producer
                .send_offsets_to_transaction([(("topic".to_owned(), 0), 42)], &group)
                .await;
            let state = *producer.txn_state.lock().await;
            drop(transaction);
            let actual = {
                let coordinator = coordinator.lock().expect("scripted coordinator");
                ScriptedSendOffsets {
                    result: outcome.into(),
                    add_offsets_requests: coordinator.add_offsets.requests,
                    txn_offset_commit_requests: coordinator.txn_offset_commit.requests,
                    coordinator_lookups: coordinator.find_coordinator_requests,
                    abortable_error: producer.txn_error.abortable().map(|error| match error {
                        AbortableError::Server(code) => code,
                        AbortableError::Timeout => i16::MIN,
                    }),
                    state,
                }
            };
            let expected = ScriptedSendOffsets {
                result,
                add_offsets_requests,
                txn_offset_commit_requests,
                coordinator_lookups,
                abortable_error,
                state: expected_state,
            };
            assert2::assert!(actual == expected, "{name}");
            mock.stop();
        }
    }

    /// The number of `FindCoordinator` requests, where `MANY_LOOKUPS` stands
    /// for that number or more.
    fn lookups(coordinator: &Coordinator) -> usize {
        coordinator.find_coordinator_requests.min(MANY_LOOKUPS)
    }

    /// More lookups than a script names: the producer retried until the
    /// deadline.
    const MANY_LOOKUPS: usize = 5;

    /// The observable result of one scripted `init_transactions`.
    #[derive(Debug, PartialEq, Eq)]
    struct ScriptedInit {
        result: TxnResult,
        coordinator_lookups: usize,
        init_producer_id_requests: usize,
        state: TxnState,
    }

    /// Kafka's `FindCoordinatorHandler` sends the lookup again after a
    /// disconnect and after every retriable code, so `initTransactions`
    /// blocks while a new broker creates `__transaction_state`. The handler
    /// makes `TRANSACTIONAL_ID_AUTHORIZATION_FAILED` and an unexpected code
    /// fatal, and `TRANSACTION_ABORTABLE` abortable. A lookup that answers a
    /// retriable code until the deadline reports that code.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn init_transactions_retries_the_lookups_that_kafka_retries() {
        use Reply::{Code, Silent};
        use TxnResult::{Fatal, Server};
        let cases = [
            ("found", vec![Code(0)], TxnResult::Ok, 1, 1, TxnState::Ready),
            (
                "unavailable twice, then found",
                vec![Code(15), Code(15), Code(0)],
                TxnResult::Ok,
                3,
                1,
                TxnState::Ready,
            ),
            (
                "loading, moved, then found",
                vec![Code(14), Code(16), Code(0)],
                TxnResult::Ok,
                3,
                1,
                TxnState::Ready,
            ),
            (
                "lost, then found",
                vec![Silent, Code(0)],
                TxnResult::Ok,
                2,
                1,
                TxnState::Ready,
            ),
            (
                "unavailable until the deadline",
                vec![Code(15); MANY_LOOKUPS],
                Server(15),
                MANY_LOOKUPS,
                0,
                TxnState::Uninitialized,
            ),
            (
                "transactional id authorization is fatal",
                vec![Code(53)],
                Fatal(53),
                1,
                0,
                TxnState::FatalError,
            ),
            (
                "an unexpected code is fatal",
                vec![Code(42)],
                Fatal(42),
                1,
                0,
                TxnState::FatalError,
            ),
            (
                "transaction abortable leaves the producer as it was",
                vec![Code(120)],
                Server(120),
                1,
                0,
                TxnState::Uninitialized,
            ),
        ];
        for (name, txn_lookup, result, coordinator_lookups, init_producer_id_requests, state) in
            cases
        {
            let exhausted = txn_lookup.last().copied().unwrap_or(Code(0));
            let (mock, producer, coordinator) = scripted_producer_raw(Coordinator {
                txn_lookup: Script::new(&txn_lookup, exhausted),
                ..Coordinator::default()
            })
            .await;
            let outcome = producer.init_transactions().await;
            let state_after = *producer.txn_state.lock().await;
            let actual = {
                let coordinator = coordinator.lock().expect("scripted coordinator");
                ScriptedInit {
                    result: outcome.into(),
                    coordinator_lookups: lookups(&coordinator),
                    init_producer_id_requests: coordinator.init_producer_id_requests,
                    state: state_after,
                }
            };
            let expected = ScriptedInit {
                result,
                coordinator_lookups,
                init_producer_id_requests,
                state,
            };
            assert2::assert!(actual == expected, "{name}");
            mock.stop();
        }
    }

    /// The lookups of `send_offsets_to_transaction` follow Kafka's
    /// `FindCoordinatorHandler` too: the group lookup of `TxnOffsetCommit`, and
    /// the lookup of the transaction coordinator after `AddOffsetsToTxn`
    /// answers `NOT_COORDINATOR`. `GROUP_AUTHORIZATION_FAILED` stops the
    /// transaction from committing, and
    /// `TRANSACTIONAL_ID_AUTHORIZATION_FAILED` is fatal.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn send_offsets_to_transaction_retries_the_lookups_that_kafka_retries() {
        use Reply::Code;
        use TxnResult::{Fatal, Server};
        let group = krabka_client_consumer::ConsumerGroupMetadata {
            group_id: "group-a".into(),
            generation_id: 3,
            member_id: "member-a".into(),
            group_instance_id: None,
        };
        // (name, transactional id lookups after the one of init_transactions,
        // group lookups, add offsets script, result, offset commit requests,
        // coordinator lookups, abortable, transaction state)
        let cases = [
            (
                "group loading, then found",
                vec![],
                vec![Code(14), Code(0)],
                vec![Code(0)],
                TxnResult::Ok,
                1,
                2,
                None,
                TxnState::InTransaction,
            ),
            (
                "group unavailable twice, then found",
                vec![],
                vec![Code(15), Code(15), Code(0)],
                vec![Code(0)],
                TxnResult::Ok,
                1,
                3,
                None,
                TxnState::InTransaction,
            ),
            (
                "group authorization is abortable",
                vec![],
                vec![Code(30)],
                vec![Code(0)],
                Server(30),
                0,
                1,
                Some(30),
                TxnState::InTransaction,
            ),
            (
                "group lookup unexpected code is fatal",
                vec![],
                vec![Code(42)],
                vec![Code(0)],
                Fatal(42),
                0,
                1,
                None,
                TxnState::FatalError,
            ),
            (
                "transaction coordinator moved, loading, then found",
                vec![Code(14), Code(0)],
                vec![Code(0)],
                vec![Code(16), Code(0)],
                TxnResult::Ok,
                1,
                3,
                None,
                TxnState::InTransaction,
            ),
            (
                "transaction coordinator moved, then transactional id authorization",
                vec![Code(53)],
                vec![Code(0)],
                vec![Code(16)],
                Fatal(53),
                0,
                1,
                None,
                TxnState::FatalError,
            ),
            (
                "transaction coordinator moved, then transaction abortable",
                vec![Code(120)],
                vec![Code(0)],
                vec![Code(16)],
                Server(120),
                0,
                1,
                Some(120),
                TxnState::InTransaction,
            ),
        ];
        for (
            name,
            txn_lookup,
            group_lookup,
            add_offsets,
            result,
            txn_offset_commit_requests,
            coordinator_lookups,
            abortable_error,
            state,
        ) in cases
        {
            let mut txn_lookups = vec![Code(0)];
            txn_lookups.extend(txn_lookup);
            let (mock, producer, coordinator) = scripted_producer(Coordinator {
                txn_lookup: Script::new(&txn_lookups, Code(0)),
                group_lookup: Script::new(&group_lookup, Code(0)),
                add_offsets: Script::new(&add_offsets, Code(0)),
                ..Coordinator::default()
            })
            .await;
            let transaction = producer
                .begin_transaction()
                .await
                .expect("begin transaction");
            let outcome = producer
                .send_offsets_to_transaction([(("topic".to_owned(), 0), 42)], &group)
                .await;
            let state_after = *producer.txn_state.lock().await;
            drop(transaction);
            let actual = {
                let coordinator = coordinator.lock().expect("scripted coordinator");
                ScriptedSendOffsets {
                    result: outcome.into(),
                    add_offsets_requests: coordinator.add_offsets.requests,
                    txn_offset_commit_requests: coordinator.txn_offset_commit.requests,
                    coordinator_lookups: lookups(&coordinator),
                    abortable_error: producer.txn_error.abortable().map(|error| match error {
                        AbortableError::Server(code) => code,
                        AbortableError::Timeout => i16::MIN,
                    }),
                    state: state_after,
                }
            };
            let expected = ScriptedSendOffsets {
                result,
                add_offsets_requests: add_offsets.len(),
                txn_offset_commit_requests,
                coordinator_lookups,
                abortable_error,
                state,
            };
            assert2::assert!(actual == expected, "{name}");
            mock.stop();
        }
    }

    /// A commit whose `EndTxn` answers `NOT_COORDINATOR` finds the
    /// coordinator again, with the retries of Kafka's
    /// `FindCoordinatorHandler`. A fatal lookup answer makes the producer
    /// fatal, and an abortable one fails the commit with that code.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn commit_retries_the_lookups_that_kafka_retries() {
        use Reply::Code;
        use TxnResult::{Fatal, Server};
        let cases = [
            (
                "moved, unavailable, then committed",
                vec![Code(15), Code(0)],
                vec![Code(16), Code(0)],
                TxnResult::Ok,
                TxnState::Ready,
            ),
            (
                "moved, then transactional id authorization",
                vec![Code(53)],
                vec![Code(16)],
                Fatal(53),
                TxnState::FatalError,
            ),
            (
                "moved, then transaction abortable",
                vec![Code(120)],
                vec![Code(16)],
                Server(120),
                TxnState::InTransaction,
            ),
        ];
        for (name, txn_lookup, end_txn, result, state) in cases {
            let mut txn_lookups = vec![Code(0)];
            txn_lookups.extend(txn_lookup);
            let (mock, producer, coordinator) = scripted_producer(Coordinator {
                txn_lookup: Script::new(&txn_lookups, Code(0)),
                end_txn: Script::new(&end_txn, Code(0)),
                ..Coordinator::default()
            })
            .await;
            let outcome = producer
                .begin_transaction()
                .await
                .expect("begin transaction")
                .commit()
                .await
                .map_err(|error| error.source);
            let state_after = *producer.txn_state.lock().await;
            let actual = ScriptedCommit {
                result: outcome.into(),
                end_txn_requests: coordinator
                    .lock()
                    .expect("scripted coordinator")
                    .end_txn
                    .requests,
                state: state_after,
            };
            let expected = ScriptedCommit {
                result,
                end_txn_requests: end_txn.len(),
                state,
            };
            assert2::assert!(actual == expected, "{name}");
            mock.stop();
        }
    }

    /// Kafka's producer follows the finalized feature `transaction.version`
    /// (`TransactionManager.maybeUpdateTransactionV2Enabled`). With level 2 it
    /// sends no `AddPartitionsToTxn` (`maybeAddPartition`) and no
    /// `AddOffsetsToTxn` (`sendOffsetsToTransaction`), and it sends Produce,
    /// `TxnOffsetCommit` and `EndTxn` at their latest versions. With level 1
    /// it caps them at `LAST_STABLE_VERSION_BEFORE_TRANSACTION_V2`: 11, 4 and
    /// 4.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn the_transaction_version_decides_the_transaction_requests() {
        let group = krabka_client_consumer::ConsumerGroupMetadata {
            group_id: "group-a".into(),
            generation_id: 3,
            member_id: "member-a".into(),
            group_instance_id: None,
        };
        let cases = [
            (
                "transaction version 1",
                1,
                vec![
                    (produce_request::API_KEY, 11),
                    (txn_offset_commit_request::API_KEY, 4),
                    (end_txn_request::API_KEY, 4),
                    (add_partitions_to_txn_request::API_KEY, 3),
                    (add_offsets_to_txn_request::API_KEY, 4),
                ],
            ),
            (
                "transaction version 2",
                2,
                // The mock metadata has no topic id, so Produce stops at v12,
                // the last version with topic names.
                vec![
                    (produce_request::API_KEY, 12),
                    (txn_offset_commit_request::API_KEY, 5),
                    (end_txn_request::API_KEY, 5),
                ],
            ),
        ];
        for (name, transaction_version, expected) in cases {
            let (mock, producer, coordinator) = scripted_producer(Coordinator {
                transaction_version: Some(transaction_version),
                ..Coordinator::default()
            })
            .await;
            coordinator
                .lock()
                .expect("scripted coordinator")
                .transaction_requests
                .clear();
            let transaction = producer
                .begin_transaction()
                .await
                .expect("begin transaction");
            let record = producer
                .enqueue(ProducerRecord {
                    topic: "topic".to_owned(),
                    partition: Some(0),
                    value: Some(bytes::Bytes::from_static(b"v")),
                    ..Default::default()
                })
                .await
                .expect("record is queued");
            producer
                .send_offsets_to_transaction([(("topic".to_owned(), 0), 42)], &group)
                .await
                .expect("send offsets");
            transaction.commit().await.expect("commit");
            record.await.expect("the record is delivered");
            let mut actual = coordinator
                .lock()
                .expect("scripted coordinator")
                .transaction_requests
                .clone();
            mock.stop();
            let mut expected = expected;
            actual.sort_unstable();
            expected.sort_unstable();
            assert2::assert!(actual == expected, "{name}");
        }
    }

    /// What one scripted `send_offsets_to_transaction` sent and gave.
    #[derive(Debug, PartialEq, Eq)]
    struct SentOffsetCommits {
        result: TxnResult,
        /// The negotiated version and the decoded topics of each
        /// `TxnOffsetCommit` request.
        requests: Vec<(i16, Vec<TxnOffsetCommitRequestTopic>)>,
    }

    /// Kafka's `TransactionManager.txnOffsetCommitHandler` reads the topic
    /// ids from the metadata that `KafkaProducer.sendOffsetsToTransaction`
    /// waited for. With an id for every topic it negotiates up to v6
    /// (KIP-1319), which names each topic by id. Without an id, or with
    /// transaction version 1, it names each topic at v5 or v4. The handler
    /// treats `UNKNOWN_TOPIC_ID` as retriable and sends the same request
    /// again.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn send_offsets_to_transaction_names_topics_by_id_at_v6() {
        use Reply::Code;
        let group = krabka_client_consumer::ConsumerGroupMetadata {
            group_id: "group-a".into(),
            generation_id: 3,
            member_id: "member-a".into(),
            group_instance_id: None,
        };
        let id = Uuid([5; 16]);
        let by_id = TxnOffsetCommitRequestTopic {
            topic_id: id,
            partitions: vec![TxnOffsetCommitRequestPartition {
                partition_index: 0,
                committed_offset: 42,
                ..Default::default()
            }],
            ..Default::default()
        };
        let by_name = TxnOffsetCommitRequestTopic {
            name: "topic".into(),
            topic_id: Uuid::ZERO,
            ..by_id.clone()
        };
        // (name, transaction version, advertised max, metadata topic id,
        // offset commit script, expected)
        let cases = [
            (
                "every topic has an id",
                2,
                6,
                id,
                vec![Code(0)],
                SentOffsetCommits {
                    result: TxnResult::Ok,
                    requests: vec![(6, vec![by_id.clone()])],
                },
            ),
            (
                "the metadata gives no id",
                2,
                6,
                Uuid::ZERO,
                vec![Code(0)],
                SentOffsetCommits {
                    result: TxnResult::Ok,
                    requests: vec![(5, vec![by_name.clone()])],
                },
            ),
            (
                "the broker stops at v5",
                2,
                5,
                id,
                vec![Code(0)],
                SentOffsetCommits {
                    result: TxnResult::Ok,
                    requests: vec![(5, vec![by_name.clone()])],
                },
            ),
            (
                "transaction version 1",
                1,
                6,
                id,
                vec![Code(0)],
                SentOffsetCommits {
                    result: TxnResult::Ok,
                    requests: vec![(4, vec![by_name.clone()])],
                },
            ),
            (
                "unknown topic id, then committed",
                2,
                6,
                id,
                vec![Code(100), Code(0)],
                SentOffsetCommits {
                    result: TxnResult::Ok,
                    requests: vec![(6, vec![by_id.clone()]), (6, vec![by_id.clone()])],
                },
            ),
            (
                "unknown topic id until the retry deadline",
                2,
                6,
                id,
                vec![Code(100)],
                SentOffsetCommits {
                    result: TxnResult::Server(100),
                    requests: Vec::new(),
                },
            ),
        ];
        for (name, transaction_version, txn_offset_commit_max, topic_id, script, expected) in cases
        {
            let exhausted = *script.last().expect("script");
            let (mock, producer, coordinator) = scripted_producer(Coordinator {
                transaction_version: Some(transaction_version),
                txn_offset_commit_max: Some(txn_offset_commit_max),
                topic_id,
                txn_offset_commit: Script::new(&script, exhausted),
                ..Coordinator::default()
            })
            .await;
            let transaction = producer
                .begin_transaction()
                .await
                .expect("begin transaction");
            let result = producer
                .send_offsets_to_transaction([(("topic".to_owned(), 0), 42)], &group)
                .await
                .into();
            drop(transaction);
            let mut requests = coordinator
                .lock()
                .expect("scripted coordinator")
                .txn_offset_commit_topics
                .clone();
            mock.stop();
            // The request goes out until the retry deadline ends, so the row
            // checks the shape of each request and not their number.
            if expected.requests.is_empty() {
                requests.dedup();
                assert2::assert!(requests == vec![(6, vec![by_id.clone()])], "{name}");
                requests.clear();
            }
            let actual = SentOffsetCommits { result, requests };
            assert2::assert!(actual == expected, "{name}");
        }
    }

    /// The observable result of a commit after a failed transactional batch.
    #[derive(Debug, PartialEq, Eq)]
    struct CommitAfterFailedBatch {
        record: TxnResult,
        commit: TxnResult,
        end_txn_requests: usize,
        abort: TxnResult,
        abortable_error_after_abort: Option<i16>,
        next_transaction: TxnResult,
    }

    /// A send after `prepare_transaction` fails at once with the state error.
    /// It does not wait for buffer memory first.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn send_in_preparing_state_fails_before_waiting_for_memory() {
        let (mock, producer, _coordinator) = scripted_producer(Coordinator::default()).await;
        let _transaction = producer
            .begin_transaction()
            .await
            .expect("begin transaction");
        *producer.txn_state.lock().await = TxnState::Preparing;
        let _all_memory = producer
            .buffer_pool
            .allocate(producer.buffer_pool.total(), Duration::ZERO)
            .await
            .expect("the pool is free");

        let send = producer.send(ProducerRecord {
            topic: "topic".to_owned(),
            partition: Some(0),
            value: Some(bytes::Bytes::from_static(b"v")),
            ..Default::default()
        });
        let outcome = time::timeout(Duration::from_secs(2), send)
            .await
            .map_or_else(
                |_| Err("waited".to_owned()),
                |result| result.map(drop).map_err(|error| error.to_string()),
            );
        mock.stop();
        assert2::assert!(
            outcome
                == Err(
                    "invalid transaction state: send is not allowed after prepare_transaction"
                        .to_owned()
                )
        );
    }

    /// What a send in a prepared transaction gave, and the `Metadata` requests
    /// that it sent.
    #[derive(Debug, PartialEq, Eq)]
    struct PreparedSend {
        outcome: Result<(), String>,
        metadata_requests: usize,
    }

    /// A send after `prepare_transaction` fails at once with the state error.
    /// It does not wait for the metadata of a topic that the producer does not
    /// know, as Kafka's `KafkaProducer.doSend` calls `throwIfInPreparedState`
    /// before `waitOnMetadata`.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn send_in_a_prepared_transaction_fails_before_the_metadata_wait() {
        for state in [TxnState::Preparing, TxnState::Prepared] {
            let (mock, producer, coordinator) = scripted_producer(Coordinator::default()).await;
            let _transaction = producer
                .begin_transaction()
                .await
                .expect("begin transaction");
            *producer.txn_state.lock().await = state;
            let before = coordinator
                .lock()
                .expect("scripted coordinator")
                .metadata_requests;

            // The mock broker never lists this topic.
            let send = producer.send(ProducerRecord {
                topic: "unknown".to_owned(),
                value: Some(bytes::Bytes::from_static(b"v")),
                ..Default::default()
            });
            let outcome = time::timeout(Duration::from_secs(2), send)
                .await
                .map_or_else(
                    |_| Err("waited".to_owned()),
                    |result| result.map(drop).map_err(|error| error.to_string()),
                );
            let actual = PreparedSend {
                outcome,
                metadata_requests: coordinator
                    .lock()
                    .expect("scripted coordinator")
                    .metadata_requests
                    - before,
            };
            mock.stop();
            assert2::assert!(
                actual
                    == PreparedSend {
                        outcome: Err("invalid transaction state: send is not allowed after \
                                      prepare_transaction"
                            .to_owned()),
                        metadata_requests: 0,
                    },
                "{state:?}"
            );
        }
    }

    /// Concurrent transactional sends to one partition all complete. A send
    /// takes the transaction state lock before the accumulator lock, also
    /// when it checks for buffer memory, so two sends cannot wait for each
    /// other.
    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
    async fn concurrent_transactional_sends_to_one_partition_complete() {
        let (mock, producer, _coordinator) = scripted_producer(Coordinator::default()).await;
        let _transaction = producer
            .begin_transaction()
            .await
            .expect("begin transaction");
        let sends = iter::repeat_with(|| {
            producer.enqueue(ProducerRecord {
                topic: "topic".to_owned(),
                partition: Some(0),
                value: Some(bytes::Bytes::from_static(b"v")),
                ..Default::default()
            })
        })
        .take(200);
        let enqueued = time::timeout(Duration::from_secs(10), join_all(sends))
            .await
            .map(|receivers| receivers.len());
        mock.stop();
        assert2::assert!(enqueued == Ok(200));
    }

    /// Kafka's `Sender.failBatch` calls
    /// `TransactionManager.handleFailedBatch`, which moves a transactional
    /// producer to `ABORTABLE_ERROR`. `commitTransaction` then fails in
    /// `maybeFailWithError`, and it sends no `EndTxn`. Only
    /// `abortTransaction` clears the state.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_failed_transactional_batch_stops_a_later_commit() {
        let (mock, producer, coordinator) = scripted_producer(Coordinator {
            // 42 is INVALID_REQUEST, which Kafka does not retry.
            produce: Script::new(&[Reply::Code(42)], Reply::Code(0)),
            ..Coordinator::default()
        })
        .await;
        let (record, commit) = {
            let transaction = producer
                .begin_transaction()
                .await
                .expect("begin transaction");
            let record = producer
                .send(ProducerRecord {
                    topic: "topic".to_owned(),
                    partition: Some(0),
                    value: Some(bytes::Bytes::from_static(b"v")),
                    ..Default::default()
                })
                .await;

            (record, transaction.commit().await)
        };
        let (commit_result, transaction) = match commit {
            Ok(()) => (TxnResult::Ok, None),
            Err(error) => (TxnResult::from(Err(error.source)), Some(error.transaction)),
        };
        let end_txn_requests = coordinator
            .lock()
            .expect("scripted coordinator")
            .end_txn
            .requests;
        let abort = match transaction {
            Some(transaction) => {
                TxnResult::from(transaction.abort().await.map_err(|error| error.source))
            }
            None => TxnResult::Other("the commit succeeded".to_owned()),
        };
        let abortable_error_after_abort = producer.txn_error.abortable().map(|error| match error {
            AbortableError::Server(code) => code,
            AbortableError::Timeout => i16::MIN,
        });
        let next_transaction = match producer.begin_transaction().await {
            Ok(transaction) => {
                TxnResult::from(transaction.abort().await.map_err(|error| error.source))
            }
            Err(error) => TxnResult::from(Err(error)),
        };

        let actual = CommitAfterFailedBatch {
            record: TxnResult::from(record.map(drop)),
            commit: commit_result,
            end_txn_requests,
            abort,
            abortable_error_after_abort,
            next_transaction,
        };
        assert2::assert!(
            actual
                == CommitAfterFailedBatch {
                    record: TxnResult::Server(42),
                    commit: TxnResult::Server(42),
                    end_txn_requests: 0,
                    abort: TxnResult::Ok,
                    abortable_error_after_abort: None,
                    next_transaction: TxnResult::Ok,
                }
        );
        mock.stop();
    }

    /// The producer state after a commit whose `EndTxn` v5 answer carried an
    /// identity.
    #[derive(Debug, PartialEq, Eq)]
    struct IdentityAfterCommit {
        identity: (i64, i16),
        sequences: Vec<((String, i32), i32)>,
    }

    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn a_new_producer_identity_resets_sequence_numbers() {
        let kept = vec![(("topic".to_owned(), 0), 5)];
        let cases = [
            ("epoch bump", (7, 4), (7, 4), vec![]),
            ("new producer id", (8, 0), (8, 0), vec![]),
            ("same identity", (7, 3), (7, 3), kept.clone()),
            ("no identity in the answer", (-1, -1), (7, 3), kept),
        ];
        for (name, answered, identity, sequences) in cases {
            let port_cell = Arc::new(AtomicU16::new(0));
            let handler_port = Arc::clone(&port_cell);
            let mock = MockBroker::start(move |api_key, version, _corr_id, _body| {
                if api_key == api_versions_request::API_KEY {
                    // Only transaction version 2 sends `EndTxn` v5, whose
                    // answer carries the new identity.
                    return Some(scripted_api_versions((0, 5), false, Some(2), 5, version));
                }
                if api_key == find_coordinator_request::API_KEY {
                    return Some(encode_v0(&FindCoordinatorResponse {
                        error_code: 0,
                        node_id: 1,
                        host: "127.0.0.1".into(),
                        port: i32::from(handler_port.load(Ordering::SeqCst)),
                        ..Default::default()
                    }));
                }
                if api_key == init_producer_id_request::API_KEY {
                    return Some(encode_v0(&InitProducerIdResponse {
                        error_code: 0,
                        producer_id: 7,
                        producer_epoch: 3,
                        ..Default::default()
                    }));
                }
                if api_key == end_txn_request::API_KEY {
                    let mut buf = BytesMut::new();
                    if version >= end_txn_response::FLEXIBLE_MIN {
                        buf.extend_from_slice(&[0]);
                    }
                    EndTxnResponse {
                        error_code: 0,
                        producer_id: answered.0,
                        producer_epoch: answered.1,
                        ..Default::default()
                    }
                    .encode(&mut buf, version)
                    .expect("encode EndTxn response");
                    return Some(buf.to_vec());
                }
                None
            })
            .await
            .unwrap();
            port_cell.store(mock.addr.port(), Ordering::SeqCst);
            let producer = Producer::builder()
                .bootstrap(mock.addr.to_string())
                .transactional_id("test-txn")
                .request_timeout(Duration::from_millis(100))
                .build()
                .await
                .expect("producer connects to the mock");
            producer
                .init_transactions()
                .await
                .expect("init_transactions against the mock coordinator");
            let transaction = producer
                .begin_transaction()
                .await
                .expect("begin transaction");
            producer.next_seq.insert(("topic".to_owned(), 0), 5);
            transaction.commit().await.expect("commit");
            let actual = IdentityAfterCommit {
                identity: *producer.txn_pid_epoch.lock().await,
                sequences: producer
                    .next_seq
                    .iter()
                    .map(|entry| (entry.key().clone(), *entry.value()))
                    .collect(),
            };
            drop(producer);
            let expected = IdentityAfterCommit {
                identity,
                sequences,
            };
            assert2::assert!(actual == expected, "{name}");
            mock.stop();
        }
    }

    /// What `prepare_transaction` gave after a batch failed during its flush.
    #[derive(Debug, PartialEq, Eq)]
    struct PrepareAfterFailedFlush {
        record: TxnResult,
        prepare: TxnResult,
        state: TxnState,
    }

    /// Kafka's `KafkaProducer.prepareTransaction` flushes, and then
    /// `TransactionManager.prepareTransaction` calls `maybeFailWithError`. A
    /// batch that fails during the flush therefore stops the prepare.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn prepare_fails_when_a_batch_fails_during_its_flush() {
        let (mock, producer, _coordinator) = scripted_producer(Coordinator {
            // 42 is INVALID_REQUEST, which Kafka does not retry.
            produce: Script::new(&[Reply::Code(42)], Reply::Code(0)),
            two_phase_commit: true,
            // The record stays in the accumulator until the prepare flushes.
            linger: Some(Duration::from_mins(1)),
            ..Coordinator::default()
        })
        .await;
        let transaction = producer
            .begin_transaction()
            .await
            .expect("begin transaction");
        let receiver = producer
            .enqueue(ProducerRecord {
                topic: "topic".to_owned(),
                partition: Some(0),
                value: Some(bytes::Bytes::from_static(b"v")),
                ..Default::default()
            })
            .await
            .expect("record is queued");
        let prepare = TxnResult::from(transaction.prepare().await.map(drop));
        let record = TxnResult::from(receiver.await.map(drop));
        let actual = PrepareAfterFailedFlush {
            record,
            prepare,
            state: *producer.txn_state.lock().await,
        };
        drop(transaction);
        mock.stop();
        assert2::assert!(
            actual
                == PrepareAfterFailedFlush {
                    record: TxnResult::Server(42),
                    prepare: TxnResult::Server(42),
                    state: TxnState::InTransaction,
                }
        );
    }

    /// The requests that a scripted coordinator got.
    #[derive(Debug, Default, PartialEq, Eq)]
    struct CoordinatorRequests {
        add_partitions: usize,
        produce: usize,
        add_offsets: usize,
        txn_offset_commit: usize,
        end_txn: usize,
        init_producer_id: usize,
    }

    impl CoordinatorRequests {
        fn of(coordinator: &SharedCoordinator) -> Self {
            let coordinator = coordinator.lock().expect("scripted coordinator");
            Self {
                add_partitions: coordinator.add_partitions.requests,
                produce: coordinator.produce.requests,
                add_offsets: coordinator.add_offsets.requests,
                txn_offset_commit: coordinator.txn_offset_commit.requests,
                end_txn: coordinator.end_txn.requests,
                init_producer_id: coordinator.init_producer_id_requests,
            }
        }
    }

    /// The operation that puts the transaction in an error state.
    #[derive(Debug, Clone, Copy)]
    enum Trigger {
        Send,
        SendOffsets,
        Commit,
    }

    /// What each transactional operation gave after a transaction error, and
    /// the requests that all of them sent.
    #[derive(Debug, PartialEq, Eq)]
    struct AfterTransactionError {
        trigger: TxnResult,
        state: TxnState,
        send: TxnResult,
        send_offsets: TxnResult,
        commit: TxnResult,
        abort: TxnResult,
        begin_and_abort: TxnResult,
        init_transactions: TxnResult,
        requests: CoordinatorRequests,
    }

    /// Kafka's `TransactionManager.maybeFailWithError` throws the stored error
    /// in `maybeAddPartition` (so `KafkaProducer.doSend` fails the send),
    /// `sendOffsetsToTransaction`, `beginCommit`, `beginTransaction` and
    /// `initializeTransactions`. An abortable error lets `beginAbort` go on,
    /// and the abort clears it. A fatal error (`fatalError` in the handlers:
    /// a fence, `TRANSACTIONAL_ID_AUTHORIZATION_FAILED`, `INVALID_TXN_STATE`,
    /// `INVALID_PRODUCER_ID_MAPPING` and an unexpected code) stops every one
    /// of them, the abort and a new `initTransactions` included.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn every_operation_fails_after_a_transaction_error() {
        use Reply::Code;
        use TxnResult::{Fatal, Fenced, Server};
        let group = krabka_client_consumer::ConsumerGroupMetadata {
            group_id: "group-a".into(),
            generation_id: 3,
            member_id: "member-a".into(),
            group_instance_id: None,
        };
        let fatal = |code, state, requests| AfterTransactionError {
            trigger: Fatal(code),
            state,
            send: Fatal(code),
            send_offsets: Fatal(code),
            commit: Fatal(code),
            abort: Fatal(code),
            begin_and_abort: Fatal(code),
            init_transactions: Fatal(code),
            requests,
        };
        let cases = [
            (
                "add partitions transactional id authorization",
                Trigger::Send,
                Coordinator {
                    add_partitions: Script::new(&[Code(53)], Code(0)),
                    ..Coordinator::default()
                },
                fatal(
                    53,
                    TxnState::FatalError,
                    CoordinatorRequests {
                        add_partitions: 1,
                        ..CoordinatorRequests::default()
                    },
                ),
            ),
            (
                "add partitions producer fenced",
                Trigger::Send,
                Coordinator {
                    add_partitions: Script::new(&[Code(90)], Code(0)),
                    ..Coordinator::default()
                },
                AfterTransactionError {
                    trigger: Fenced,
                    state: TxnState::Fenced,
                    send: Fenced,
                    send_offsets: Fenced,
                    commit: Fenced,
                    abort: Fenced,
                    begin_and_abort: Fenced,
                    init_transactions: Fenced,
                    requests: CoordinatorRequests {
                        add_partitions: 1,
                        ..CoordinatorRequests::default()
                    },
                },
            ),
            (
                "add offsets invalid producer id mapping",
                Trigger::SendOffsets,
                Coordinator {
                    add_offsets: Script::new(&[Code(49)], Code(0)),
                    ..Coordinator::default()
                },
                fatal(
                    49,
                    TxnState::FatalError,
                    CoordinatorRequests {
                        add_offsets: 1,
                        ..CoordinatorRequests::default()
                    },
                ),
            ),
            (
                "end txn invalid txn state",
                Trigger::Commit,
                Coordinator {
                    end_txn: Script::new(&[Code(48)], Code(0)),
                    ..Coordinator::default()
                },
                fatal(
                    48,
                    TxnState::FatalError,
                    CoordinatorRequests {
                        end_txn: 1,
                        ..CoordinatorRequests::default()
                    },
                ),
            ),
            (
                "failed batch is abortable",
                Trigger::Send,
                Coordinator {
                    // 42 is INVALID_REQUEST, which Kafka does not retry.
                    produce: Script::new(&[Code(42)], Code(0)),
                    ..Coordinator::default()
                },
                AfterTransactionError {
                    trigger: Server(42),
                    state: TxnState::InTransaction,
                    send: Server(42),
                    send_offsets: Server(42),
                    commit: Server(42),
                    abort: TxnResult::Ok,
                    begin_and_abort: TxnResult::Ok,
                    init_transactions: TxnResult::Ok,
                    requests: CoordinatorRequests {
                        add_partitions: 1,
                        produce: 1,
                        end_txn: 2,
                        init_producer_id: 1,
                        ..CoordinatorRequests::default()
                    },
                },
            ),
        ];
        for (name, trigger, coordinator, expected) in cases {
            let (mock, producer, coordinator) = scripted_producer(coordinator).await;
            let record = || ProducerRecord {
                topic: "topic".to_owned(),
                partition: Some(0),
                value: Some(bytes::Bytes::from_static(b"v")),
                ..Default::default()
            };
            let transaction = producer
                .begin_transaction()
                .await
                .expect("begin transaction");
            let (trigger_result, transaction) = match trigger {
                Trigger::Send => {
                    let result = producer.send(record()).await.map(drop);
                    (TxnResult::from(result), transaction)
                }
                Trigger::SendOffsets => {
                    let result = producer
                        .send_offsets_to_transaction([(("topic".to_owned(), 0), 42)], &group)
                        .await;
                    (TxnResult::from(result), transaction)
                }
                Trigger::Commit => match transaction.commit().await {
                    Ok(()) => panic!("{name}: the commit succeeded"),
                    Err(error) => (TxnResult::from(Err(error.source)), error.transaction),
                },
            };
            let state = *producer.txn_state.lock().await;
            let send = TxnResult::from(producer.send(record()).await.map(drop));
            let send_offsets = TxnResult::from(
                producer
                    .send_offsets_to_transaction([(("topic".to_owned(), 0), 43)], &group)
                    .await,
            );
            let (commit, transaction) = match transaction.commit().await {
                Ok(()) => (TxnResult::Ok, None),
                Err(error) => (TxnResult::from(Err(error.source)), Some(error.transaction)),
            };
            let abort = match transaction {
                Some(transaction) => {
                    TxnResult::from(transaction.abort().await.map_err(|error| error.source))
                }
                None => TxnResult::Other("the commit succeeded".to_owned()),
            };
            let begin_and_abort = match producer.begin_transaction().await {
                Ok(transaction) => {
                    TxnResult::from(transaction.abort().await.map_err(|error| error.source))
                }
                Err(error) => TxnResult::from(Err(error)),
            };
            let init_transactions = TxnResult::from(producer.init_transactions().await);
            let actual = AfterTransactionError {
                trigger: trigger_result,
                state,
                send,
                send_offsets,
                commit,
                abort,
                begin_and_abort,
                init_transactions,
                requests: CoordinatorRequests::of(&coordinator),
            };
            mock.stop();
            assert2::assert!(actual == expected, "{name}");
        }
    }

    /// How a coordinator call with a rejected authentication ended.
    #[derive(Debug, PartialEq, Eq)]
    struct RejectedCoordinator {
        rejected: bool,
        rejecting_handshakes: usize,
        state: TxnState,
    }

    /// Kafka's `Sender.runOnce` catches an `AuthenticationException` from the
    /// transaction coordinator and fails every pending transactional request
    /// (`TransactionManager.authenticationFailed`). The producer does not
    /// retry `AddPartitionsToTxn` or `EndTxn`, and it does not find another
    /// coordinator.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn coordinator_sasl_rejection_is_not_retried() {
        use krabka_client_core::{
            Client, MockReply, MockSaslAnswer,
            security::{ClientSecurity, SaslCredentials},
        };
        use krabka_security::ListenerProtocol;

        #[derive(Clone, Copy, Debug)]
        enum Call {
            AddPartitions,
            Commit,
        }

        for (call, state) in [
            (Call::AddPartitions, TxnState::InTransaction),
            (Call::Commit, TxnState::RecoveryRequired),
        ] {
            let (mock, producer, _coordinator) = scripted_producer(Coordinator::default()).await;
            let handshakes = Arc::new(AtomicUsize::new(0));
            let counter = Arc::clone(&handshakes);
            let answer = MockSaslAnswer::AuthenticateError(58);
            let rejecting =
                MockBroker::start_with_replies(move |api_key, version, _corr, _body| {
                    if api_key == sasl_handshake_request::API_KEY {
                        counter.fetch_add(1, Ordering::SeqCst);
                    }
                    answer
                        .reply(api_key, version)
                        .unwrap()
                        .unwrap_or(MockReply::Silent)
                })
                .await
                .unwrap();
            let rejecting_client = Client::builder()
                .bootstrap(rejecting.addr.to_string())
                .security(ClientSecurity {
                    protocol: ListenerProtocol::SaslPlaintext,
                    tls: None,
                    sasl: Some(SaslCredentials::Plain {
                        username: "alice".into(),
                        password: "secret".into(),
                    }),
                    sasl_host: None,
                })
                .build()
                .await
                .expect("client");
            let transaction = producer
                .begin_transaction()
                .await
                .expect("begin transaction");
            *producer.txn_coord_client.lock().await = Some(rejecting_client);

            let result = match call {
                Call::AddPartitions => producer.register_transaction_partition("topic", 0).await,
                Call::Commit => transaction.commit().await.map_err(|error| error.source),
            };
            let actual = RejectedCoordinator {
                rejected: matches!(
                    result,
                    Err(ProducerError::Client(ClientError::Authentication { .. }))
                ),
                rejecting_handshakes: handshakes.load(Ordering::SeqCst),
                state: *producer.txn_state.lock().await,
            };

            rejecting.stop();
            mock.stop();
            assert2::assert!(
                actual
                    == RejectedCoordinator {
                        rejected: true,
                        rejecting_handshakes: 1,
                        state,
                    },
                "{call:?}"
            );
        }
    }

    /// The transaction state each `send` / `send_offsets_to_transaction`
    /// case starts from.
    #[derive(Debug, Clone, Copy)]
    enum StartState {
        /// Before `init_transactions`.
        Uninitialized,
        /// After `init_transactions`, before `begin_transaction`.
        Ready,
        /// Between `begin_transaction` and `commit`/`abort`.
        InTransaction,
    }

    #[derive(Debug, Clone, Copy)]
    enum Operation {
        SendRecord,
        SendOffsets,
    }

    /// Kafka's `TransactionManager.maybeAddPartition` and
    /// `.sendOffsetsToTransaction` throw `IllegalStateException` synchronously,
    /// and send nothing, when `currentState != IN_TRANSACTION`. `send` and
    /// `send_offsets_to_transaction` must refuse the same way, sending nothing,
    /// outside `InTransaction`, and must succeed, sending requests, inside it.
    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
    async fn send_and_send_offsets_require_an_open_transaction() {
        use Operation::{SendOffsets, SendRecord};
        use StartState::{InTransaction, Ready, Uninitialized};

        let cases = [
            ("send, uninitialized", Uninitialized, SendRecord, false),
            ("send, ready", Ready, SendRecord, false),
            ("send, in transaction", InTransaction, SendRecord, true),
            (
                "send_offsets_to_transaction, uninitialized",
                Uninitialized,
                SendOffsets,
                false,
            ),
            (
                "send_offsets_to_transaction, ready",
                Ready,
                SendOffsets,
                false,
            ),
            (
                "send_offsets_to_transaction, in transaction",
                InTransaction,
                SendOffsets,
                true,
            ),
        ];

        for (name, state, operation, should_succeed) in cases {
            let (mock, producer, coordinator) = match state {
                Uninitialized => scripted_producer_raw(Coordinator::default()).await,
                Ready | InTransaction => scripted_producer(Coordinator::default()).await,
            };
            let transaction = match state {
                InTransaction => Some(
                    producer
                        .begin_transaction()
                        .await
                        .expect("begin_transaction"),
                ),
                Uninitialized | Ready => None,
            };
            assert2::assert!(
                matches!(
                    (*producer.txn_state.lock().await, state),
                    (TxnState::Uninitialized, Uninitialized)
                        | (TxnState::Ready, Ready)
                        | (TxnState::InTransaction, InTransaction)
                ),
                "{name}"
            );

            let group = krabka_client_consumer::ConsumerGroupMetadata {
                group_id: "group-a".into(),
                generation_id: 3,
                member_id: "member-a".into(),
                group_instance_id: None,
            };
            let result = match operation {
                SendRecord => producer
                    .send(ProducerRecord {
                        topic: "topic".into(),
                        value: Some(bytes::Bytes::from_static(b"v")),
                        ..Default::default()
                    })
                    .await
                    .map(|_| ()),
                SendOffsets => {
                    producer
                        .send_offsets_to_transaction([(("topic".to_owned(), 0), 5)], &group)
                        .await
                }
            };

            assert2::assert!(result.is_ok() == should_succeed, "{name}: {result:?}");
            if !should_succeed {
                assert2::assert!(
                    matches!(result, Err(ProducerError::InvalidTransactionState(_))),
                    "{name}: {result:?}"
                );
            }

            let requests_sent = coordinator
                .lock()
                .expect("scripted coordinator")
                .transaction_requests
                .len();
            if should_succeed {
                assert2::assert!(requests_sent > 0, "{name}: expected requests to be sent");
            } else {
                assert2::assert!(requests_sent == 0, "{name}: sent {requests_sent} requests");
            }

            drop(transaction);
            mock.stop();
        }
    }
}