Struct ReplicatedRecord 

pub struct ReplicatedRecord {
    pub topic: String,
    pub partition: PartitionIndex,
    pub offset: Offset,
    pub timestamp: Timestamp,
    pub key: Option<Bytes>,
    pub value: Option<Bytes>,
    pub headers: Vec<(String, Option<Bytes>)>,
}

One record from the source cluster, carried as the connect-value type V through the [krabka_connect] runtime.

Replication keeps its full source envelope here. The specialized target sink consumes the offset and the provenance metadata, in addition to the generic [krabka_connect::ConnectRecord] routing fields.

Fields

topic: String

The source topic name.

partition: PartitionIndex

The source partition index.

offset: Offset

The source offset (0-based).

timestamp: Timestamp

Record timestamp in epoch milliseconds.

key: Option<Bytes>

Record key, or None for a null key.

value: Option<Bytes>

Record value, or None for a tombstone.

headers: Vec<(String, Option<Bytes>)>

Per-record headers in declaration order.

Trait Implementations

impl Clone for ReplicatedRecord

fn clone(&self) -> ReplicatedRecord

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for ReplicatedRecord

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Eq for ReplicatedRecord

impl PartialEq for ReplicatedRecord

fn eq(&self, other: &ReplicatedRecord) -> bool

Tests for self and other values to be equal, and is used by ==.

fn ne(&self, other: &Rhs) -> bool

Tests for !=. The default implementation is almost always sufficient, and should not be overridden without very good reason.

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.

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.

impl StructuralPartialEq for ReplicatedRecord