Warmup Guide — Apache Flink

Zero-to-principal primer for Phase 04: Flink's execution and state model, event time and watermarks, the distributed-snapshot checkpoint algorithm, savepoints and rescaling, exactly-once sinks, and how to read the failure signatures interviewers love (rising checkpoint duration + growing state + delayed watermarks + lag).

Table of Contents

Flink runs a streaming dataflow: a directed graph of operators (sources → transforms → sinks) processing records one at a time (not micro-batches — that's Spark, P05). The runtime pieces:

  • JobManager (the coordinator): builds the execution graph, schedules tasks, triggers checkpoints, coordinates recovery. TaskManagers (the workers): run operator subtasks in task slots (a slot = a unit of resource isolation; one slot can run a pipeline of chained operators).
  • Parallelism: each operator has N parallel subtasks; records are distributed across them. keyBy(k) partitions the stream by key (hash) so all records for a key go to the same subtask — keyed streams are where keyed state lives, and the per-key ordering of P01 applies.
  • Operator chaining: Flink fuses adjacent operators (e.g. map→filter) into one task to avoid serialization/handoff — a real performance lever and a thing to know when reading the job graph.
  • Sources/sinks: connectors (Kafka, Kinesis, files, Iceberg). Modern sources implement a unified interface with split enumeration and per-split watermarks (Ch. 2).

Chapter 2: Event Time and Watermarks, Precisely

P01 established why event time (wall clocks lie). Flink's implementation:

  • Each event carries a timestamp (extracted at the source). A WatermarkStrategy generates watermarks — typically boundedOutOfOrderness(d): W = max_ts_seen − d. The watermark is a record-like marker flowing with the stream that asserts "no event with ts ≤ W is still coming."
  • Per-partition watermarks + the min rule: each source partition tracks its own watermark; an operator with multiple inputs takes the minimum across them (the slowest input gates progress). This is why a single idle partition can stall watermarks org-wide — Flink offers withIdleness(...) to let idle partitions be ignored. A classic prod incident: "watermarks stopped advancing" → one partition went silent.
  • Allowed lateness: after a window fires at the watermark, keep it for allowedLateness to absorb stragglers (re-emitting corrections); beyond that, route to a side output (OutputTag) — late data is information, never silently dropped.
  • The triangle (P01): larger d = more complete first results but higher latency and more retained state; you measure the real out-of-order distribution and choose d against it.

Chapter 3: Windows and Triggers

Windows turn an unbounded stream into bounded computations:

  • Tumbling (fixed, non-overlapping), sliding/hopping (fixed size, overlapping — one event in size/slide windows, multiplying state), session (gap-based, dynamic boundaries that merge when a late event bridges two sessions — the trickiest), and global (one window, you supply the trigger).
  • Assigner → trigger → (evictor) → function: the assigner places events in windows; the trigger decides when to fire (default: on watermark past window end; can fire early on processing time or counts); the function aggregates (reduce/aggregate are incremental and cheap; process buffers all elements and is expensive).
  • State cost: open windows × keys × accumulator size. This is the number behind "our RocksDB state is growing" — usually too many open windows (huge allowed lateness, sliding windows, or unbounded key cardinality). The lab keeps an explicit lifecycle (open→closeable→expired) for exactly this reason.

Chapter 4: State and State Backends

State is what makes streaming powerful and dangerous:

  • Keyed state (scoped to the current key): ValueState, ListState, MapState, ReducingState, AggregatingState. Operator state (scoped to a subtask): used by sources/sinks (e.g. Kafka offsets). Broadcast state: a low-throughput stream (rules, config) replicated to every subtask — the fraud-rules pattern.
  • State backends: HashMapStateBackend (state on the JVM heap — fast, but bounded by memory, full snapshots) vs EmbeddedRocksDBStateBackend (state in an embedded LSM tree on local disk — supports state far larger than memory, incremental checkpoints, at the cost of serialization + disk I/O per access). The choice is "does my state fit in heap?" — large keyed state ⇒ RocksDB.
  • RocksDB specifics that show up on-call: it's an LSM (P08/P11) so writes go to memtables then SST files with background compaction; state grows when compaction can't keep up or when TTL isn't set; read amplification from many SST levels. State TTL expires old entries so state doesn't grow unbounded (the bounded-dedup idea of P01, as a feature).
  • The lab's snapshot()/restore() is a state backend's job: serialize all keyed state to a durable structure and rebuild it.

Chapter 5: Checkpoints — The Distributed Snapshot

How do you snapshot a distributed, running dataflow consistently, without stopping it? The Chandy-Lamport asynchronous barrier snapshotting algorithm — Flink's crown jewel:

  1. The JobManager injects a checkpoint barrier (numbered N) into the source streams.
  2. Barriers flow with the data. When an operator receives barrier N on all its inputs (alignment), it snapshots its state (asynchronously, to durable storage — S3) and forwards barrier N downstream.
  3. When all operators (including sinks) confirm checkpoint N, the JobManager marks it complete. On failure, the whole job restarts from the last complete checkpoint — every operator restores its state and sources rewind to the offsets in that checkpoint.

Key subtleties:

  • Alignment is where exactly-once is bought and where it can hurt: an operator must wait for the barrier on every input, buffering the faster inputs. Under skew or backpressure, alignment time balloons → rising checkpoint duration (the Round-2 sym ptom). Unaligned checkpoints trade some of this by snapshotting in-flight data, helping under backpressure.
  • Exactly-once vs at-least-once checkpointing: at-least-once skips alignment (lower latency) but can double-count on recovery. Exactly-once aligns.
  • Because sources rewind to checkpointed offsets and operators restore exact state, the effect is exactly-once within the Flink job — output exactly-once still needs Ch. 7.
  • The lab's checkpoint-recovery test (restore-and-continue == one run) is this property in miniature.

Chapter 6: Savepoints, Upgrades, and Rescaling

  • Checkpoint vs savepoint: same mechanism, different ownership. Checkpoints are automatic, engine-owned, optimized for fast recovery, and may be cleaned up. Savepoints are manual, user-owned, portable, self-contained snapshots you take on purpose to upgrade code, change parallelism, migrate clusters, or A/B a new version. The deploy pattern: take a savepoint → stop the job → start the new job from the savepoint.
  • Stateful upgrades require operator UIDs (so Flink can map saved state to operators across code changes) and state-schema compatibility (evolving the state's types — P03 ideas applied to state). Forget a UID and your state won't restore.
  • Rescaling: changing parallelism means redistributing keyed state. Flink solves this with key groups: each key maps to one of maxParallelism key groups (fixed at job creation); key groups are assigned to subtasks in contiguous ranges; rescaling moves whole key-group ranges between subtasks. The lab implements key_group and task_for_key_group exactly. The catch interviewers probe: maxParallelism is fixed forever at first run — set it generously, because you can't exceed it later.

Chapter 7: Exactly-Once Sinks

Checkpoints give exactly-once state; the output to an external system needs more, because a checkpoint restore replays the records since the last checkpoint — so the sink must not re-emit them. Two patterns:

  • Idempotent sink: writes are naturally repeatable — upsert by key, set-union, overwrite-by-id. Replaying produces the same final state. Simplest; works when the sink supports it (Cassandra upsert, a keyed KV store).
  • Two-phase commit (Flink's TwoPhaseCommitSinkFunction): on each checkpoint barrier, pre-commit the buffered output as a pending transaction (durable, invisible); on the checkpoint-complete notification, commit it (visible). On recovery, re-commit pending transactions (they were part of the checkpoint) — idempotently. This is how transactional Kafka sinks, the file sink ("pending" → "finished" file rename), and the Iceberg/Delta sink (P09: stage data files, commit the snapshot) achieve exactly-once. The lab implements this protocol, including the crucial property: buffered-but-not-yet- pre-committed data is dropped on crash — and that's correct, because the source offset for it wasn't checkpointed either, so it'll be re-read.

Chapter 8: Backpressure and the Failure Signatures

Flink uses credit-based flow control: a downstream subtask grants the upstream credits (buffer space); when it's slow, credits dry up and the upstream slows — backpressure propagates cleanly to the source (which slows its Kafka consumption → lag). The Round-2 signature and how the symptoms connect:

slow operator / sink  →  backpressure  →  buffers fill  →  checkpoint barriers move slowly
                                                         →  alignment time ↑  →  checkpoint DURATION ↑
                                                         →  source slows  →  Kafka LAG ↑
late/out-of-order data + huge allowed lateness/sliding   →  many open windows  →  STATE size ↑
no state TTL / unbounded key cardinality                 →  RocksDB grows  →  STATE size ↑, GC ↑

The diagnostic discipline: find the backpressured operator (Flink UI colors it), then ask why it's slow — a slow sink (external system, the usual culprit), data skew on a hot key (P01), an expensive process function buffering elements, or under-parallelism. The fix matches the cause: scale the sink / async I/O, re-key or salt the hot key, switch to incremental aggregation, add state TTL, or raise parallelism (via savepoint rescale, Ch. 6). Saying that chain out loud is passing Round 2.

Not everything needs the DataStream API. Flink SQL treats streams as dynamic tables (a changelog) and lets you write windowed aggregations, joins, and pattern matching (MATCH_ RECOGNIZE) declaratively — with the same event-time/watermark/state semantics underneath. The streaming-batch unification: the same SQL can run over a bounded source (batch) or unbounded (streaming). Use SQL for the 80% of standard transforms (it's less code and optimizable); drop to DataStream for custom state, timers, and complex logic. Know that the state/exactly-once story is identical — SQL doesn't escape watermarks or checkpoints, it sits on them.

Lab Walkthrough Guidance

Lab 01 — Flink-Grade Stream Processor, suggested order:

  1. assign_window + KeyedWindowProcessor.process/flush (watermark, lateness, side output, dedup) — get the order-independence and lateness tests green first.
  2. snapshot()/restore() — then the headline test: process half, snapshot, restore into a fresh operator, process the rest → identical to one run (checkpoint recovery).
  3. key_group / task_for_key_group — deterministic, contiguous, covers all tasks (rescaling).
  4. TwoPhaseCommitSink — invisible until commit; crash before commit → recover exactly-once; double commit idempotent; uncheckpointed buffer dropped.

Success Criteria

You are ready for Phase 05 when you can, from memory:

  1. Draw Flink's execution model (JobManager/TaskManager/slots/parallelism/keyBy/chaining).
  2. Explain watermark generation, the min-across-inputs rule, and the idle-partition stall.
  3. Explain the barrier-snapshot algorithm and why alignment buys exactly-once and can stall.
  4. Contrast checkpoints and savepoints and describe a stateful upgrade (UIDs!).
  5. Explain key groups and why maxParallelism is fixed at job creation.
  6. Explain idempotent vs two-phase-commit sinks and where each fits.
  7. Walk the Round-2 failure chain (backpressure → checkpoint duration + state + lag) and give the cause-matched fix.

Interview Q&A

Q (Round 2): A Flink job has rising checkpoint duration, growing RocksDB state, delayed watermarks, and growing Kafka lag. Diagnose. These are one story, not four bugs. Something downstream is slow (most often the sink, or a hot-key skew, or an expensive process function). That backpressures upstream: buffers fill, so checkpoint barriers move slowly and alignment time climbs → checkpoint duration ↑; the source is throttled → Kafka lag ↑. Meanwhile delayed watermarks point at event-time/idle-partition issues or the same backpressure delaying barrier/marker flow, and growing RocksDB state points at too many open windows (huge allowed lateness / sliding windows / unbounded keys) or missing state TTL. I'd find the backpressured operator in the UI, confirm whether it's the sink or a skewed key, then fix the cause: async/scaled sink, re-key/salt the hot key, incremental aggregation, add state TTL, or rescale via savepoint — and consider unaligned checkpoints to relieve the alignment stall while I fix the root cause.

Q: Checkpoints vs savepoints — when do you use each? Checkpoints are the engine's automatic safety net for failure recovery — frequent, fast, engine-owned, possibly cleaned up. Savepoints are my tool for planned change — upgrade the job's code, change parallelism, migrate clusters, fork for A/B — taken on purpose, self-contained, portable. The upgrade ritual: savepoint → stop → start new version from savepoint, which only works if operators have stable UIDs and the state schema is compatible.

Q: How does Flink rescale stateful jobs, and what's the gotcha? Keyed state is partitioned by key group (hash(key) % maxParallelism), and key groups are assigned to subtasks in contiguous ranges, so changing parallelism just moves whole key-group ranges — no per-key reshuffle needed. The gotcha is that maxParallelism is fixed at job creation and caps your future parallelism; set it generously up front (changing it later requires a state migration), because you can scale up to it but never beyond it.

Q: You need exactly-once into S3/Iceberg. How? Checkpoints give exactly-once state inside Flink, but the S3 write needs a transactional sink: two-phase commit. On each checkpoint barrier the sink pre-commits — for files, it writes data to a temp/pending location; for Iceberg, it stages data files. On checkpoint-complete it commits — rename pending files to final, or commit the Iceberg snapshot atomically. On recovery it re-commits pending transactions idempotently. Crucially, data buffered since the last checkpoint is dropped on a crash, which is correct because the source offset wasn't advanced past it either — so it's re-read and re-processed, netting exactly-once.

References