Struct SourceConsumer 

pub struct SourceConsumer { /* private fields */ }

A [Source] implementation backed by a Kafka consumer on the source cluster.

This type wraps a Consumer and translates each krabka_client_consumer::ConsumerRecord into a ReplicatedRecord. That record carries the full envelope of topic, partition, offset, timestamp, and headers next to the raw payload. The connect runtime receives the whole ReplicatedRecord value, including the source coordinates that offset-sync generation needs.

Implementations

impl SourceConsumer

pub async fn start( bootstrap: &str, group_id: &str, topics: &[String], security: Option<ClientSecurity>, ) -> Result<Self, ConnectError>

Build and start a SourceConsumer that subscribes to topics on the cluster at bootstrap and joins group_id.

Offsets reset to earliest, because the group has no previously committed offset. Pass security when the source cluster needs authentication or TLS.

Errors

Returns [ConnectError::Backend] if the consumer cannot join the group.

pub async fn start_with_policy( bootstrap: &str, group_id: &str, topics: &[String], security: Option<ClientSecurity>, client_resource_policy: ClientResourcePolicy, ) -> Result<Self, ConnectError>

Build with the deployment’s client resource policy.

Errors

Returns a [ConnectError] if the consumer cannot connect.

Trait Implementations

impl Source<(), ReplicatedRecord> for SourceConsumer

fn poll<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<Option<ConnectRecord<(), ReplicatedRecord>>, ConnectError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Poll the source cluster for the next record.

Returns Ok(None) when the consumer is momentarily caught up. The runtime should back off and retry. Returns Ok(Some(_)) with the next ReplicatedRecord in every other case.

Errors

Returns [ConnectError::Backend] if the underlying consumer poll fails.

fn checkpoint(&self) -> Option<SourceOffset>

Snapshot the current read positions for all partitions seen so far.

Returns None before the first successful poll, when there is nothing to commit yet.

fn seek<'life0, 'async_trait>( &'life0 mut self, offset: SourceOffset, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Restore the read position from a previously-checkpointed [SourceOffset].

The runtime calls this method once before the first poll and passes the position that it loaded from the durable checkpoint store on the target. Each position entry maps the key "<topic>-<partition>" to [OffsetValue::Long](next_offset), the value that checkpoint wrote as last_consumed + 1. This method decodes each key back into (topic, partition) and hands the offset to the seek of the consumer.

The consumer holds each seek as pending. It materialises the seek at the top of the first poll that sees the partition assigned, after the post-assignment offset prime of the group but before any Fetch. The sought offset is therefore the offset that the consumer fetches. A restart then resumes from the last fully-committed record and does not re-read the topic from offset 0. No record below the sought offset is re-delivered and no record above it is skipped, so there is no data gap. An at-least-once flow can re-deliver a batch after a crash between the sink flush and checkpoint save. An exactly-once flow writes this same position inside the target producer transaction, so output and recovery position become visible together.

This method skips a malformed key with a warning and does not fail the restore. A malformed key has no -, or a non-integer partition or offset. One corrupt entry must not strand recovery for the partitions that decoded cleanly.

Errors

Returns [ConnectError::Backend] if the consumer is already closed.

fn close<'life0, 'async_trait>( &'life0 mut self, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectError>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait,

Close the underlying consumer. The close sends LeaveGroup, so a restarted replicator can rejoin the group immediately and does not wait out the session timeout of the departed member.

Errors

Returns [ConnectError::Backend] if the consumer fails to close cleanly.

fn acknowledge<'life0, 'life1, 'async_trait>( &'life0 mut self, _offset: &'life1 SourceOffset, ) -> Pin<Box<dyn Future<Output = Result<(), ConnectError>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, Self: 'async_trait,

Acknowledge that offset is durable end-to-end.