Lab 01 — Flink-Grade Stream Processor

Phase: 04 — Apache Flink | Difficulty: ⭐⭐⭐⭐⭐ | Time: 8–10 hours

This is the flagship of the whole streaming half of the track. You build the core of Flink's event-time machinery by hand — windows, watermarks, allowed lateness, keyed state — and then the two things that make Flink Flink: checkpoint/restore (so a crash resumes with exact state) and a two-phase-commit sink (so output is exactly-once). Plus key-group rescaling, so you can change parallelism safely.

What you build

  • KeyedWindowProcessor — tumbling event-time sum/count per key with watermark-driven emission, allowed-lateness re-emission, and a late-event side output (never dropped)
  • snapshot() / restore() — capture ALL operator state into a portable structure and rebuild from it: checkpoint recovery. The test proves restore-and-continue == one uninterrupted run.
  • key_group / task_for_key_group — Flink's mechanism: keys → fixed key groups → contiguous task ranges, so rescaling redistributes whole key-group ranges
  • TwoPhaseCommitSink — write → pre_commit(barrier) → commit(complete); nothing is visible until its checkpoint completes, a crash mid-way recovers exactly-once, and a double commit is idempotent

Key concepts

ConceptWhat to understand
Watermarkmax_ts − delay; emission is watermark-driven, not arrival-driven
Allowed latenessemitted ≠ done; windows linger and re-fire corrections
Checkpointa consistent snapshot of ALL state; restore resumes exactly
Savepointa portable, user-owned checkpoint (upgrades/migrations)
Key groupthe unit of state redistribution; why maxParallelism is fixed
Two-phase commitexactly-once output: commit only on checkpoint complete, idempotently

Run

pip install -r requirements.txt
pytest test_lab.py -v
LAB_MODULE=solution pytest test_lab.py -v

Success criteria

  • All 15 tests pass — test_checkpoint_restore_continues_identically and test_crash_between_precommit_and_commit_recovers_exactly_once are the ones that certify you understand Flink's correctness model.
  • You can explain why output is invisible until commit, and what happens to buffered data on a crash (dropped — and why that's still exactly-once).
  • You can explain why key groups (not keys) are the rescaling unit and why that fixes maxParallelism at job creation.

Extensions

  • Add sliding and session windows (session needs merge-on-late-bridge — genuinely hard; do it to feel why state is the cost center).
  • Make seen_ids bounded by a TTL behind the watermark (real dedup state can't grow forever — P01's bounded-dedup tradeoff, now in a real operator).
  • Add broadcast state: a second "rules" stream that every key sees (the fraud-rules pattern from the JD's Problem 2).
  • Add a RocksDB-style spilling state backend behind an interface and measure the heap-vs-disk tradeoff — the abstraction real Flink exposes.