krabka_client_consumer/
error.rs

1//! Error type for `krabka-client-consumer`.
2
3use std::collections::BTreeSet;
4
5use krabka_client_core::ClientError;
6use thiserror::Error;
7
8pub fn is_transport_error(error: &krabka_client_core::ClientError, includes_timeout: bool) -> bool {
9    matches!(
10        error,
11        ClientError::Connect { .. }
12            | ClientError::Tls { .. }
13            | ClientError::Sasl { .. }
14            | ClientError::Disconnected
15            | ClientError::Io(_)
16    ) || (includes_timeout && matches!(error, ClientError::Timeout(_)))
17}
18
19/// Errors returned by [`Consumer`](crate::Consumer).
20#[derive(Debug, Error)]
21#[non_exhaustive]
22pub enum ConsumerError {
23    #[error("client: {0}")]
24    Client(#[from] krabka_client_core::ClientError),
25
26    #[error("protocol: {0}")]
27    Protocol(#[from] krabka_protocol::ProtocolError),
28
29    #[error("rebalance failed: {0}")]
30    RebalanceFailed(String),
31
32    /// A rebalance listener callback failed. Kafka's `poll` throws a
33    /// `KafkaException` with the cause.
34    #[error("rebalance listener failed: {0}")]
35    RebalanceListenerFailed(String),
36
37    /// A builder setting is not valid. Kafka's `ConfigException`.
38    #[error("invalid configuration: {0}")]
39    InvalidConfig(String),
40
41    #[error("startup failed after joining group: {0}")]
42    StartupAfterJoin(Box<Self>),
43
44    #[error("not subscribed to any topic")]
45    NotSubscribed,
46
47    #[error("illegal state: {0}")]
48    IllegalState(String),
49
50    /// An argument of a call is not valid. Kafka's `IllegalArgumentException`.
51    #[error("invalid argument: {0}")]
52    InvalidArgument(String),
53
54    /// The broker answered `INVALID_TOPIC_EXCEPTION` for the topic. Kafka's
55    /// `InvalidTopicException`.
56    #[error("topic '{0}' is invalid")]
57    InvalidTopic(String),
58
59    /// [`WakeupHandle::wakeup`](crate::WakeupHandle::wakeup) woke the call.
60    /// Kafka's `WakeupException`.
61    #[error("the consumer was woken up")]
62    Wakeup,
63
64    /// The call needs a group, and the consumer has no group id. Kafka's
65    /// `InvalidGroupIdException`.
66    #[error(
67        "To use the group management or offset commit APIs, you must provide a valid group.id in the consumer configuration."
68    )]
69    InvalidGroupId,
70
71    /// A call did not complete before its timeout. Kafka's
72    /// `TimeoutException`.
73    ///
74    /// `cause` is the last retriable error when a retry loop ran out of time,
75    /// as Kafka's `CommitRequestManager.maybeWrapAsTimeoutException` wraps it.
76    /// It is `None` when the call ran out of time while it waited.
77    #[error("timeout: {message}")]
78    Timeout {
79        message: String,
80        #[source]
81        cause: Option<Box<Self>>,
82    },
83
84    #[error("invalid seek offset {0}: must be non-negative")]
85    InvalidOffset(i64),
86
87    /// The call names a partition that the consumer does not own. Kafka's
88    /// `SubscriptionState.assignedState` raises `IllegalStateException`.
89    #[error("no current assignment for partition {topic}-{partition}")]
90    NoCurrentAssignment { topic: String, partition: i32 },
91
92    /// An asynchronous commit failed with a retriable error. A later commit
93    /// can succeed. Kafka's consumer gives `RetriableCommitFailedException` to
94    /// the `OffsetCommitCallback`.
95    #[error("offset commit failed with a retriable exception: {0}")]
96    RetriableCommitFailed(Box<Self>),
97
98    #[error("commit conflict: rejoined since this poll")]
99    CommitInvalid,
100
101    #[error("coordinator unavailable")]
102    CoordinatorUnavailable,
103
104    #[error(
105        "log truncation detected on {topic}-{partition}: fetch offset {fetch_offset} is past the leader's log; safe offset {safe_offset}"
106    )]
107    LogTruncation {
108        topic: String,
109        partition: i32,
110        fetch_offset: i64,
111        safe_offset: i64,
112    },
113
114    #[error("broker error_code {0}")]
115    Server(i16),
116
117    /// The coordinator answered `GROUP_AUTHORIZATION_FAILED` (30) for the
118    /// group. Kafka's consumer raises `GroupAuthorizationException`.
119    #[error("not authorized to access group: {0}")]
120    GroupAuthorizationFailed(String),
121
122    /// The coordinator answered `TOPIC_AUTHORIZATION_FAILED` (29) for
123    /// partitions of these topics. Kafka's consumer raises
124    /// `TopicAuthorizationException`.
125    #[error(
126        "not authorized to access topics: [{}]",
127        .0.iter().map(String::as_str).collect::<Vec<_>>().join(", ")
128    )]
129    TopicAuthorizationFailed(BTreeSet<String>),
130
131    /// An `OffsetFetch` response carried an error code that Kafka's consumer
132    /// does not retry.
133    #[error("unexpected error in offset fetch response: error_code {0}")]
134    OffsetFetchFailed(i16),
135
136    /// The coordinator answered `FENCED_INSTANCE_ID` (82) for this
137    /// `group.instance.id`, because another consumer joined with the same
138    /// id. `poll` returns this error once. The next `poll` joins the group
139    /// again. Kafka's consumer raises `FencedInstanceIdException`.
140    #[error(
141        "fenced group.instance.id {0}: another consumer with the same group.instance.id joined the group"
142    )]
143    FencedInstanceId(String),
144
145    /// A commit found that the consumer is not part of an active group:
146    /// a fatal coordinator error removed the member and no `poll` joined the
147    /// group again yet, or the coordinator task stopped. Kafka's consumer raises
148    /// `CommitFailedException`.
149    #[error(
150        "offset commit failed: the consumer is not part of an active group; it is likely that the consumer was kicked out of the group"
151    )]
152    CommitFailed,
153
154    /// The coordinator answered `ILLEGAL_GENERATION` (22), `UNKNOWN_MEMBER_ID`
155    /// (25) or `FENCED_INSTANCE_ID` (82) while the group was still
156    /// `PREPARING_REBALANCE`, or `REBALANCE_IN_PROGRESS` (27). Kafka's
157    /// `ConsumerCoordinator.OffsetCommitResponseHandler` raises
158    /// `RebalanceInProgressException` for these, and `commitOffsetsSync` does
159    /// not retry: the offsets are not committed, and the caller must call
160    /// `poll()` to finish the rebalance before it commits again.
161    #[error(
162        "offset commit cannot be completed since the consumer is undergoing a rebalance for group {0}: call poll() and retry"
163    )]
164    RebalanceInProgress(String),
165
166    /// A partition has no committed offset and `auto.offset.reset = none`.
167    /// Kafka never auto-resets under `none`
168    /// (`SubscriptionState.resetInitializingPositions`); it raises
169    /// `NoOffsetForPartitionException` from `poll()` and `position()` instead.
170    #[error(
171        "undefined offset with no reset policy for partitions: [{}]",
172        .0.iter().map(|(topic, partition)| format!("{topic}-{partition}")).collect::<Vec<_>>().join(", ")
173    )]
174    NoOffsetForPartition(BTreeSet<(String, i32)>),
175}
176
177impl ConsumerError {
178    /// A [`ConsumerError::Timeout`] with no cause.
179    pub(crate) fn timeout(message: impl Into<String>) -> Self {
180        Self::Timeout {
181            message: message.into(),
182            cause: None,
183        }
184    }
185
186    /// Whether this is an `OffsetFetch` error that Kafka's consumer does not
187    /// retry, so the application must see it.
188    pub(crate) fn is_fatal_offset_fetch_error(&self) -> bool {
189        matches!(
190            self,
191            Self::GroupAuthorizationFailed(_)
192                | Self::TopicAuthorizationFailed(_)
193                | Self::OffsetFetchFailed(_)
194        )
195    }
196}
197
198#[cfg(test)]
199mod tests {
200
201    use super::*;
202
203    #[test]
204    fn display_messages() {
205        for (_name, error, expected) in [
206            (
207                "not subscribed",
208                ConsumerError::NotSubscribed,
209                "not subscribed to any topic",
210            ),
211            ("server", ConsumerError::Server(25), "broker error_code 25"),
212            (
213                "invalid config",
214                ConsumerError::InvalidConfig("group_id required".into()),
215                "invalid configuration: group_id required",
216            ),
217            (
218                "group authorization failed",
219                ConsumerError::GroupAuthorizationFailed("workers".into()),
220                "not authorized to access group: workers",
221            ),
222            (
223                "topic authorization failed",
224                ConsumerError::TopicAuthorizationFailed(
225                    maplit::btreeset! {"orders".to_owned(), "payments".to_owned()},
226                ),
227                "not authorized to access topics: [orders, payments]",
228            ),
229            (
230                "offset fetch failed",
231                ConsumerError::OffsetFetchFailed(6),
232                "unexpected error in offset fetch response: error_code 6",
233            ),
234            (
235                "fenced instance id",
236                ConsumerError::FencedInstanceId("instance-a".into()),
237                "fenced group.instance.id instance-a: another consumer with the same group.instance.id joined the group",
238            ),
239            (
240                "commit failed",
241                ConsumerError::CommitFailed,
242                "offset commit failed: the consumer is not part of an active group; it is likely that the consumer was kicked out of the group",
243            ),
244            (
245                "rebalance in progress",
246                ConsumerError::RebalanceInProgress("group-a".into()),
247                "offset commit cannot be completed since the consumer is undergoing a rebalance for group group-a: call poll() and retry",
248            ),
249            (
250                "no current assignment",
251                ConsumerError::NoCurrentAssignment {
252                    topic: "t".into(),
253                    partition: 9,
254                },
255                "no current assignment for partition t-9",
256            ),
257            (
258                "retriable commit failed",
259                ConsumerError::RetriableCommitFailed(Box::new(ConsumerError::Server(15))),
260                "offset commit failed with a retriable exception: broker error_code 15",
261            ),
262            (
263                "log truncation",
264                ConsumerError::LogTruncation {
265                    topic: "t".into(),
266                    partition: 3,
267                    fetch_offset: 100,
268                    safe_offset: 42,
269                },
270                "log truncation detected on t-3: fetch offset 100 is past the leader's log; safe offset 42",
271            ),
272            (
273                "no offset for partition",
274                ConsumerError::NoOffsetForPartition(
275                    maplit::btreeset! {("orders".to_owned(), 0), ("orders".to_owned(), 1)},
276                ),
277                "undefined offset with no reset policy for partitions: [orders-0, orders-1]",
278            ),
279        ] {
280            assert2::assert!(error.to_string() == expected);
281        }
282    }
283}
krabka_client_consumer/error.rs

krabka_client_consumer/
error.rs
