llm_optimizer_processor/kafka/
mod.rs

1//! Kafka integration for stream processing
2//!
3//! This module provides Kafka source and sink implementations for ingesting
4//! and producing events in the stream processing pipeline.
5//!
6//! # Features
7//!
8//! ## KafkaSource (Consumer)
9//! - Robust Kafka consumer with metrics and offset management
10//! - Automatic reconnection on errors
11//! - Configurable commit strategies (auto, manual, periodic, on-checkpoint)
12//! - Multiple partition support with rebalancing
13//! - Graceful shutdown handling
14//! - Message deserialization with error handling
15//! - Comprehensive metrics tracking (throughput, lag, etc.)
16//!
17//! ## KafkaSink (Producer)
18//! - High-performance Kafka producer with batching and compression
19//! - Idempotent producer support
20//! - Transactional support for exactly-once semantics
21//! - Circuit breaker for fault tolerance
22//! - Retry logic with exponential backoff
23//! - Configurable partitioning strategies
24//! - Comprehensive metrics tracking (latency, throughput, errors)
25//!
26//! ## Configuration
27//! - Comprehensive configuration types for source and sink
28//! - Support for all rdkafka settings
29//! - Security configuration (SASL, SSL)
30//! - Performance tuning options
31//!
32//! ## Error Handling
33//! - Kafka-specific error types with proper context
34//! - Error severity levels for monitoring
35//! - Retryable vs fatal error classification
36//!
37//! # Example: Source Usage
38//!
39//! ```rust,no_run
40//! use processor::kafka::{KafkaSource, KafkaSourceConfig, CommitStrategy};
41//! use serde::{Deserialize, Serialize};
42//! use tokio::sync::mpsc;
43//!
44//! #[derive(Clone, Debug, Serialize, Deserialize)]
45//! struct MyEvent {
46//!     value: f64,
47//!     timestamp: i64,
48//! }
49//!
50//! # async fn example() -> anyhow::Result<()> {
51//! // Configure Kafka source
52//! let config = KafkaSourceConfig {
53//!     brokers: "localhost:9092".to_string(),
54//!     group_id: "my-consumer-group".to_string(),
55//!     topics: vec!["my-topic".to_string()],
56//!     commit_strategy: CommitStrategy::Periodic,
57//!     commit_interval_secs: Some(5),
58//!     ..Default::default()
59//! };
60//!
61//! // Create Kafka source
62//! let source = KafkaSource::<MyEvent>::new(config, None).await?;
63//!
64//! // Create channel for messages
65//! let (tx, mut rx) = mpsc::channel(1000);
66//!
67//! // Start consuming in background
68//! let source_handle = tokio::spawn(async move {
69//!     source.start(tx).await
70//! });
71//!
72//! // Process messages
73//! while let Some(msg) = rx.recv().await {
74//!     println!("Received: {:?}", msg.payload);
75//! }
76//!
77//! # Ok(())
78//! # }
79//! ```
80//!
81//! # Example: Sink Usage
82//!
83//! ```rust,no_run
84//! use processor::kafka::{KafkaSink, KafkaSinkConfig, KafkaMessage};
85//! use serde::{Deserialize, Serialize};
86//!
87//! #[derive(Clone, Debug, Serialize, Deserialize)]
88//! struct MetricResult {
89//!     avg_latency: f64,
90//!     count: u64,
91//! }
92//!
93//! # async fn example() -> anyhow::Result<()> {
94//! let config = KafkaSinkConfig {
95//!     brokers: "localhost:9092".to_string(),
96//!     topic: "results".to_string(),
97//!     enable_idempotence: true,
98//!     ..Default::default()
99//! };
100//!
101//! let sink = KafkaSink::<MetricResult>::new(config).await?;
102//!
103//! let message = KafkaMessage::new(MetricResult {
104//!     avg_latency: 123.45,
105//!     count: 1000,
106//! }).with_key("service-a".to_string());
107//!
108//! sink.send(message).await?;
109//! sink.close().await?;
110//! # Ok(())
111//! # }
112//! ```
113
114pub mod config;
115pub mod error;
116pub mod integration;
117pub mod offset_manager;
118pub mod sink;
119pub mod source;
120
121#[cfg(test)]
122mod tests;
123
124// Re-export config types
125pub use config::{
126    CompressionType, KafkaSinkConfig as DetailedKafkaSinkConfig,
127    KafkaSourceConfig as DetailedKafkaSourceConfig, SaslMechanism, SecurityProtocol, SslConfig,
128};
129
130// Re-export error types
131pub use error::{ErrorSeverity, KafkaError, Result as KafkaResult};
132
133// Re-export offset manager types
134pub use offset_manager::{
135    InMemoryOffsetStore, OffsetCommitStrategy, OffsetInfo, OffsetManager, OffsetResetStrategy,
136    OffsetStats, OffsetStore, StateBackendOffsetStore, TopicPartition,
137};
138
139// Re-export source types
140pub use source::{
141    CommitStrategy, KafkaMessage as SourceMessage, KafkaSource, KafkaSourceConfig,
142    KafkaSourceMetrics,
143};
144
145// Re-export sink types
146pub use sink::{
147    BincodeSerializer, DeliveryGuarantee, JsonSerializer, KafkaMessage as SinkMessage,
148    KafkaSink, KafkaSinkConfig, MessageSerializer, PartitionStrategy, SinkMetrics,
149};
150
151// Re-export integration types
152pub use integration::{
153    KafkaStreamConfig, KafkaStreamPipeline, KafkaStreamSink, KafkaStreamSource,
154    KafkaStreamStats, OffsetManager as StreamOffsetManager,
155};
llm_optimizer_processor/kafka/mod.rs

llm_optimizer_processor/kafka/
mod.rs
