Struct OffsetManager 

pub struct OffsetManager<B: StateBackend, S: OffsetStore> { /* private fields */ }

Main offset manager

Coordinates offset tracking, committing, and recovery for Kafka consumers.

Features

• Per-partition offset tracking
• Watermark-based commit strategies
• Checkpoint integration
• Offset lag monitoring
• Automatic and manual commit modes

Implementations

impl<B: StateBackend, S: OffsetStore> OffsetManager<B, S>

pub fn new( consumer_group: String, backend: Arc<B>, store: Arc<S>, strategy: OffsetCommitStrategy, ) -> Self

Create a new offset manager

Arguments

• consumer_group - Consumer group ID
• backend - State backend for checkpointing
• store - Offset storage implementation
• strategy - Commit strategy

pub async fn update_offset( &mut self, topic: &str, partition: u32, offset: i64, ) -> StateResult<()>

Update offset for a partition

Arguments

• topic - Kafka topic
• partition - Partition number
• offset - New offset value

Returns

• Ok(()) - Offset updated
• Err(StateError) - Update failed

pub async fn update_offset_with_watermark( &mut self, topic: &str, partition: u32, offset: i64, high_watermark: i64, ) -> StateResult<()>

Update offset with high watermark

pub async fn get_offset( &self, topic: &str, partition: u32, ) -> StateResult<Option<i64>>

Get current offset for a partition

Returns the committed offset, or None if no offset has been committed.

pub async fn get_offset_info( &self, topic: &str, partition: u32, ) -> StateResult<Option<OffsetInfo>>

Get offset info (includes metadata and watermarks)

pub async fn commit(&mut self) -> StateResult<usize>

Commit all pending offsets

Returns

• Ok(committed_count) - Number of offsets committed
• Err(StateError) - Commit failed

pub async fn lag( &self, topic: &str, partition: u32, high_watermark: i64, ) -> StateResult<i64>

Calculate offset lag for a partition

Arguments

• topic - Kafka topic
• partition - Partition number
• high_watermark - Current high watermark (latest offset in partition)

Returns

• Ok(lag) - Offset lag (high_watermark - current_offset)
• Err(StateError) - Calculation failed

pub async fn total_lag(&self) -> StateResult<i64>

Get total lag across all partitions

pub async fn restore(&mut self) -> StateResult<usize>

Restore offsets from storage

Loads all committed offsets for the consumer group from the offset store.

Returns

• Ok(restored_count) - Number of offsets restored
• Err(StateError) - Restore failed

pub async fn reset_offset( &mut self, topic: &str, partition: u32, strategy: OffsetResetStrategy, partition_high_watermark: Option<i64>, ) -> StateResult<i64>

Reset offset for a partition

Arguments

• topic - Kafka topic
• partition - Partition number
• strategy - Reset strategy (earliest, latest, specific offset, etc.)
• partition_high_watermark - Current high watermark for the partition

Returns

• Ok(new_offset) - The new offset value
• Err(StateError) - Reset failed

pub async fn notify_watermark( &mut self, watermark: Watermark, ) -> StateResult<bool>

Notify manager of watermark advancement (for OnWatermark strategy)

Arguments

• watermark - New watermark value

Returns

• Ok(committed) - True if commit occurred
• Err(StateError) - Notification failed

pub async fn stats(&self) -> OffsetStats

Get statistics

pub async fn get_all_committed(&self) -> HashMap<TopicPartition, OffsetInfo>

Get all committed offsets

pub async fn get_all_pending(&self) -> HashMap<TopicPartition, OffsetInfo>

Get all pending offsets (not yet committed)

pub async fn clear(&mut self) -> StateResult<()>

Clear all offsets for this consumer group