ruststream_rdkafka/
error.rs

1//! The crate error type shared by the broker, publishers, and subscribers.
2
3use std::error::Error as StdError;
4
5use thiserror::Error;
6
7/// Errors returned by [`KafkaBroker`](crate::KafkaBroker) and the types it hands out.
8///
9/// Underlying [`rdkafka`](https://docs.rs/rdkafka) errors are boxed as sources so the client
10/// library does not leak into this crate's public API surface.
11#[derive(Debug, Error)]
12#[non_exhaustive]
13pub enum KafkaError {
14    /// Creating a client failed or the cluster was unreachable during the connect probe.
15    #[error("kafka connection error: {0}")]
16    Connect(#[source] Box<dyn StdError + Send + Sync>),
17
18    /// Publishing a message failed or the broker did not confirm its delivery.
19    #[error("kafka publish error: {0}")]
20    Publish(#[source] Box<dyn StdError + Send + Sync>),
21
22    /// Creating a consumer or subscribing it to its topic failed.
23    #[error("kafka subscribe error: {0}")]
24    Subscribe(#[source] Box<dyn StdError + Send + Sync>),
25
26    /// Receiving a delivery from an open consumer failed.
27    #[error("kafka consume error: {0}")]
28    Consume(#[source] Box<dyn StdError + Send + Sync>),
29
30    /// A [`KafkaRetryPublisher`](crate::KafkaRetryPublisher) was used before its broker
31    /// connected.
32    ///
33    /// Only the early publisher can report this: it is the one handle minted before
34    /// [`Broker::connect`](ruststream::Broker::connect), for builder-time wiring that needs a
35    /// live publisher (`retry_via`). Everything on the policy path pairs with the connected
36    /// broker, so "not connected" is not representable there.
37    #[error("kafka broker is not connected yet; cannot reach {topic}")]
38    NotConnected {
39        /// The topic the operation targeted.
40        topic: String,
41    },
42
43    /// A handle aliasing the connection was used after the broker shut down.
44    ///
45    /// The lifecycle ladder makes misuse through the owner's handle a compile error:
46    /// [`ConnectedBroker::shutdown`](ruststream::ConnectedBroker::shutdown) consumes the
47    /// connected broker. Publishers paired off it earlier, and subscriptions still open, keep
48    /// aliasing the closed connection, so their operations report this instead of silently
49    /// succeeding against a dead connection.
50    #[error("kafka connection is closed; cannot reach {topic}")]
51    Closed {
52        /// The topic the operation targeted, or the transactional id of a transaction control
53        /// call.
54        topic: String,
55    },
56
57    /// The requested combination of options cannot be executed.
58    ///
59    /// The message names the offending option and the remediation.
60    #[error("invalid options: {0}")]
61    InvalidOptions(String),
62
63    /// `begin_transaction` found a transaction already open on this publisher.
64    ///
65    /// One producer runs one transaction at a time, so a second begin means two flows share
66    /// one publisher; erroring beats silently merging their messages into one transaction.
67    /// Concurrent transactional flows need distinct publishers - one per partition via
68    /// [`TransactionalPartitions`](crate::TransactionalPartitions), or distinct explicit ids.
69    #[error(
70        "a transaction is already open on publisher {id}; one publisher runs one transaction \
71         at a time - use distinct publishers (for example the per-partition set) for \
72         concurrent transactional flows"
73    )]
74    TransactionBusy {
75        /// The transactional id of the publisher that already has an open transaction.
76        id: String,
77    },
78
79    /// `commit` or `abort` was called with no transaction open on this publisher.
80    #[error("no transaction is open on publisher {id}; `begin_transaction` opens one")]
81    NoTransaction {
82        /// The transactional id of the publisher the call was made on.
83        id: String,
84    },
85
86    /// A Schema Registry request failed: unreachable registry, rejected credentials, an
87    /// unknown schema id or subject, or a schema the registry refused.
88    #[cfg(feature = "schema-registry")]
89    #[error("schema registry error: {0}")]
90    SchemaRegistry(#[source] Box<dyn StdError + Send + Sync>),
91}
92
93impl KafkaError {
94    pub(crate) fn connect(err: rdkafka::error::KafkaError) -> Self {
95        Self::Connect(Box::new(err))
96    }
97
98    pub(crate) fn publish(err: rdkafka::error::KafkaError) -> Self {
99        Self::Publish(Box::new(err))
100    }
101
102    #[cfg(feature = "schema-registry")]
103    pub(crate) fn schema_registry(err: impl StdError + Send + Sync + 'static) -> Self {
104        Self::SchemaRegistry(Box::new(err))
105    }
106
107    pub(crate) fn subscribe(err: rdkafka::error::KafkaError) -> Self {
108        Self::Subscribe(Box::new(err))
109    }
110
111    pub(crate) fn consume(err: rdkafka::error::KafkaError) -> Self {
112        Self::Consume(Box::new(err))
113    }
114}
ruststream_rdkafka/error.rs

ruststream_rdkafka/
error.rs
