Build your own mini stream processor · Stage 3 of 8

Event time + watermarks (min across inputs)

Flink is an event-time engine. A watermark of t asserts that no more events with timestamp <= t will arrive, which lets downstream operators decide when time has "passed" a point. You implement watermark generation and the rule that governs multi-input operators: the output watermark is the minimum of the input watermarks.

New API this stage

BoundedOutOfOrderness(delay)        # .on_event(ts); .current_watermark()
assign_watermarks(events, delay)    # -> [Event, Watermark, Event, Watermark, ...]
watermark_min(input_watermarks)     # -> min
merge_watermarks(streams)           # merge N inputs; combined wm = min across them

This stage's task

BoundedOutOfOrderness(delay): track the max event ts seen; the current watermark is max_ts - delay, and it is monotone non-decreasing (an out-of-order older event must not lower it).
assign_watermarks(events, delay): emit each event, then — only when the watermark advances — a Watermark after it.
watermark_min(inputs) = min(inputs).
merge_watermarks(streams): read inputs round-robin; the combined watermark is min across all inputs' latest watermarks, advancing only when that min advances. A slow/silent input therefore pins the combined watermark (the classic stall that watermark idleness is designed to relieve).

Constraints

Watermarks strictly monotone. Deterministic; no sleeps. Suite < 60 s.
Earlier stages still pass.

Run it

de start FLK-BYO/03
de test

Hints

Hint 1

A watermark of t means 'no more events with ts <= t are expected'. BoundedOutOfOrderness tracks the max event ts seen and emits watermark = max_ts - delay. It must be MONOTONE: never emit a watermark lower than a previous one (keep last_wm = max(last_wm, candidate)).

Hint 2

assign_watermarks walks the events in order; after each event it updates the generator and, only if the watermark ADVANCED, appends a Watermark(wm) after that event. So out-of-order (older) events do not emit a regressing watermark.

Hint 3

watermark_min(inputs) = min(inputs). merge_watermarks reads N input streams round-robin, remembers each input's latest watermark, and emits a combined Watermark = min across ALL inputs (only once ALL inputs have a watermark, and only when the min advances). This is why one silent/slow input pins the combined watermark.

Run it locally

uv run de start FLK-BYO/03-event-time-watermarks
uv run de test