Struct ReplicatedRecord 

pub struct ReplicatedRecord {
    pub topic: String,
    pub partition: i32,
    pub offset: i64,
    pub timestamp: i64,
    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 crabka_connect runtime.

Because crabka_connect::ConnectRecord has no topic/partition fields, all envelope metadata is bundled here alongside the raw payload.

Fields

topic: String

The source topic name.

partition: i32

The source partition index.

offset: i64

The source offset (0-based).

timestamp: i64

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, 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.

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 otherwise.

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 (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 once before the first poll, passing the position loaded from the durable checkpoint store on the target. Each position entry is keyed "<topic>-<partition>" → OffsetValue::Long(next_offset) (the value checkpoint wrote: last_consumed + 1). We decode each key back into (topic, partition) and hand the offset to the consumer’s seek.

The consumer holds each seek as pending and materialises it at the top of the first poll that sees the partition assigned — after the group’s post-assignment offset prime, but before any Fetch — so the sought offset is the one fetched. That makes restart resume from the last fully-committed record rather than re-reading the topic from offset 0: no record below the sought offset is re-delivered, and none above it is skipped (no data gap). Delivery remains at-least-once — a crash between a sink flush and the checkpoint save can re-deliver the in-flight batch, but never lose a record.

A malformed key (no -, or a non-integer partition/offset) is skipped with a warning rather than failing the restore: 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, sending LeaveGroup so a restarted replicator can rejoin the group immediately instead of waiting out the departed member’s session timeout.

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