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    /// An operation needed the live connection before `Broker::connect` resolved it.
31    ///
32    /// The runtime connects the broker once at startup; a publisher handed out earlier resolves
33    /// the shared connection on first use. Seeing this error means the operation ran before
34    /// `connect` completed.
35    #[error("kafka broker is not connected; `Broker::connect` must complete first")]
36    NotConnected,
37
38    /// The requested combination of options cannot be executed.
39    ///
40    /// The message names the offending option and the remediation.
41    #[error("invalid options: {0}")]
42    InvalidOptions(String),
43
44    /// `begin_transaction` found a transaction already open on this publisher.
45    ///
46    /// One producer runs one transaction at a time, so a second begin means two flows share
47    /// one publisher; erroring beats silently merging their messages into one transaction.
48    /// Concurrent transactional flows need distinct publishers - one per partition via
49    /// [`TransactionalPartitions`](crate::TransactionalPartitions), or distinct explicit ids.
50    #[error(
51        "a transaction is already open on this publisher; one publisher runs one transaction \
52         at a time - use distinct publishers (for example TransactionalPartitions) for \
53         concurrent transactional flows"
54    )]
55    TransactionBusy,
56
57    /// A Schema Registry request failed: unreachable registry, rejected credentials, an
58    /// unknown schema id or subject, or a schema the registry refused.
59    #[cfg(feature = "schema-registry")]
60    #[error("schema registry error: {0}")]
61    SchemaRegistry(#[source] Box<dyn StdError + Send + Sync>),
62}
63
64impl KafkaError {
65    pub(crate) fn connect(err: rdkafka::error::KafkaError) -> Self {
66        Self::Connect(Box::new(err))
67    }
68
69    pub(crate) fn publish(err: rdkafka::error::KafkaError) -> Self {
70        Self::Publish(Box::new(err))
71    }
72
73    #[cfg(feature = "schema-registry")]
74    pub(crate) fn schema_registry(err: impl StdError + Send + Sync + 'static) -> Self {
75        Self::SchemaRegistry(Box::new(err))
76    }
77
78    pub(crate) fn subscribe(err: rdkafka::error::KafkaError) -> Self {
79        Self::Subscribe(Box::new(err))
80    }
81
82    pub(crate) fn consume(err: rdkafka::error::KafkaError) -> Self {
83        Self::Consume(Box::new(err))
84    }
85}
ruststream_rdkafka/error.rs

ruststream_rdkafka/
error.rs
