Struct PartitionedWatermarkTracker 

pub struct PartitionedWatermarkTracker { /* private fields */ }

Tracks watermarks across partitions within sources.

Extends WatermarkTracker to support partition-level granularity. The combined watermark is the minimum across all active partitions.

Thread Safety

This tracker is NOT thread-safe. For thread-per-core architectures, use CoreWatermarkState for per-core tracking and coordinate via the runtime.

Example

use laminar_core::time::{PartitionedWatermarkTracker, PartitionId, Watermark};

let mut tracker = PartitionedWatermarkTracker::new();

// Register Kafka source with 4 partitions
tracker.register_source(0, 4);

// Update individual partitions
tracker.update_partition(PartitionId::new(0, 0), 5000);
tracker.update_partition(PartitionId::new(0, 1), 3000);
tracker.update_partition(PartitionId::new(0, 2), 4000);
tracker.update_partition(PartitionId::new(0, 3), 4500);

// Combined watermark is minimum (3000)
assert_eq!(tracker.current_watermark(), Some(Watermark::new(3000)));

// Mark partition 1 as idle
tracker.mark_partition_idle(PartitionId::new(0, 1));

// Now combined watermark advances to 4000 (min of active partitions)
assert_eq!(tracker.current_watermark(), Some(Watermark::new(4000)));

Implementations

impl PartitionedWatermarkTracker

pub const DEFAULT_IDLE_TIMEOUT: Duration

Default idle timeout (30 seconds).

pub fn new() -> Self

Creates a new partitioned watermark tracker.

pub fn with_idle_timeout(idle_timeout: Duration) -> Self

Creates a new tracker with custom idle timeout.

pub fn register_source(&mut self, source_id: usize, num_partitions: usize)

Registers a source with the specified number of partitions.

Creates partition state for each partition in the source. Call this when a source is created or when Kafka partitions are assigned.

pub fn add_partition( &mut self, partition: PartitionId, ) -> Result<(), WatermarkError>

Adds a partition to a source (for Kafka rebalancing).

Errors

Returns an error if the partition already exists.

pub fn remove_partition( &mut self, partition: PartitionId, ) -> Option<PartitionWatermarkState>

Removes a partition from tracking (for Kafka rebalancing).

Returns the partition state if it existed.

pub fn update_partition( &mut self, partition: PartitionId, watermark: i64, ) -> Option<Watermark>

Updates the watermark for a specific partition.

This also marks the partition as active and updates its last activity time.

Returns

Some(Watermark) if the combined watermark advances.

pub fn update_partition_from_event( &mut self, partition: PartitionId, event_time: i64, max_lateness: i64, ) -> Option<Watermark>

Updates watermark from an event timestamp (applies bounded lateness).

Convenience method that subtracts the configured lateness.

pub fn mark_partition_idle( &mut self, partition: PartitionId, ) -> Option<Watermark>

Marks a partition as idle, excluding it from watermark calculation.

Returns

Some(Watermark) if marking idle causes the combined watermark to advance.

pub fn mark_partition_active(&mut self, partition: PartitionId)

Marks a partition as active again.

Called automatically when update_partition is called, but can be called explicitly to reactivate a partition before receiving events.

pub fn check_idle_partitions(&mut self) -> Option<Watermark>

Checks for partitions that have been idle longer than the timeout.

Should be called periodically from Ring 1.

Returns

Some(Watermark) if marking idle partitions causes the combined watermark to advance.

pub fn current_watermark(&self) -> Option<Watermark>

Returns the current combined watermark.

pub fn partition_watermark(&self, partition: PartitionId) -> Option<i64>

Returns the watermark for a specific partition.

pub fn source_watermark(&self, source_id: usize) -> Option<i64>

Returns the watermark for a source (minimum across its partitions).

pub fn is_partition_idle(&self, partition: PartitionId) -> bool

Returns whether a partition is idle.

pub fn active_partition_count(&self, source_id: usize) -> usize

Returns the number of active partitions for a source.

pub fn partition_count(&self, source_id: usize) -> usize

Returns the total number of partitions for a source.

pub fn metrics(&self) -> &PartitionedWatermarkMetrics

Returns metrics.

pub fn num_sources(&self) -> usize

Returns the number of sources registered.

pub fn assign_partition_to_core( &mut self, partition: PartitionId, core_id: usize, )

Assigns a partition to a core (for thread-per-core routing).

pub fn partition_core(&self, partition: PartitionId) -> Option<usize>

Returns the core assignment for a partition.

pub fn partitions_for_core(&self, core_id: usize) -> Vec<PartitionId>

Returns all partitions assigned to a specific core.

pub fn idle_timeout(&self) -> Duration

Returns the idle timeout.

pub fn set_idle_timeout(&mut self, timeout: Duration)

Sets the idle timeout.

pub fn partition_state( &self, partition: PartitionId, ) -> Option<&PartitionWatermarkState>

Returns partition state for inspection.

Trait Implementations

impl Debug for PartitionedWatermarkTracker

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

Formats the value using the given formatter.

impl Default for PartitionedWatermarkTracker

fn default() -> Self

Returns the “default value” for a type.