1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
//! The crate error type shared by the broker, publishers, and subscribers.
use std::error::Error as StdError;
use thiserror::Error;
/// Errors returned by [`KafkaBroker`](crate::KafkaBroker) and the types it hands out.
///
/// Underlying [`rdkafka`](https://docs.rs/rdkafka) errors are boxed as sources so the client
/// library does not leak into this crate's public API surface.
#[derive(Debug, Error)]
#[non_exhaustive]
pub enum KafkaError {
/// Creating a client failed or the cluster was unreachable during the connect probe.
#[error("kafka connection error: {0}")]
Connect(#[source] Box<dyn StdError + Send + Sync>),
/// Publishing a message failed or the broker did not confirm its delivery.
#[error("kafka publish error: {0}")]
Publish(#[source] Box<dyn StdError + Send + Sync>),
/// Creating a consumer or subscribing it to its topic failed.
#[error("kafka subscribe error: {0}")]
Subscribe(#[source] Box<dyn StdError + Send + Sync>),
/// Receiving a delivery from an open consumer failed.
#[error("kafka consume error: {0}")]
Consume(#[source] Box<dyn StdError + Send + Sync>),
/// An operation needed the live connection before `Broker::connect` resolved it.
///
/// The runtime connects the broker once at startup; a publisher handed out earlier resolves
/// the shared connection on first use. Seeing this error means the operation ran before
/// `connect` completed.
#[error("kafka broker is not connected; `Broker::connect` must complete first")]
NotConnected,
/// The requested combination of options cannot be executed.
///
/// The message names the offending option and the remediation.
#[error("invalid options: {0}")]
InvalidOptions(String),
/// `begin_transaction` found a transaction already open on this publisher.
///
/// One producer runs one transaction at a time, so a second begin means two flows share
/// one publisher; erroring beats silently merging their messages into one transaction.
/// Concurrent transactional flows need distinct publishers - one per partition via
/// [`TransactionalPartitions`](crate::TransactionalPartitions), or distinct explicit ids.
#[error(
"a transaction is already open on this publisher; one publisher runs one transaction \
at a time - use distinct publishers (for example TransactionalPartitions) for \
concurrent transactional flows"
)]
TransactionBusy,
/// A Schema Registry request failed: unreachable registry, rejected credentials, an
/// unknown schema id or subject, or a schema the registry refused.
#[cfg(feature = "schema-registry")]
#[error("schema registry error: {0}")]
SchemaRegistry(#[source] Box<dyn StdError + Send + Sync>),
}
impl KafkaError {
pub(crate) fn connect(err: rdkafka::error::KafkaError) -> Self {
Self::Connect(Box::new(err))
}
pub(crate) fn publish(err: rdkafka::error::KafkaError) -> Self {
Self::Publish(Box::new(err))
}
#[cfg(feature = "schema-registry")]
pub(crate) fn schema_registry(err: impl StdError + Send + Sync + 'static) -> Self {
Self::SchemaRegistry(Box::new(err))
}
pub(crate) fn subscribe(err: rdkafka::error::KafkaError) -> Self {
Self::Subscribe(Box::new(err))
}
pub(crate) fn consume(err: rdkafka::error::KafkaError) -> Self {
Self::Consume(Box::new(err))
}
}