Struct TargetSink 

pub struct TargetSink { /* private fields */ }

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

At-least-once flows drain pending acknowledgements and write MM2-compatible offset-syncs on flush. Exactly-once flows put target data, offset-syncs, and the source checkpoint in one producer transaction and make all three visible together on commit.

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 async fn start_with_policy( params: SinkParams, client_resource_policy: ClientResourcePolicy, ) -> Result<Self, ConnectError>

Start with the deployment’s client resource policy.

Errors

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

pub fn set_offset_sync_observation_capacity(&mut self, capacity: usize)

Retain at most capacity recent completed offset-syncs for observation.

Observation is disabled by default. A zero capacity disables it and releases retained records. Older observations are dropped when the capacity is exceeded; durable sync publication is unaffected.

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

Return and clear retained OffsetSync observations, oldest first.

Returns an empty vector unless observation was explicitly enabled with Self::set_offset_sync_observation_capacity. Only successful flushes and committed transactions become observable.

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, apply filtering and loop-guard logic, then enqueue produce calls for the accepted records.

The sink drops a record, and does not buffer it, when:

• value is None, that is, a tombstone or no payload.
• The identity-naming loop-guard fires: the __krabka_origin header of the record matches our own source_alias.
• The residency gate blocks the topic for the zones of the target.

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 or abort.

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

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

Open a transaction that encloses the puts that follow, up to the next commit or abort. The runtime calls this 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 Self: 'async_trait, 'life0: 'async_trait,

Atomically commit the records written since begin and make them durable. The default has no transaction support and delegates to flush.

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

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

Discard the records written since begin and do not make them durable. The runtime calls this 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 Self: 'async_trait, 'life0: 'async_trait,

Release any resources that the sink holds, such as connections and buffers. The default is a no-op. After close, the sink writes no further records.