Enum KafkaError 

pub enum KafkaError {
    Connection {
        message: String,
        broker: String,
    },
    Consumption {
        message: String,
        topic: String,
        partition: Option<i32>,
        offset: Option<i64>,
    },
    Production {
        message: String,
        topic: String,
        partition: Option<i32>,
    },
    Serialization {
        message: String,
        context: Option<String>,
    },
    Deserialization {
        message: String,
        context: Option<String>,
        payload_sample: Option<Vec<u8>>,
    },
    InvalidConfiguration {
        message: String,
        field: Option<String>,
    },
    OffsetCommit {
        message: String,
        topic: String,
        partition: i32,
        offset: i64,
    },
    GroupCoordination {
        message: String,
        group_id: String,
    },
    PartitionAssignment {
        message: String,
        topics: Vec<String>,
    },
    AuthError {
        message: String,
        protocol: String,
    },
    IoError {
        message: String,
        source: Option<Error>,
    },
    Timeout {
        operation: String,
        timeout_ms: u64,
    },
    TopicMetadata {
        message: String,
        topic: String,
    },
    PartitionNotFound {
        topic: String,
        partition: i32,
    },
    ResourceExhausted {
        message: String,
        resource: String,
    },
    KafkaClient {
        message: String,
        code: Option<i32>,
    },
    Other(String),
}

Comprehensive error type for Kafka operations.

This enum covers all possible error scenarios when working with Kafka, providing detailed context for debugging and error handling.

Examples

use processor::kafka::error::{KafkaError, Result};

fn connect_to_kafka() -> Result<()> {
    // Simulate connection error
    Err(KafkaError::Connection {
        message: "Failed to connect to broker".to_string(),
        broker: "localhost:9092".to_string(),
    })
}

match connect_to_kafka() {
    Ok(_) => println!("Connected successfully"),
    Err(KafkaError::Connection { message, broker }) => {
        eprintln!("Connection error to {}: {}", broker, message);
    }
    Err(e) => eprintln!("Other error: {}", e),
}

Variants

Connection

Error establishing connection to Kafka broker.

Fields

message: String

Error message describing the connection failure.

broker: String

Broker address that failed to connect.

Consumption

Error during message consumption.

Fields

message: String

Error message describing the consumption failure.

topic: String

Topic from which consumption failed.

partition: Option<i32>

Partition number if available.

offset: Option<i64>

Offset if available.

Production

Error during message production.

Fields

message: String

Error message describing the production failure.

topic: String

Topic to which production failed.

partition: Option<i32>

Partition number if available.

Serialization

Error during serialization of messages.

Fields

message: String

Error message describing the serialization failure.

context: Option<String>

Optional context about what was being serialized.

Deserialization

Error during deserialization of messages.

Fields

message: String

Error message describing the deserialization failure.

context: Option<String>

Optional context about what was being deserialized.

payload_sample: Option<Vec<u8>>

Raw bytes that failed to deserialize (truncated for display).

InvalidConfiguration

Configuration validation error.

Fields

message: String

Error message describing the configuration issue.

field: Option<String>

Field name that has invalid configuration.

OffsetCommit

Error committing offsets.

Fields

message: String

Error message describing the commit failure.

topic: String

Topic for which offset commit failed.

partition: i32

Partition for which offset commit failed.

offset: i64

Offset that failed to commit.

GroupCoordination

Error during consumer group coordination.

Fields

message: String

Error message describing the coordination failure.

group_id: String

Consumer group ID.

PartitionAssignment

Error during partition assignment.

Fields

message: String

Error message describing the assignment failure.

topics: Vec<String>

Topics involved in the assignment.

AuthError

Authentication or authorization error.

Fields

message: String

Error message describing the auth failure.

protocol: String

Security protocol being used.

IoError

Network or I/O error.

Fields

message: String

Error message describing the I/O failure.

source: Option<Error>

Optional underlying I/O error.

Timeout

Timeout error during Kafka operations.

Fields

operation: String

Description of the operation that timed out.

timeout_ms: u64

Timeout duration in milliseconds.

