Struct TargetSink 

pub struct TargetSink { /* private fields */ }

A Sink that applies residency filtering, topic renaming, and loop-guard logic before producing records to the target cluster.

On each flush it drains pending produce acknowledgements, sets the downstream offset on each OffsetSync, and writes those syncs back to the target’s offset-syncs topic (MM2-compatible).

Implementations

impl TargetSink

pub async fn start(params: SinkParams) -> Result<Self, ConnectError>

Build a TargetSink, ensure the offset-syncs topic exists, and connect the producer to the target cluster.

Errors

Returns ConnectError::Backend if the producer cannot connect or the offset-syncs topic cannot be created.

pub fn drain_offset_syncs(&mut self) -> Vec<OffsetSync>

Return and clear all completed OffsetSync records accumulated since the last call (or since construction).

Trait Implementations

impl Sink<(), ReplicatedRecord> for TargetSink

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

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

Accept a batch of replicated records, applying filtering and loop-guard logic, then enqueue produce calls for the accepted records.

Records are dropped (not buffered) when:

• value is None (tombstone or no payload).
• Identity-naming loop-guard fires: the record’s __crabka_origin header matches our own source_alias.
• Residency gate blocks the topic for the target’s zones.

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

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

Await all pending produce acks, write offset-syncs to the target, and flush the producer.

fn supports_transactions(&self) -> bool

Whether this sink writes atomically and so can take part in exactly-once delivery. When false (the default) the runtime drives it at-least-once and never calls begin / abort.

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

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

Open a transaction enclosing the puts that follow, up to the next commit or abort. Called by the runtime only when supports_transactions is true and only before a non-empty batch. Default: no-op.

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

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

Atomically commit the records written since begin, making them durable. The default (no transaction support) delegates to flush.

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

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

Discard the records written since begin without making them durable. Called by the runtime to roll back a failed interval. Default: no-op.

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

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

Release any resources held by the sink (connections, buffers). The default is a no-op. After close, no further records are written.