Struct OffsetTracker 

pub struct OffsetTracker { /* private fields */ }

Tracks consumed offsets per topic-partition.

Offsets stored are the last-consumed offset (not the next offset to fetch). When committing to Kafka, to_topic_partition_list() returns offset+1 (the next offset to consume) per Kafka convention.

Implementations

impl OffsetTracker

pub fn new() -> Self

Offsets start empty; populated by update() / from_checkpoint().

pub fn update(&mut self, topic: &str, partition: i32, offset: i64)

Updates the offset (monotonic: only advances forward).

pub fn update_arc(&mut self, topic: &Arc<str>, partition: i32, offset: i64)

Updates the offset using a pre-interned topic Arc (avoids allocation).

pub fn update_force(&mut self, topic: &str, partition: i32, offset: i64)

Unconditionally sets the offset for a topic-partition (used by restore).

pub fn get(&self, topic: &str, partition: i32) -> Option<i64>

Gets the last-consumed offset for a topic-partition.

pub fn partition_count(&self) -> usize

Returns the total number of tracked partitions across all topics.

pub fn to_checkpoint(&self) -> SourceCheckpoint

Converts tracked offsets to a SourceCheckpoint.

Key format: "{topic}-{partition}", value: offset as string.

pub fn from_checkpoint(cp: &SourceCheckpoint) -> Self

Restores offset state from a SourceCheckpoint.

Parses keys in "{topic}-{partition}" format. Logs warnings for unparseable entries rather than silently dropping them.

pub fn to_topic_partition_list(&self) -> TopicPartitionList

Builds an rdkafka [TopicPartitionList] for committing.

Per Kafka convention, committed offsets are next-to-fetch (offset+1).

pub fn to_seek_tpl(&self, assigned: &[(String, i32)]) -> TopicPartitionList

Builds a [TopicPartitionList] for seeking after a rebalance assign.

Only includes partitions where the tracker holds a known offset, using Offset::Offset(offset + 1) (next-to-fetch). Partitions NOT in the tracker are omitted — callers should let auto.offset.reset handle those.

pub fn retain_assigned(&mut self, assigned: &HashSet<(String, i32)>)

Removes partitions that are not in the assigned set.

Called after a rebalance revoke to purge offsets for partitions this consumer no longer owns. This prevents stale offsets from leaking into checkpoints and causing incorrect partition assignment on recovery.

pub fn to_checkpoint_filtered( &self, assigned: &HashSet<(String, i32)>, ) -> SourceCheckpoint

Builds a checkpoint containing only offsets for currently assigned partitions.

Non-mutating alternative to retain_assigned() + to_checkpoint(). When assigned is empty (either before first rebalance or after full revocation), returns an empty checkpoint — this is correct because there are no partitions this consumer owns.

pub fn to_topic_partition_list_filtered( &self, assigned: &HashSet<(String, i32)>, ) -> TopicPartitionList

Builds an rdkafka [TopicPartitionList] for only assigned partitions.

Like Self::to_topic_partition_list but filtered to the assigned set. When assigned is empty, returns an empty TPL.

pub fn clear(&mut self)

Clears all tracked offsets.

Trait Implementations

impl Clone for OffsetTracker

fn clone(&self) -> OffsetTracker

Returns a duplicate of the value.

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

Performs copy-assignment from source.

impl Debug for OffsetTracker

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

Formats the value using the given formatter.

impl Default for OffsetTracker

fn default() -> OffsetTracker

Returns the “default value” for a type.