krabka_client_producer/
error.rs

1use std::time::Duration;
2
3use thiserror::Error;
4
5#[derive(Debug, Error)]
6#[non_exhaustive]
7pub enum ProducerError {
8    #[error("client: {0}")]
9    Client(#[from] krabka_client_core::ClientError),
10
11    #[error("protocol: {0}")]
12    Protocol(#[from] krabka_protocol::ProtocolError),
13
14    #[error("broker error_code {0}")]
15    Server(i16),
16
17    #[error("fenced by newer producer instance")]
18    FencedProducer,
19
20    #[error("invalid config: {0}")]
21    InvalidConfig(String),
22
23    /// The record names a negative partition. Kafka's `ProducerRecord`
24    /// constructor throws `IllegalArgumentException` with the same message.
25    #[error("Invalid partition: {0}. Partition number should always be non-negative or null.")]
26    InvalidPartition(i32),
27
28    /// The record carries a negative timestamp. Kafka's `ProducerRecord`
29    /// constructor throws `IllegalArgumentException` with the same message.
30    #[error("Invalid timestamp: {0}. Timestamp should always be non-negative or null.")]
31    InvalidTimestamp(i64),
32
33    /// A custom partitioner (`partitioner.class`) returned a negative
34    /// partition. Kafka's `KafkaProducer.partition` throws
35    /// `IllegalArgumentException` with the same message.
36    #[error(
37        "The partitioner generated an invalid partition number: {0}. Partition number should always be non-negative."
38    )]
39    InvalidPartitionerResult(i32),
40
41    #[error("batch too large: {batch_size} > max")]
42    BatchTooLarge { batch_size: usize },
43
44    /// The record is larger than a limit of the producer. Kafka's
45    /// `KafkaProducer.ensureValidRecordSize` throws `RecordTooLargeException`
46    /// with the same message for each limit.
47    #[error("{}", record_too_large_message(*record_size, *limit))]
48    RecordTooLarge {
49        /// Kafka's upper bound of the serialized size of the record, in bytes.
50        record_size: usize,
51        /// The limit that the record is larger than.
52        limit: RecordSizeLimit,
53    },
54
55    /// `send` waited `max_block` (the part of `max_block` that the metadata
56    /// wait left) for buffer memory, and the memory did not become free.
57    /// Kafka's `BufferPool.allocate` throws `BufferExhaustedException` with the
58    /// same message.
59    #[error(
60        "Failed to allocate {size} bytes within the configured max blocking time {} ms. Total memory: {total} bytes. Available memory: {available} bytes. Poolable size: {poolable} bytes",
61        max_block.as_millis()
62    )]
63    BufferExhausted {
64        size: usize,
65        max_block: Duration,
66        total: usize,
67        available: usize,
68        poolable: usize,
69    },
70
71    #[error("producer closed")]
72    Closed,
73
74    #[error("flush timed out")]
75    FlushTimeout,
76
77    /// `send` waited `waited` (the configured `max_block`) for metadata that
78    /// holds the topic, and the partition of the record when it names one.
79    /// Kafka's `KafkaProducer.waitOnMetadata` throws `TimeoutException` with
80    /// the same message.
81    #[error("{}", metadata_timeout_message(topic, *partition, *partition_count, *waited))]
82    MetadataTimeout {
83        topic: String,
84        partition: Option<i32>,
85        partition_count: Option<i32>,
86        waited: Duration,
87    },
88
89    /// The batch ran out of retries, or its routing budget ended, before the
90    /// broker acknowledged it. Kafka raises `TimeoutException` for the same
91    /// case (`Sender.sendProducerData` and `RecordAccumulator.expiredBatches`).
92    #[error("the batch was not acknowledged before its retries ran out")]
93    SendTimeout,
94
95    #[error("compression: {0}")]
96    Compression(#[from] krabka_compression::CompressionError),
97
98    #[error("producer is not transactional (no transactional_id configured)")]
99    NotTransactional,
100
101    #[error("invalid transaction state: {0}")]
102    InvalidTransactionState(&'static str),
103
104    /// The record's batch was still queued in the accumulator, undrained,
105    /// when the application aborted the transaction. Kafka's
106    /// `Sender.maybeSendAndPollTransactionalRequest` fails every undrained
107    /// batch this way instead of sending it:
108    /// `accumulator.abortUndrainedBatches(new TransactionAbortedException())`.
109    #[error("the transaction was aborted; the record was never sent")]
110    TransactionAborted,
111
112    #[error("concurrent transactions on the same transactional_id")]
113    ConcurrentTransactions,
114
115    #[error(
116        "transaction outcome is unknown; call init_transactions before sending or beginning another transaction"
117    )]
118    RecoveryRequired,
119
120    /// A transaction coordinator answered with a fatal code, and the
121    /// transactional producer can do no more transactional work. Every later
122    /// `init_transactions`, `begin_transaction`, `send`,
123    /// `send_offsets_to_transaction`, commit and abort fails with this error.
124    /// Kafka's `TransactionManager` moves to `FATAL_ERROR` for the same codes,
125    /// and the application can only close the producer.
126    #[error(
127        "the transactional producer is in a fatal error state after broker error_code {0}; close the producer"
128    )]
129    FatalTransactionError(i16),
130}
131
132/// The producer limit that a record is larger than.
133#[derive(Debug, Clone, Copy, PartialEq, Eq)]
134pub enum RecordSizeLimit {
135    /// The `max_request_size` setting, in bytes. Kafka's `max.request.size`.
136    MaxRequestSize(usize),
137    /// The `buffer_memory` setting. Kafka's `buffer.memory`.
138    BufferMemory,
139}
140
141/// Kafka's `KafkaProducer.ensureValidRecordSize` messages.
142fn record_too_large_message(record_size: usize, limit: RecordSizeLimit) -> String {
143    match limit {
144        RecordSizeLimit::MaxRequestSize(max_request_size) => format!(
145            "The message is {record_size} bytes when serialized which is larger than {max_request_size}, which is the value of the max_request_size configuration."
146        ),
147        RecordSizeLimit::BufferMemory => format!(
148            "The message is {record_size} bytes when serialized which is larger than the total memory buffer you have configured with the buffer_memory configuration."
149        ),
150    }
151}
152
153/// Kafka's `KafkaProducer.getErrorMessage`.
154fn metadata_timeout_message(
155    topic: &str,
156    partition: Option<i32>,
157    partition_count: Option<i32>,
158    waited: Duration,
159) -> String {
160    let waited_ms = waited.as_millis();
161    match (partition, partition_count) {
162        (Some(partition), Some(count)) => format!(
163            "Partition {partition} of topic {topic} with partition count {count} is not present in metadata after {waited_ms} ms."
164        ),
165        _ => format!("Topic {topic} not present in metadata after {waited_ms} ms."),
166    }
167}
168
169#[cfg(test)]
170mod tests {
171
172    use super::*;
173
174    #[test]
175    fn display_messages() {
176        for (_name, error, expected) in [
177            (
178                "fenced producer",
179                ProducerError::FencedProducer,
180                "fenced by newer producer instance",
181            ),
182            (
183                "invalid config",
184                ProducerError::InvalidConfig("idempotence requires acks=all".to_owned()),
185                "invalid config: idempotence requires acks=all",
186            ),
187        ] {
188            assert2::assert!(error.to_string() == expected);
189        }
190    }
191}
krabka_client_producer/error.rs

krabka_client_producer/
error.rs
