Module kafka 

Kafka integration for stream processing

This module provides Kafka source and sink implementations for ingesting and producing events in the stream processing pipeline.

Features

KafkaSource (Consumer)

• Robust Kafka consumer with metrics and offset management
• Automatic reconnection on errors
• Configurable commit strategies (auto, manual, periodic, on-checkpoint)
• Multiple partition support with rebalancing
• Graceful shutdown handling
• Message deserialization with error handling
• Comprehensive metrics tracking (throughput, lag, etc.)

KafkaSink (Producer)

• High-performance Kafka producer with batching and compression
• Idempotent producer support
• Transactional support for exactly-once semantics
• Circuit breaker for fault tolerance
• Retry logic with exponential backoff
• Configurable partitioning strategies
• Comprehensive metrics tracking (latency, throughput, errors)

Configuration

• Comprehensive configuration types for source and sink
• Support for all rdkafka settings
• Security configuration (SASL, SSL)
• Performance tuning options

Error Handling

• Kafka-specific error types with proper context
• Error severity levels for monitoring
• Retryable vs fatal error classification

Example: Source Usage

use processor::kafka::{KafkaSource, KafkaSourceConfig, CommitStrategy};
use serde::{Deserialize, Serialize};
use tokio::sync::mpsc;

#[derive(Clone, Debug, Serialize, Deserialize)]
struct MyEvent {
    value: f64,
    timestamp: i64,
}

// Configure Kafka source
let config = KafkaSourceConfig {
    brokers: "localhost:9092".to_string(),
    group_id: "my-consumer-group".to_string(),
    topics: vec!["my-topic".to_string()],
    commit_strategy: CommitStrategy::Periodic,
    commit_interval_secs: Some(5),
    ..Default::default()
};

// Create Kafka source
let source = KafkaSource::<MyEvent>::new(config, None).await?;

// Create channel for messages
let (tx, mut rx) = mpsc::channel(1000);

// Start consuming in background
let source_handle = tokio::spawn(async move {
    source.start(tx).await
});

// Process messages
while let Some(msg) = rx.recv().await {
    println!("Received: {:?}", msg.payload);
}

Example: Sink Usage

use processor::kafka::{KafkaSink, KafkaSinkConfig, KafkaMessage};
use serde::{Deserialize, Serialize};

#[derive(Clone, Debug, Serialize, Deserialize)]
struct MetricResult {
    avg_latency: f64,
    count: u64,
}

let config = KafkaSinkConfig {
    brokers: "localhost:9092".to_string(),
    topic: "results".to_string(),
    enable_idempotence: true,
    ..Default::default()
};

let sink = KafkaSink::<MetricResult>::new(config).await?;

let message = KafkaMessage::new(MetricResult {
    avg_latency: 123.45,
    count: 1000,
}).with_key("service-a".to_string());

sink.send(message).await?;
sink.close().await?;

Re-exports

pub use config::CompressionType;

pub use config::KafkaSinkConfig as DetailedKafkaSinkConfig;

pub use config::KafkaSourceConfig as DetailedKafkaSourceConfig;

pub use config::SaslMechanism;

pub use config::SecurityProtocol;

pub use config::SslConfig;

pub use error::ErrorSeverity;

pub use error::KafkaError;

pub use error::Result as KafkaResult;

pub use offset_manager::InMemoryOffsetStore;

pub use offset_manager::OffsetCommitStrategy;

pub use offset_manager::OffsetInfo;

pub use offset_manager::OffsetManager;

pub use offset_manager::OffsetResetStrategy;

pub use offset_manager::OffsetStats;

pub use offset_manager::OffsetStore;

pub use offset_manager::StateBackendOffsetStore;

pub use offset_manager::TopicPartition;

pub use source::CommitStrategy;

pub use source::KafkaMessage as SourceMessage;

pub use source::KafkaSource;

pub use source::KafkaSourceConfig;

pub use source::KafkaSourceMetrics;

pub use sink::BincodeSerializer;

pub use sink::DeliveryGuarantee;

pub use sink::JsonSerializer;

pub use sink::KafkaMessage as SinkMessage;

pub use sink::KafkaSink;

pub use sink::KafkaSinkConfig;

pub use sink::MessageSerializer;

pub use sink::PartitionStrategy;

pub use sink::SinkMetrics;

pub use integration::KafkaStreamConfig;

pub use integration::KafkaStreamPipeline;

pub use integration::KafkaStreamSink;

pub use integration::KafkaStreamSource;

pub use integration::KafkaStreamStats;

pub use integration::OffsetManager as StreamOffsetManager;

Modules

config

Kafka configuration structures for sources and sinks.

error

Kafka-specific error types and error handling.

integration

Kafka integration for stream processing pipeline

offset_manager

Kafka offset management and checkpointing

sink

Kafka Sink for publishing processed results

source

Kafka source consumer implementation