Build your own mini stream processor · Stage 5 of 8

Keyed state + timers (process functions)

KeyedProcessFunction is Flink's lowest-level primitive: per-key state plus event-time timers you register and that fire when the watermark passes them. It is how you build session windows, timeouts and custom triggers.

New API this stage

ValueState(default=None)   # value/update/clear/snapshot/restore (provided)
TimerService()             # .register(key, ts); .advance(watermark) -> fired timers
class ProcessFunction:     # subclass: process_element(event, out) / on_timer(key, ts, out)
keyed.process(process_fn)  # -> KeyedProcessResult
run_process(KeyedProcessResult) -> list   # collected outputs in order

This stage's task

TimerService.advance(watermark): return the (key, ts) timers with ts <= watermark, in (ts, key) order, and remove them (fire once).
run_process: open the process function with one shared ValueState and TimerService. For each data Event, set the current key and call process_element. For each Watermark, advance timers and call on_timer for each due timer in order. out is a callback that appends to the result.

The tests build a session-timeout (close a key's session GAP after its last event) and a counter (emit per-key counts on a flush timer); outputs match an independent Python recomputation.

Constraints

Event-time timers only (watermark-driven). Deterministic fire order. No sleeps.
Earlier stages still pass. Suite < 60 s.

Run it

de start FLK-BYO/05
de test

Hints

Hint 1

TimerService.advance(watermark) returns the (key, ts) timers with ts <= watermark, in (ts, key) order, and removes them (they fire once). ValueState is already implemented: value/update/clear/snapshot/restore keyed by the current key.

Hint 2

run_process drives the chain with one ValueState and one TimerService shared by the ProcessFunction (call pf.open(state, timers) once). For each data Event: set pf._current_key = key_fn(event), then call pf.process_element(event, out). For each Watermark: fire due timers in order, calling pf.on_timer(key, ts, out) for each.

Hint 3

A session-timeout function registers a timer at event.ts + GAP on every element and stores the last ts in state; on_timer only closes the session if ts == last + GAP (i.e. no later event extended it). out is a plain callback appending to the result list.

Run it locally

uv run de start FLK-BYO/05-keyed-state-timers
uv run de test