use std::{fmt, io};
pub type KafkaClientResult<T> = Result<T, KafkaClientError>;
#[derive(Debug)]
pub enum KafkaClientError {
InvalidConfig(String),
Protocol(String),
Unsupported(String),
Security(String),
Tls {
broker: String,
message: String,
},
Sasl {
broker: String,
mechanism: String,
message: String,
},
Decompression {
codec: &'static str,
message: String,
},
DecompressionLimitExceeded {
codec: &'static str,
limit: usize,
},
Broker {
operation: &'static str,
code: i16,
message: String,
},
AssignmentLost {
topic: String,
partition: i32,
},
Io(io::Error),
ChannelClosed,
}
impl KafkaClientError {
pub(crate) fn protocol(message: impl Into<String>) -> Self {
Self::Protocol(message.into())
}
pub(crate) fn unsupported(message: impl Into<String>) -> Self {
Self::Unsupported(message.into())
}
pub(crate) fn security(message: impl Into<String>) -> Self {
Self::Security(message.into())
}
pub(crate) fn decompression(codec: &'static str, message: impl Into<String>) -> Self {
Self::Decompression {
codec,
message: message.into(),
}
}
pub(crate) fn decompression_limit(codec: &'static str, limit: usize) -> Self {
Self::DecompressionLimitExceeded { codec, limit }
}
pub(crate) fn broker(operation: &'static str, code: i16) -> Self {
Self::Broker {
operation,
code,
message: kafka_error_message(code).to_owned(),
}
}
pub(crate) fn retriable(&self) -> bool {
match self {
Self::Io(_) | Self::Tls { .. } | Self::ChannelClosed => true,
Self::Broker { code, .. } => matches!(
*code,
3 | 5 | 6 | 7 | 9 | 14 | 15 | 16 | 19 | 41 | 56 | 57 | 68 | 87 | 100
),
_ => false,
}
}
}
impl fmt::Display for KafkaClientError {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
match self {
Self::InvalidConfig(message) => write!(f, "invalid Kafka client config: {message}"),
Self::Protocol(message) => write!(f, "Kafka protocol error: {message}"),
Self::Unsupported(message) => write!(f, "unsupported Kafka feature: {message}"),
Self::Security(message) => write!(f, "Kafka security error: {message}"),
Self::Tls { broker, message } => {
write!(
f,
"Kafka TLS handshake with broker {broker} failed: {message}"
)
}
Self::Sasl {
broker,
mechanism,
message,
} => write!(
f,
"Kafka SASL {mechanism} authentication with broker {broker} failed: {message}"
),
Self::Decompression { codec, message } => {
write!(
f,
"Kafka {codec} record-batch decompression failed: {message}"
)
}
Self::DecompressionLimitExceeded { codec, limit } => write!(
f,
"Kafka {codec} record batch exceeds the {limit}-byte decompression limit"
),
Self::Broker {
operation,
code,
message,
} => write!(
f,
"Kafka broker error during {operation}: {message} ({code})"
),
Self::AssignmentLost { topic, partition } => write!(
f,
"Kafka assignment was lost for {topic}:{partition}; refusing to commit offset"
),
Self::Io(error) => write!(f, "Kafka IO error: {error}"),
Self::ChannelClosed => write!(f, "Kafka connection task closed"),
}
}
}
impl std::error::Error for KafkaClientError {
fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
match self {
Self::Io(error) => Some(error),
_ => None,
}
}
}
impl From<io::Error> for KafkaClientError {
fn from(error: io::Error) -> Self {
Self::Io(error)
}
}
pub(crate) fn kafka_error_message(code: i16) -> &'static str {
match code {
0 => "none",
1 => "offset out of range",
2 => "corrupt message",
3 => "unknown topic or partition",
5 => "leader not available",
6 => "not leader for partition",
7 => "request timed out",
9 => "replica not available",
14 => "coordinator load in progress",
15 => "coordinator not available",
16 => "not coordinator",
19 => "not enough replicas",
22 => "illegal generation",
25 => "unknown member id",
27 => "rebalance in progress",
30 => "group authorization failed",
31 => "cluster authorization failed",
35 => "unsupported version",
33 => "unsupported SASL mechanism",
34 => "illegal SASL state",
58 => "SASL authentication failed",
41 => "not controller",
45 => "out of order sequence number",
46 => "duplicate sequence number",
47 => "invalid producer epoch",
56 => "kafka storage error",
57 => "log directory not found",
59 => "unknown producer id",
68 => "unstable offset commit",
79 => "member id required",
87 => "offset moved to tiered storage",
100 => "unknown topic id",
_ => "unknown broker error",
}
}