Mock Watermark Late Event Side Output in Python

Simulates watermarking in a streaming pipeline by classifying events as on-time or late using timestamps and delays.

Medium Python 3.9+ Aug 9, 2026 Streaming & messaging 11 views 0 copies

watermark streaming side output late events simulation

Python code

31 lines

Python 3.9+

from datetime import datetime, timedelta
from typing import List, Tuple


def watermark_mock(
    events: List[Tuple[datetime, str]], watermark_delay: timedelta, max_delay: timedelta
) -> Tuple[List[Tuple[datetime, str]], List[Tuple[datetime, str]]]:
    """Simulate watermarking: events arriving on time vs. late by checking arrival time."""
    on_time_events = []
    late_events = []
    current_watermark = events[0][0] - max_delay  # initial watermark
    for event_time, _ in events:
        if event_time < current_watermark:
            late_events.append((event_time, "late"))
        else:
            on_time_events.append((event_time, "on_time"))
            current_watermark = max(current_watermark, event_time - watermark_delay)
    return on_time_events, late_events


if __name__ == "__main__":
    base = datetime(2024, 5, 1, 12, 0, 0)
    sample = [
        (base, "A"),
        (base + timedelta(minutes=5), "B"),
        (base - timedelta(minutes=10), "C"),  # late
        (base + timedelta(minutes=2), "D"),
    ]
    on_time, late = watermark_mock(sample, timedelta(minutes=3), timedelta(minutes=15))
    print("On-time:", [(t.strftime("%H:%M"), label) for t, label in on_time])
    print("Late:", [(t.strftime("%H:%M"), label) for t, label in late])

Output

stdout

On-time: [('12:00', 'on_time'), ('12:05', 'on_time'), ('12:02', 'on_time')]
Late: [('11:50', 'late')]

How it works

This mock mimics Flink's watermark and side output concept by tracking a current_watermark. It starts with an initial watermark computed from the first event time minus the max allowed delay. As events arrive, if an event's timestamp is earlier than the current watermark, it is flagged as late. Otherwise, it is considered on-time and the watermark advances to the max of its previous value and the event time minus the watermark delay. This provides a simple, deterministic way to test late-event handling logic without a real stream engine.

Common mistakes

Assuming events arrive in order; the code handles out-of-order but uses first event time for initial watermark, which may be too optimistic.
Using a fixed watermark instead of advancing it based on event times, causing correct events to be misclassified as late.
Forgetting that early events before the initial watermark are also classified as late, which may not match all frameworks.
Mixing up `watermark_delay` (allowed lateness) and `max_delay` (initial offset) roles.

Variations

Use a generator function that yields each event classification, returning a label per event.
Implement a monotonic clock version that updates the watermark after each event to simulate an ascending timestamp stream.

Real-world use cases

Unit-testing a data transformation pipeline that needs to handle late arriving sensor data without blocking the main stream.
Prototyping a fraud detection system that must flag late transaction events for separate reprocessing.
Simulating event-time processing for a financial ticker before integrating with Apache Flink or Kafka Streams.

Mock Watermark Late Event Side Output in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Streaming & messaging

Keep learning

Tutorials

Quizzes