Struct KafkaSource 

pub struct KafkaSource<T> { /* private fields */ }

Kafka source consumer for ingesting events

Implementations

impl<T> KafkaSource<T>

where T: DeserializeOwned + Send + Sync + 'static,

pub async fn new( config: KafkaSourceConfig, state_backend: Option<Arc<dyn StateBackend>>, ) -> Result<Self>

Create a new Kafka source consumer

Arguments

• config - Kafka source configuration
• state_backend - Optional state backend for offset management

Returns

• Result<Self> - New Kafka source instance

Example

use processor::kafka::{KafkaSource, KafkaSourceConfig};
use serde::{Deserialize, Serialize};

#[derive(Debug, Deserialize, Serialize)]
struct MyEvent {
    id: String,
    value: f64,
}

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let config = KafkaSourceConfig {
        brokers: "localhost:9092".to_string(),
        group_id: "my-consumer-group".to_string(),
        topics: vec!["my-topic".to_string()],
        ..Default::default()
    };

    let source = KafkaSource::<MyEvent>::new(config, None).await?;
    Ok(())
}

pub async fn subscribe(&self, topics: &[String]) -> Result<()>

Subscribe to topics

Arguments

• topics - Topics to subscribe to

Returns

• Result<()> - Success or error

pub async fn poll(&self) -> Result<Option<KafkaMessage<T>>>

Poll for a single message

Returns

• Result<Option<KafkaMessage<T>>> - Message or None if timeout/shutdown

Example

while let Some(msg) = source.poll().await? {
    println!("Received: {:?}", msg.payload);
    source.commit().await?;
}

pub async fn commit(&self) -> Result<()>

Commit current offsets

Returns

• Result<()> - Success or error

pub async fn seek(&self, topic: &str, partition: i32, offset: i64) -> Result<()>

Seek to a specific offset for a partition

Arguments

• topic - Topic name
• partition - Partition number
• offset - Offset to seek to

Returns

• Result<()> - Success or error

pub async fn metrics(&self) -> KafkaSourceMetrics

Get current consumption metrics

Returns

• KafkaSourceMetrics - Current metrics snapshot

pub async fn start(&self, tx: Sender<KafkaMessage<T>>) -> Result<()>

Start consuming messages and send them to a channel

This method runs in a loop, consuming messages and sending them to the provided channel. It handles automatic commits based on the configured commit strategy.

Arguments

• tx - Channel sender for messages

Returns

• Result<()> - Success or error

pub async fn close(&self) -> Result<()>

Stop the consumer gracefully

pub async fn get_lag(&self) -> Result<HashMap<i32, i64>>

Get current lag for all assigned partitions

This fetches watermarks from Kafka to calculate lag