Struct Checkpointer 

pub struct Checkpointer { /* private fields */ }

Aggregates batch resolutions into per-partition committable watermarks.

use etl_core::checkpoint::{AckStatus, Checkpointer};
use etl_core::record::PartitionId;

let mut cp = Checkpointer::new();
let p = PartitionId(0);
cp.begin_epoch(&[p], 1);
let mut issuer = cp.handle();

let ack = issuer.issue(p, 99); // batch covering offsets ..=99
drop(ack); // all records delivered

cp.drain();
assert_eq!(cp.take_watermarks(), vec![(p, 100)]);
assert_eq!(cp.take_watermarks(), vec![]); // idempotent until new acks

Implementations

impl Checkpointer

pub fn new() -> Checkpointer

A checkpointer with no assignment. Call begin_epoch when the source reports its first assignment.

pub fn handle(&self) -> AckIssuer

An issuer handle for a pipeline thread.

pub fn begin_epoch(&mut self, partitions: &[PartitionId], epoch: u32)

Start a new assignment epoch covering exactly partitions. Every rebalance bumps the epoch; in-flight batches from earlier epochs resolve as stale and their offsets are re-delivered by the source (at-least-once). Epochs must be strictly increasing.

Ordering contract: the runtime calls this before distributing the new assignment’s lanes to pipeline threads, so issuers observe the new epoch before issuing for it.

pub fn revoke(&mut self, partitions: &[PartitionId])

Drop tracking for revoked partitions mid-epoch (partial revocation or shutdown). Later resolutions for them are discarded as stale. A partition revoked this way can only return in a new epoch.

pub fn drain(&mut self) -> DrainStats

Apply all pending registrations and resolutions.

Two passes exploit the causal order guaranteed by AckIssuer (registration is sent before the batch’s AckRef exists): a resolution whose registration has not been drained yet is retried once after re-draining registrations; if it is still unknown, the driver is buggy and the resolution is counted and dropped.

pub fn take_watermarks(&mut self) -> Vec<(PartitionId, i64)>

Watermarks that advanced since the last call: (partition, committable offset) pairs ready for Source::commit. Empty when nothing moved — callers skip the commit entirely.

pub fn pending(&self, partition: PartitionId) -> usize

Unadvanced batches for one partition (backpressure trigger).

pub fn max_pending(&self) -> usize

The largest per-partition pending count.

pub fn stalled_partitions(&self) -> Vec<(PartitionId, Instant)>

Partitions whose watermark is permanently stalled behind a failed batch, with the stall start (health-probe input).

Trait Implementations

impl Debug for Checkpointer

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

Formats the value using the given formatter.

impl Default for Checkpointer

fn default() -> Checkpointer

Returns the “default value” for a type.