kafkit_client/core/

error.rs

//! Error types returned by Kafkit.
//!
//! Most applications can use [`Result`] directly and match on [`crate::Error`] only
//! when they need to handle a specific category.
//!
//! ```no_run
//! # async fn example() -> kafkit_client::Result<()> {
//! use kafkit_client::{Error, KafkaClient};
//!
//! let admin = KafkaClient::new("localhost:9092").admin().connect().await?;
//! if let Err(Error::Admin(error)) = admin.create_topics(Vec::<kafkit_client::NewTopic>::new()).await {
//!     eprintln!("admin request failed: {error}");
//! }
//! # Ok(())
//! # }
//! ```
//!
use thiserror::Error;
use tokio::task::JoinError;

use kafka_protocol::error::ResponseError;

/// Result type used by the client APIs.
pub type Result<T> = std::result::Result<T, Error>;

/// Operational classification for a client error.
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct ErrorClassification {
    /// The same operation may succeed if retried after backoff or metadata refresh.
    pub retriable: bool,
    /// The client instance, producer id, or transaction is no longer usable.
    pub fatal: bool,
    /// A transactional producer must abort the current transaction before more work.
    pub transaction_abort_required: bool,
}

#[derive(Debug, Error)]
/// Top-level error returned by the crate.
pub enum Error {
    /// Operation was cancelled before it completed.
    #[error("operation cancelled")]
    Cancelled,
    /// Admin operation failed.
    #[error(transparent)]
    Admin(#[from] AdminError),
    /// Consumer operation failed.
    #[error(transparent)]
    Consumer(#[from] ConsumerError),
    /// Producer operation failed.
    #[error(transparent)]
    Producer(#[from] ProducerError),
    /// Consumer group metadata was invalid.
    #[error(transparent)]
    ConsumerGroupMetadata(#[from] ConsumerGroupMetadataError),
    /// Transaction state prevented the operation.
    #[error(transparent)]
    TransactionState(#[from] TransactionStateError),
    /// A Kafka broker returned a protocol error code.
    #[error(transparent)]
    Broker(#[from] BrokerError),
    /// Public input validation failed before a broker request was sent.
    #[error(transparent)]
    Validation(#[from] ValidationError),
    /// The client could not decode or validate a Kafka protocol response.
    #[error(transparent)]
    Protocol(#[from] ProtocolError),
    /// Internal error from protocol handling, IO, or validation.
    #[error(transparent)]
    Internal(#[from] anyhow::Error),
}

impl Error {
    /// Classifies this error for retry and transaction recovery decisions.
    pub fn classification(&self) -> ErrorClassification {
        match self {
            Self::Cancelled => ErrorClassification::default(),
            Self::Admin(error) => error.classification(),
            Self::Consumer(error) => error.classification(),
            Self::Producer(error) => error.classification(),
            Self::ConsumerGroupMetadata(_) => ErrorClassification::default(),
            Self::TransactionState(error) => error.classification(),
            Self::Broker(error) => error.classification(),
            Self::Validation(error) => error.classification(),
            Self::Protocol(error) => error.classification(),
            Self::Internal(_) => ErrorClassification::default(),
        }
    }

    /// Returns whether retrying the same operation may succeed.
    pub fn is_retriable(&self) -> bool {
        self.classification().retriable
    }

    /// Returns whether the client or transaction is permanently unusable.
    pub fn is_fatal(&self) -> bool {
        self.classification().fatal
    }

    /// Returns whether the current transaction must be aborted before reuse.
    pub fn transaction_abort_required(&self) -> bool {
        self.classification().transaction_abort_required
    }
}

#[derive(Debug, Clone, Error)]
/// Kafka broker error returned in a response payload.
pub enum BrokerError {
    /// A broker rejected an operation with a Kafka error code.
    #[error("{operation} failed with broker error {name} ({code}){resource}")]
    Response {
        /// Client operation that received the error.
        operation: &'static str,
        /// Optional topic, partition, group, or coordinator context.
        resource: String,
        /// Numeric Kafka protocol error code.
        code: i16,
        /// Kafka protocol error name.
        name: String,
        /// Whether Kafka marks the error retriable.
        retriable: bool,
        /// Whether this error makes the client or transaction unusable.
        fatal: bool,
        /// Whether a transactional producer must abort before reuse.
        transaction_abort_required: bool,
    },
}

impl BrokerError {
    /// Builds a broker-response error from a Kafka protocol error.
    pub fn response(
        operation: &'static str,
        resource: impl Into<Option<String>>,
        error: ResponseError,
    ) -> Self {
        let resource = resource
            .into()
            .map(|value| format!(" for {value}"))
            .unwrap_or_default();
        Self::Response {
            operation,
            resource,
            code: error.code(),
            name: error.to_string(),
            retriable: error.is_retriable(),
            fatal: false,
            transaction_abort_required: false,
        }
    }

    /// Marks this broker error as fatal to the current client or transaction.
    pub fn fatal(mut self) -> Self {
        let Self::Response { fatal, .. } = &mut self;
        *fatal = true;
        self
    }

    /// Marks this broker error as requiring transaction abort before reuse.
    pub fn transaction_abort_required(mut self) -> Self {
        let Self::Response {
            transaction_abort_required,
            ..
        } = &mut self;
        *transaction_abort_required = true;
        self
    }

    fn classification(&self) -> ErrorClassification {
        match self {
            Self::Response {
                retriable,
                fatal,
                transaction_abort_required,
                ..
            } => ErrorClassification {
                retriable: *retriable,
                fatal: *fatal,
                transaction_abort_required: *transaction_abort_required,
            },
        }
    }
}

#[derive(Debug, Clone, Error)]
/// Errors raised while validating public API inputs.
pub enum ValidationError {
    /// A topic name was empty after trimming.
    #[error("{operation} requires a non-empty topic name")]
    EmptyTopicName {
        /// Operation validating the topic.
        operation: &'static str,
    },
    /// A message did not provide a topic and no default topic was configured.
    #[error("{operation} requires a topic")]
    MissingTopic {
        /// Operation validating the message.
        operation: &'static str,
    },
    /// A partition was negative.
    #[error("{operation} requires a non-negative partition: {partition}")]
    NegativePartition {
        /// Operation validating the partition.
        operation: &'static str,
        /// Requested partition.
        partition: i32,
    },
    /// A resource name was empty after trimming.
    #[error("{resource} names must be non-empty")]
    EmptyResourceName {
        /// Resource type being validated.
        resource: &'static str,
    },
    /// An ACL resource type cannot be used in a concrete ACL binding.
    #[error("ACL resource_type must not be {resource_type}")]
    InvalidAclResourceType {
        /// Invalid resource type.
        resource_type: String,
    },
    /// An ACL pattern type cannot be used in a concrete ACL binding.
    #[error("ACL pattern_type must not be {pattern_type}")]
    InvalidAclPatternType {
        /// Invalid pattern type.
        pattern_type: String,
    },
    /// An ACL principal was empty after trimming.
    #[error("ACL principal must be non-empty")]
    EmptyAclPrincipal,
    /// An ACL host was empty after trimming.
    #[error("ACL host must be non-empty")]
    EmptyAclHost,
    /// An ACL operation cannot be used in a concrete ACL binding.
    #[error("ACL operation must not be {operation}")]
    InvalidAclOperation {
        /// Invalid ACL operation.
        operation: String,
    },
    /// An ACL permission type cannot be used in a concrete ACL binding.
    #[error("ACL permission_type must not be {permission_type}")]
    InvalidAclPermissionType {
        /// Invalid permission type.
        permission_type: String,
    },
    /// A consumer group id was empty after trimming.
    #[error("consumer group id must be non-empty")]
    EmptyConsumerGroupId,
    /// A finalized feature name was empty after trimming.
    #[error("feature names must be non-empty")]
    EmptyFeatureName,
}

impl ValidationError {
    fn classification(&self) -> ErrorClassification {
        ErrorClassification::default()
    }
}

#[derive(Debug, Clone, Error)]
/// Kafka protocol-level errors raised before a typed broker error is available.
pub enum ProtocolError {
    /// A response omitted data the client needs to complete the operation.
    #[error("{operation} response was missing required data: {detail}")]
    MissingResponseData {
        /// Client operation being decoded.
        operation: &'static str,
        /// Missing or malformed response detail.
        detail: String,
    },
    /// A broker advertised an API version lower than required.
    #[error("{api} v{min_version}+ is required, broker only supports v{broker_version}")]
    UnsupportedApiVersion {
        /// API name.
        api: &'static str,
        /// Minimum required API version.
        min_version: i16,
        /// Version advertised by the broker.
        broker_version: i16,
    },
}

impl ProtocolError {
    fn classification(&self) -> ErrorClassification {
        match self {
            Self::MissingResponseData { .. } | Self::UnsupportedApiVersion { .. } => {
                ErrorClassification {
                    fatal: true,
                    ..ErrorClassification::default()
                }
            }
        }
    }
}

#[derive(Debug, Error)]
/// Errors raised before or during admin operations.
pub enum AdminError {
    /// A topic name was empty.
    #[error("topic names must be non-empty")]
    EmptyTopicName,
    /// A topic was requested with an invalid partition count.
    #[error("topic partition count must be positive: {partitions}")]
    InvalidPartitionCount {
        /// Requested partition count.
        partitions: i32,
    },
    /// A topic was requested with an invalid replication factor.
    #[error("topic replication factor must be positive: {replication_factor}")]
    InvalidReplicationFactor {
        /// Requested replication factor.
        replication_factor: i16,
    },
}

impl AdminError {
    fn classification(&self) -> ErrorClassification {
        ErrorClassification::default()
    }
}

#[derive(Debug, Error)]
/// Errors raised by the consumer API.
pub enum ConsumerError {
    /// The runtime stopped before the operation was sent.
    #[error("consumer runtime stopped before {operation}")]
    ThreadStoppedBefore {
        /// Operation that was waiting to be sent.
        operation: &'static str,
    },
    /// The runtime stopped while the operation was waiting for a reply.
    #[error("consumer runtime stopped during {operation}")]
    ThreadStoppedDuring {
        /// Operation that was in flight.
        operation: &'static str,
    },
    /// The background consumer task could not be joined.
    #[error("failed to join consumer runtime: {0}")]
    Join(#[source] JoinError),
    /// A subscription was requested without any topic names.
    #[error("subscribe requires at least one non-empty topic name")]
    EmptySubscription,
    /// A regex subscription was requested with an empty pattern.
    #[error("subscribe_pattern requires a non-empty pattern")]
    EmptySubscriptionPattern,
    /// A regex subscription could not be compiled locally.
    #[error("subscribe_regex requires a valid regular expression: {message}")]
    InvalidSubscriptionRegex {
        /// Regex parser error message.
        message: String,
    },
    /// Another poll call was already active.
    #[error("concurrent poll calls are not supported by this simple consumer")]
    ConcurrentPoll,
    /// A blocking poll was interrupted with `wakeup`.
    #[error("poll was interrupted by wakeup()")]
    Wakeup,
    /// A seek offset was negative.
    #[error("seek offset must be non-negative: {offset}")]
    InvalidSeekOffset {
        /// Requested offset.
        offset: i64,
    },
    /// A topic partition had an empty topic name.
    #[error("topic partition names must be non-empty")]
    EmptyTopicPartition,
    /// The operation needs a partition currently assigned to this consumer.
    #[error("{operation} requires an assigned partition, but {topic}:{partition} is not assigned")]
    PartitionNotAssigned {
        /// Operation that needed the assignment.
        operation: &'static str,
        /// Topic name.
        topic: String,
        /// Partition number.
        partition: i32,
    },
    /// The broker rejected the subscription regex.
    #[error("broker rejected the subscription regex: {message}")]
    InvalidRegularExpression {
        /// Broker error message.
        message: String,
    },
    /// The broker rejected the configured server-side assignor.
    #[error("broker rejected the configured server assignor '{assignor}': {message}")]
    UnsupportedAssignor {
        /// Requested assignor name.
        assignor: String,
        /// Broker error message.
        message: String,
    },
    /// A static member id is still owned by another consumer instance.
    #[error("static member '{instance_id}' is still owned by another consumer: {message}")]
    UnreleasedInstanceId {
        /// Static member instance id.
        instance_id: String,
        /// Broker error message.
        message: String,
    },
    /// The broker fenced this static member instance.
    #[error("static member '{instance_id}' was fenced: {message}")]
    FencedInstanceId {
        /// Static member instance id.
        instance_id: String,
        /// Broker error message.
        message: String,
    },
    /// The consumer runtime is already shutting down.
    #[error("consumer is shutting down")]
    ShuttingDown,
    /// The consumer runtime is in a fatal state.
    #[error("consumer runtime is fatal: {message}")]
    Fatal {
        /// Fatal state reason.
        message: String,
    },
}

impl ConsumerError {
    fn classification(&self) -> ErrorClassification {
        match self {
            Self::ThreadStoppedBefore { .. } | Self::ThreadStoppedDuring { .. } | Self::Join(_) => {
                ErrorClassification {
                    fatal: true,
                    ..ErrorClassification::default()
                }
            }
            Self::UnreleasedInstanceId { .. } => ErrorClassification {
                fatal: true,
                ..ErrorClassification::default()
            },
            Self::FencedInstanceId { .. } | Self::Fatal { .. } => ErrorClassification {
                fatal: true,
                ..ErrorClassification::default()
            },
            Self::EmptySubscription
            | Self::EmptySubscriptionPattern
            | Self::InvalidSubscriptionRegex { .. }
            | Self::ConcurrentPoll
            | Self::Wakeup
            | Self::InvalidSeekOffset { .. }
            | Self::EmptyTopicPartition
            | Self::PartitionNotAssigned { .. }
            | Self::InvalidRegularExpression { .. }
            | Self::UnsupportedAssignor { .. }
            | Self::ShuttingDown => ErrorClassification::default(),
        }
    }
}

#[derive(Debug, Error)]
/// Errors raised by the producer API.
pub enum ProducerError {
    /// Idempotence was enabled without `acks=-1`.
    #[error("idempotent producers require acks=-1")]
    IdempotenceRequiresAcksAll,
    /// Idempotence was enabled without retries.
    #[error("idempotent producers require max_retries > 0")]
    IdempotenceRequiresRetries,
    /// Transactions require all replicas to acknowledge records.
    #[error(
        "transactional producers require acks=-1 so the broker can commit the full transaction"
    )]
    TransactionalRequiresAcksAll,
    /// The broker did not report the transaction feature level.
    #[error("broker did not advertise finalized feature level for transaction.version")]
    MissingTransactionVersionFeature,
    /// The broker transaction feature level is too old.
    #[error(
        "broker finalized transaction.version={level}, but transaction v2 requires transaction.version>=2"
    )]
    UnsupportedTransactionVersion {
        /// Broker feature level.
        level: i16,
    },
    /// A broker API version is older than this client needs.
    #[error("{api} v{min_version}+ is required, broker only supports v{broker_version}")]
    UnsupportedApiVersion {
        /// API name.
        api: &'static str,
        /// Minimum required API version.
        min_version: i16,
        /// Version advertised by the broker.
        broker_version: i16,
    },
    /// The operation requires a configured transactional id.
    #[error("producer is not configured with a transactional_id")]
    NotTransactional,
    /// Metadata reported no partitions for the topic.
    #[error("topic '{topic}' has no partitions in metadata")]
    TopicHasNoPartitions {
        /// Topic name.
        topic: String,
    },
    /// Metadata did not include partition data for a topic.
    #[error("missing partition metadata for topic '{topic}'")]
    MissingPartitionMetadata {
        /// Topic name.
        topic: String,
    },
    /// Metadata was not available before `max_block` elapsed.
    #[error("metadata for topic '{topic}' was not available within max_block={max_block_ms}ms")]
    MetadataTimeout {
        /// Topic name.
        topic: String,
        /// Configured max block in milliseconds.
        max_block_ms: u128,
    },
    /// The transaction manager unexpectedly disappeared.
    #[error("transaction manager is unavailable during {operation}")]
    TransactionManagerUnavailable {
        /// Operation that needed the transaction manager.
        operation: &'static str,
    },
    /// A transactional producer id was needed but had not been initialized.
    #[error("transactional producer id is not initialized during {operation}")]
    TransactionalProducerNotInitialized {
        /// Operation that needed the producer id.
        operation: &'static str,
    },
    /// A coordinator connection was needed but had not been opened.
    #[error("transaction coordinator connection is missing")]
    TransactionCoordinatorConnectionMissing,
    /// A producer coordinator request exhausted all attempts.
    #[error("{operation} exhausted {attempts} attempts{resource}")]
    AttemptsExhausted {
        /// Operation that exhausted retries.
        operation: &'static str,
        /// Optional transactional id, group id, topic, or partition context.
        resource: String,
        /// Number of attempts made.
        attempts: usize,
    },
    /// A produce batch failed before the broker accepted it.
    #[error("produce batch failed for {topic}:{partition}: {message}")]
    BatchFailed {
        /// Topic name.
        topic: String,
        /// Partition number.
        partition: i32,
        /// Failure detail.
        message: String,
    },
    /// A record is too large for the configured producer limits.
    #[error("producer record is {size} bytes, larger than {limit_name}={limit}")]
    RecordTooLarge {
        /// Estimated serialized record size.
        size: usize,
        /// Config limit name.
        limit_name: &'static str,
        /// Configured limit.
        limit: usize,
    },
    /// A Produce request is too large for the configured producer limits.
    #[error("produce request is {size} bytes, larger than max_request_size={limit}")]
    RequestTooLarge {
        /// Estimated serialized request size.
        size: usize,
        /// Configured limit.
        limit: usize,
    },
    /// The accumulator did not have room for a record before `max_block` elapsed.
    #[error(
        "producer buffer has {buffered} of {limit} bytes queued and cannot accept {required} more bytes within max_block={max_block_ms}ms"
    )]
    BufferExhausted {
        /// Current estimated queued bytes.
        buffered: usize,
        /// Additional bytes needed for the record.
        required: usize,
        /// Configured buffer size.
        limit: usize,
        /// Configured max block in milliseconds.
        max_block_ms: u128,
    },
    /// A transactional operation failed and the open transaction must be aborted.
    #[error("{operation} failed and the transaction must be aborted: {message}")]
    TransactionAbortRequired {
        /// Operation that failed.
        operation: &'static str,
        /// Failure detail.
        message: String,
    },
    /// A transactional operation failed and the producer is no longer usable.
    #[error("{operation} failed and the transactional producer is no longer usable: {message}")]
    TransactionFatal {
        /// Operation that failed.
        operation: &'static str,
        /// Failure detail.
        message: String,
    },
    /// The sender task stopped before the operation was sent.
    #[error("producer runtime stopped before {operation}")]
    RuntimeStoppedBefore {
        /// Operation that was waiting to be sent.
        operation: &'static str,
    },
    /// The sender task stopped while the operation was waiting for a reply.
    #[error("producer runtime stopped during {operation}")]
    RuntimeStoppedDuring {
        /// Operation that was in flight.
        operation: &'static str,
    },
    /// The sender task could not be joined.
    #[error("failed to join producer runtime: {0}")]
    Join(#[source] JoinError),
}

impl ProducerError {
    fn classification(&self) -> ErrorClassification {
        match self {
            Self::MissingTransactionVersionFeature
            | Self::UnsupportedTransactionVersion { .. }
            | Self::UnsupportedApiVersion { .. }
            | Self::TransactionFatal { .. }
            | Self::TransactionManagerUnavailable { .. }
            | Self::TransactionalProducerNotInitialized { .. }
            | Self::TransactionCoordinatorConnectionMissing
            | Self::RuntimeStoppedBefore { .. }
            | Self::RuntimeStoppedDuring { .. }
            | Self::Join(_) => ErrorClassification {
                fatal: true,
                ..ErrorClassification::default()
            },
            Self::TransactionAbortRequired { .. } => ErrorClassification {
                transaction_abort_required: true,
                ..ErrorClassification::default()
            },
            Self::AttemptsExhausted { .. } => ErrorClassification {
                retriable: true,
                ..ErrorClassification::default()
            },
            Self::IdempotenceRequiresAcksAll
            | Self::IdempotenceRequiresRetries
            | Self::TransactionalRequiresAcksAll
            | Self::NotTransactional
            | Self::TopicHasNoPartitions { .. }
            | Self::MissingPartitionMetadata { .. }
            | Self::MetadataTimeout { .. }
            | Self::BatchFailed { .. }
            | Self::RecordTooLarge { .. }
            | Self::RequestTooLarge { .. }
            | Self::BufferExhausted { .. } => ErrorClassification::default(),
        }
    }
}

#[derive(Debug, Error)]
/// Errors found while validating consumer group metadata.
pub enum ConsumerGroupMetadataError {
    /// The group id was empty.
    #[error("consumer group metadata requires a non-empty group_id")]
    EmptyGroupId,
    /// Active group metadata needs a member id.
    #[error("consumer group metadata has generation_id > 0 but no member_id")]
    MissingMemberId,
}

#[derive(Debug, Error)]
/// Errors raised by the producer transaction state machine.
pub enum TransactionStateError {
    /// A transaction is already open.
    #[error("transaction already in progress")]
    AlreadyInProgress,
    /// The transaction is already completing.
    #[error("transaction is already completing with {0}")]
    Completing(&'static str),
    /// The transaction must be aborted before it can be used again.
    #[error("transaction must be aborted before reuse: {0}")]
    MustAbortBeforeReuse(String),
    /// The transaction is permanently failed.
    #[error("transaction is unusable: {0}")]
    Fatal(String),
    /// Records were sent before a transaction was started.
    #[error("transactional send requires begin_transaction() before send()")]
    AppendWithoutBegin,
    /// Offsets were sent before a transaction was started.
    #[error("send_offsets_to_transaction requires begin_transaction() first")]
    SendOffsetsWithoutBegin,
    /// The transaction needs an abort before more work can happen.
    #[error("transaction has failed and must be aborted: {0}")]
    AbortRequired(String),
    /// The transaction cannot be committed and must be aborted.
    #[error("transaction cannot be committed and must be aborted: {0}")]
    CommitRequiresAbort(String),
    /// There is no active transaction to finish.
    #[error("no active transaction to complete")]
    NoActiveTransaction,
    /// Shutdown found an active transaction.
    #[error("shutdown stopped with an active transaction still in progress")]
    ShutdownWithActiveTransaction,
    /// Shutdown found a transaction completion in progress.
    #[error("shutdown stopped while transaction was still completing with {0}")]
    ShutdownWhileCompleting(&'static str),
    /// Shutdown found a failed transaction that still needs aborting.
    #[error("shutdown stopped with an un-aborted failed transaction: {0}")]
    ShutdownAbortRequired(String),
    /// Shutdown found a fatal transaction error.
    #[error("transaction failed before shutdown: {0}")]
    ShutdownFatal(String),
}

impl TransactionStateError {
    fn classification(&self) -> ErrorClassification {
        match self {
            Self::Fatal(_) | Self::ShutdownFatal(_) => ErrorClassification {
                fatal: true,
                ..ErrorClassification::default()
            },
            Self::MustAbortBeforeReuse(_)
            | Self::AbortRequired(_)
            | Self::CommitRequiresAbort(_)
            | Self::ShutdownAbortRequired(_) => ErrorClassification {
                transaction_abort_required: true,
                ..ErrorClassification::default()
            },
            Self::AlreadyInProgress
            | Self::Completing(_)
            | Self::AppendWithoutBegin
            | Self::SendOffsetsWithoutBegin
            | Self::NoActiveTransaction
            | Self::ShutdownWithActiveTransaction
            | Self::ShutdownWhileCompleting(_) => ErrorClassification::default(),
        }
    }
}

#[cfg(test)]
mod tests {
    use kafka_protocol::error::ResponseError;

    use super::{
        BrokerError, ConsumerError, Error, ProducerError, TransactionStateError, ValidationError,
    };

    #[test]
    fn broker_errors_preserve_retriable_flag() {
        let error = Error::Broker(BrokerError::response(
            "produce",
            Some("orders:0".to_owned()),
            ResponseError::NotLeaderOrFollower,
        ));

        assert!(error.is_retriable());
        assert!(!error.is_fatal());
        assert!(!error.transaction_abort_required());
    }

    #[test]
    fn broker_errors_can_mark_transaction_abort_required() {
        let error = Error::Broker(
            BrokerError::response(
                "send_offsets_to_transaction",
                Some("orders-reader".to_owned()),
                ResponseError::UnknownServerError,
            )
            .transaction_abort_required(),
        );

        assert!(!error.is_fatal());
        assert!(error.transaction_abort_required());
    }

    #[test]
    fn broker_errors_can_mark_fatal() {
        let error = Error::Broker(
            BrokerError::response(
                "end_transaction",
                None::<String>,
                ResponseError::ProducerFenced,
            )
            .fatal(),
        );

        assert!(error.is_fatal());
        assert!(!error.transaction_abort_required());
    }

    #[test]
    fn static_member_ownership_errors_are_fatal() {
        let unreleased = Error::Consumer(ConsumerError::UnreleasedInstanceId {
            instance_id: "instance-a".to_owned(),
            message: "still owned".to_owned(),
        });
        let fenced = Error::Consumer(ConsumerError::FencedInstanceId {
            instance_id: "instance-a".to_owned(),
            message: "fenced".to_owned(),
        });

        assert!(unreleased.is_fatal());
        assert!(!unreleased.is_retriable());
        assert!(fenced.is_fatal());
        assert!(!fenced.is_retriable());
    }

    #[test]
    fn transaction_state_errors_are_classified() {
        let abort_required = Error::TransactionState(TransactionStateError::AbortRequired(
            "send failed".to_owned(),
        ));
        let fatal = Error::TransactionState(TransactionStateError::Fatal("fenced".to_owned()));

        assert!(abort_required.transaction_abort_required());
        assert!(!abort_required.is_fatal());
        assert!(fatal.is_fatal());
    }

    #[test]
    fn producer_transaction_runtime_errors_are_classified() {
        let abort_required = Error::Producer(ProducerError::TransactionAbortRequired {
            operation: "send_offsets_to_transaction",
            message: "connection reset".to_owned(),
        });
        let fatal = Error::Producer(ProducerError::TransactionFatal {
            operation: "commit_transaction",
            message: "connection reset".to_owned(),
        });

        assert!(abort_required.transaction_abort_required());
        assert!(!abort_required.is_fatal());
        assert!(fatal.is_fatal());
        assert!(!fatal.transaction_abort_required());
    }

    #[test]
    fn validation_errors_are_not_retriable_or_fatal() {
        let error = Error::Validation(ValidationError::MissingTopic {
            operation: "message conversion",
        });

        assert!(!error.is_retriable());
        assert!(!error.is_fatal());
        assert!(!error.transaction_abort_required());
    }

    #[test]
    fn producer_attempt_exhaustion_is_retriable_and_typed() {
        let error = Error::Producer(ProducerError::AttemptsExhausted {
            operation: "find_transaction_coordinator",
            resource: " for transactional_id 'tx-a'".to_owned(),
            attempts: 11,
        });

        assert!(error.is_retriable());
        assert!(!error.is_fatal());
        assert_eq!(
            error.to_string(),
            "find_transaction_coordinator exhausted 11 attempts for transactional_id 'tx-a'"
        );
    }
}