TopicMetadata

Error related to topic metadata.

Fields

message: String

Error message describing the metadata failure.

topic: String

Topic name.

PartitionNotFound

Error when a topic partition doesn’t exist.

Fields

topic: String

Topic name.

partition: i32

Partition number.

ResourceExhausted

Error when reaching resource limits.

Fields

message: String

Error message describing the resource exhaustion.

resource: String

Resource type (e.g., “buffer”, “connections”).

KafkaClient

Error from the underlying rdkafka library.

Fields

message: String

Error message from rdkafka.

code: Option<i32>

rdkafka error code.

Other(String)

Generic error for cases not covered by specific variants.

Implementations

impl KafkaError

pub fn connection(message: impl Into<String>, broker: impl Into<String>) -> Self

Create a connection error.

pub fn consumption( message: impl Into<String>, topic: impl Into<String>, partition: Option<i32>, offset: Option<i64>, ) -> Self

Create a consumption error.

pub fn production( message: impl Into<String>, topic: impl Into<String>, partition: Option<i32>, ) -> Self

Create a production error.

pub fn serialization( message: impl Into<String>, context: Option<String>, ) -> Self

Create a serialization error.

pub fn deserialization( message: impl Into<String>, context: Option<String>, payload_sample: Option<Vec<u8>>, ) -> Self

Create a deserialization error.

pub fn invalid_config(message: impl Into<String>, field: Option<String>) -> Self

Create an invalid configuration error.

pub fn offset_commit( message: impl Into<String>, topic: impl Into<String>, partition: i32, offset: i64, ) -> Self

Create an offset commit error.

pub fn group_coordination( message: impl Into<String>, group_id: impl Into<String>, ) -> Self

Create a group coordination error.

pub fn partition_assignment( message: impl Into<String>, topics: Vec<String>, ) -> Self

Create a partition assignment error.

pub fn auth_error( message: impl Into<String>, protocol: impl Into<String>, ) -> Self

Create an authentication/authorization error.

pub fn timeout(operation: impl Into<String>, timeout_ms: u64) -> Self

Create a timeout error.

pub fn topic_metadata( message: impl Into<String>, topic: impl Into<String>, ) -> Self

Create a topic metadata error.

pub fn partition_not_found(topic: impl Into<String>, partition: i32) -> Self

Create a partition not found error.

pub fn resource_exhausted( message: impl Into<String>, resource: impl Into<String>, ) -> Self

Create a resource exhausted error.

pub fn kafka_client(message: impl Into<String>, code: Option<i32>) -> Self

Create a Kafka client error.

pub fn is_retryable(&self) -> bool

Check if the error is retryable.

Returns true if the operation that caused this error can be retried.

pub fn is_fatal(&self) -> bool

Check if the error is fatal and requires system shutdown.

Returns true if the error indicates an unrecoverable state.

pub fn severity(&self) -> ErrorSeverity

Get error severity level.

Returns a severity level suitable for logging and monitoring.

Trait Implementations

impl Debug for KafkaError

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Display for KafkaError

fn fmt(&self, __formatter: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Error for KafkaError

fn source(&self) -> Option<&(dyn Error + 'static)>

Returns the lower-level source of this error, if any.

fn description(&self) -> &str

👎Deprecated since 1.42.0:

use the Display impl or to_string()

fn cause(&self) -> Option<&dyn Error>

👎Deprecated since 1.33.0:

replaced by Error::source, which can support downcasting

fn provide<'a>(&'a self, request: &mut Request<'a>)

🔬This is a nightly-only experimental API. (error_generic_member_access)

Provides type-based access to context intended for error reports.

impl From<Box<ErrorKind>> for KafkaError

fn from(err: Error) -> Self

Converts to this type from the input type.

impl From<Error> for KafkaError

fn from(err: Error) -> Self

Converts to this type from the input type.

impl From<Error> for KafkaError

fn from(err: Error) -> Self

Converts to this type from the input type.

impl From<KafkaError> for KafkaError

fn from(err: KafkaError) -> Self

Converts to this type from the input type.