Struct BoundedOutOfOrdernessWatermark 

pub struct BoundedOutOfOrdernessWatermark { /* private fields */ }

Bounded out-of-orderness watermark generator

This generator tracks the maximum timestamp seen and generates watermarks by subtracting a fixed delay. It handles per-partition watermarks and can detect idle partitions.

Example

use processor::watermark::{BoundedOutOfOrdernessWatermark, WatermarkGenerator};
use std::time::Duration;

let mut generator = BoundedOutOfOrdernessWatermark::new(
    Duration::from_secs(5),
    None,
);

// Process events
generator.on_event(1000, 0);
generator.on_event(2000, 0);
if let Some(wm) = generator.on_event(3000, 0) {
    // Watermark at 3000 - 5000 = -2000 (clamped to min)
}

Implementations

impl BoundedOutOfOrdernessWatermark

pub fn new( max_out_of_orderness: Duration, idle_timeout: Option<Duration>, ) -> Self

Creates a new bounded out-of-orderness watermark generator

Arguments

• max_out_of_orderness - The maximum allowed delay for late events
• idle_timeout - Optional timeout for detecting idle partitions

pub fn with_config(config: BoundedOutOfOrdernessConfig) -> Self

Creates a new watermark generator with custom configuration

pub fn active_partitions(&self) -> Vec<u32>

Gets all active partitions (non-idle)

pub fn idle_partitions(&self) -> Vec<u32>

Gets all idle partitions

pub fn is_late_event(&self, timestamp: i64) -> bool

Checks if an event is late based on the current watermark

pub fn get_lateness(&self, timestamp: i64) -> i64

Gets the lateness of an event (how late it is compared to the watermark)

Returns 0 if the event is not late

Trait Implementations

impl WatermarkGenerator for BoundedOutOfOrdernessWatermark

fn on_event(&mut self, timestamp: i64, partition: u32) -> Option<Watermark>

Called when a new event arrives

fn on_periodic_check(&mut self) -> Option<Watermark>

Called periodically to check if a watermark should be emitted

fn current_watermark(&self) -> Watermark

Gets the current watermark without advancing it

fn reset(&mut self)

Resets the watermark generator to its initial state

fn partition_watermark(&self, partition: u32) -> Option<Watermark>

Gets the watermark for a specific partition