krabka_client_core/
error.rs

1//! Error type for `krabka-client-core`.
2
3use std::{io, net::SocketAddr};
4
5use krabka_units::{Time, fmt::Human as _};
6use thiserror::Error;
7
8use crate::sasl;
9
10/// Errors returned by `Client`, `Connection`, and the broker pool.
11#[derive(Debug, Error)]
12#[non_exhaustive]
13pub enum ClientError {
14    #[error("connect to {addr}: {source}")]
15    Connect {
16        addr: SocketAddr,
17        #[source]
18        source: io::Error,
19    },
20
21    /// The TCP connection opened, but the TLS handshake failed before the
22    /// peer gave a verdict, for example on EOF, a reset, or a `close_notify`
23    /// alert. The caller can connect again. A rejection by the peer is
24    /// [`ClientError::Authentication`].
25    #[error("TLS handshake with {addr}: {source}")]
26    Tls {
27        addr: SocketAddr,
28        #[source]
29        source: io::Error,
30    },
31
32    /// The SASL exchange failed after the TCP connection, and TLS if the
33    /// protocol uses it, came up, and the broker did not reject the
34    /// credentials. The stream failed, a frame was too large, or the broker
35    /// sent an error code that is not an authentication error. The caller can
36    /// connect again. A rejection by the broker is
37    /// [`ClientError::Authentication`].
38    #[error("SASL authentication with {addr}: {source}")]
39    Sasl {
40        addr: SocketAddr,
41        #[source]
42        source: sasl::OutboundSaslError,
43    },
44
45    /// The TLS peer or the SASL broker rejected authentication.
46    ///
47    /// Kafka's `NetworkClient.processDisconnection` stores this
48    /// `AuthenticationException` for the node, and the client raises it from
49    /// the next call without a retry. Do not evict and retry on this error.
50    #[error("authentication with {addr} failed: {source}")]
51    Authentication {
52        addr: SocketAddr,
53        #[source]
54        source: AuthenticationError,
55    },
56
57    #[error("connection closed")]
58    Disconnected,
59
60    #[error("invalid client configuration: {0}")]
61    InvalidConfig(String),
62
63    /// An argument of a call is invalid, as Kafka's clients throw an
64    /// `IllegalArgumentException`.
65    #[error("invalid argument: {0}")]
66    InvalidArgument(String),
67
68    /// The client does not push metrics (`enable_metrics_push` is `false`),
69    /// so it has no client instance id. Kafka's clients throw an
70    /// `IllegalStateException` with this message from `clientInstanceId`.
71    #[error("Telemetry is not enabled. Set config `enable.metrics.push` to `true`.")]
72    TelemetryDisabled,
73
74    #[error("request timed out after {}", .0.human())]
75    Timeout(Time),
76
77    #[error(
78        "incompatible version: broker supports {broker_min}..={broker_max}, \
79         client wants {client_min}..={client_max} for api_key {api_key}"
80    )]
81    IncompatibleVersion {
82        api_key: i16,
83        broker_min: i16,
84        broker_max: i16,
85        client_min: i16,
86        client_max: i16,
87    },
88
89    #[error("protocol error from server: {error_code}")]
90    Server { error_code: i16 },
91
92    #[error("FindCoordinator returned no entry for key {key:?}")]
93    NoCoordinator { key: String },
94
95    #[error("codec: {0}")]
96    Codec(#[from] krabka_protocol::ProtocolError),
97
98    #[error("I/O: {0}")]
99    Io(#[from] io::Error),
100}
101
102impl ClientError {
103    /// Whether the peer rejected authentication. Kafka raises such an
104    /// `AuthenticationException` to the application without a retry.
105    #[must_use]
106    pub const fn is_authentication_failure(&self) -> bool {
107        matches!(self, Self::Authentication { .. })
108    }
109}
110
111/// The reason that a peer rejected authentication. Each variant is a subclass
112/// of Kafka's `AuthenticationException`.
113#[derive(Debug, Error)]
114pub enum AuthenticationError {
115    /// Kafka `SslAuthenticationException`: the TLS handshake failed with an
116    /// error from the TLS engine, for example an untrusted certificate, a
117    /// fatal alert from the peer, or a peer that does not speak TLS.
118    #[error("TLS handshake failed: {0}")]
119    Tls(#[source] io::Error),
120
121    /// The broker or the SASL mechanism rejected authentication.
122    #[error(transparent)]
123    Sasl(#[from] sasl::SaslAuthenticationError),
124}
125
126#[cfg(test)]
127mod tests {
128    use assert2::assert;
129
130    use super::*;
131
132    #[test]
133    fn display_is_useful() {
134        let e = ClientError::Timeout(krabka_units::secs(5));
135        assert!(e.to_string() == "request timed out after 5s");
136    }
137
138    #[test]
139    fn incompatible_version_displays_full_range() {
140        let e = ClientError::IncompatibleVersion {
141            api_key: 0,
142            broker_min: 0,
143            broker_max: 5,
144            client_min: 7,
145            client_max: 10,
146        };
147        assert!(e.to_string().contains("api_key 0"));
148        assert!(e.to_string().contains("broker supports 0..=5"));
149    }
150}
krabka_client_core/error.rs

krabka_client_core/
error.rs
