Flink, simulated

The diagram. The JobManager at the top (job status, the CheckpointCoordinator’s current or last checkpoint, attempts and restarts); the Kafka topic on the left with each partition’s end offset, read offset and lag (⏸ when watermark alignment paused it); one box per subtask, column by column (source → window or process → sink; chained operators share one box); the sink’s output on the right; and the TaskManagers with their slots at the bottom. A subtask shows its watermark, a busy / backpressured / idle bar (ms per second, as the web UI reports them), what it reads or which key groups it owns and how many keys and windows it holds, and one gauge per input channel — its buffers filling up (red when out of credit), ⏸ while blocked by checkpoint alignment, ↯ while its in-flight data is being persisted for an unaligned checkpoint. Records travel the channels coloured by key (labelled key@event time when traffic is light), watermarks as W, barriers as a black bar labelled CP n, credits as small green dots going back, and the JobManager’s triggers, acknowledgements, notifications and restores in violet. Tap anything to inspect it and act on it.

Stats. Job status, records read, windows fired, results visible to the consumer, uncommitted results, duplicates, late records dropped or sent to the side output, expired state, checkpoints, the last checkpoint’s duration and alignment (or persisted in-flight records), restarts and the highest backpressure.

Watermarks. For every subtask, the last watermark of each input (split or channel) and the resulting minimum.

Output. What a consumer sees (with read_committed, only committed transactions), the whole topic log (open, pre-committed, aborted records too), and the late side output.

Checkpoints. Recent checkpoints and savepoints: trigger time, duration, size (incremental and full for RocksDB), alignment time or persisted in-flight records, and the Kafka offsets stored.

Keyed state. Each window subtask’s key-group range and its state: windows per key with their aggregate (✓ when fired), or the ValueState counter with its age and TTL.

Kafka partitions. End offset, read offset, lag and which source subtask reads each partition.

Do something. Send an event with any key, partition and event time (now, out of order, late); pause or resume a partition; trigger a checkpoint; stop with a savepoint; rescale; kill a TaskManager; switch to unaligned checkpoints, idleness or a read_uncommitted consumer.

Settings rebuild the job: window type, size, slide, gap, allowed lateness, side output, aggregate, a filter, time characteristic, watermark bound, idleness, alignment, parallelism of each operator, max parallelism, slots, chaining, state backend, TTL, checkpoint interval and mode, sink, consumer isolation, the topic’s partitions, rate, keys, disorder and late records, and per-record costs. Each shows Flink’s default.

Kafka source (FLIP-27, Kafka connector): partitions are splits assigned to readers as KafkaSourceEnumerator.getSplitOwner does — ((topic.hashCode() × 31) & 0x7FFFFFFF) % readers, then round-robin; per-split watermarks; offsets in checkpoints, committed back to Kafka only for monitoring; earliest or latest starting offsets for new jobs.

Watermarks (generating watermarks): forBoundedOutOfOrderness (max timestamp − bound − 1 ms) emitted every pipeline.auto-watermark-interval (200 ms); a reader’s watermark is the minimum over its active splits; withIdleness marks quiet splits idle and sends WatermarkStatus.IDLE; an operator’s watermark is the minimum over its non-idle input channels; watermark alignment (FLIP-182) pauses splits ahead of the group’s slowest by more than the allowed drift.

Windows (windows): tumbling and sliding assigners with TimeWindow.getWindowStartWithOffset, session windows merged with TimeWindow.mergeWindows (touching windows merge), EventTimeTrigger (fire when the watermark reaches end − 1 ms; fire at once for a late element of a live window), cleanup at end − 1 + allowed lateness, numLateRecordsDropped and sideOutputLateData; processing-time windows on the clock. The window model follows Akidau et al., The Dataflow Model (VLDB 2015).

Keyed state and key groups: KeyGroupRangeAssignment — key group = MathUtils.murmurHash(key.hashCode()) % maxParallelism, owner = key group × parallelism / max parallelism, contiguous ranges [⌈i·M/p⌉, ⌊((i+1)·M − 1)/p⌋], default max parallelism min(max(nextPowerOfTwo(p + p/2), 128), 32768) — checked against Flink 1.20’s own classes. ValueState with TTL (OnCreateAndWrite, NeverReturnExpired, cleanup in full snapshots). State backends (docs): HashMap (objects on the heap) and RocksDB (serialized, slower access, incremental checkpoints).

Network stack: credit-based flow control — a channel gets buffers-per-channel exclusive buffers plus a share of floating-buffers-per-gate; a sender sends only with credit; full output buffers block the producing task, which is how backpressure propagates back to the sources and into Kafka lag (network stack deep-dive, back pressure).

Checkpoints: asynchronous barrier snapshots (Carbone et al., Lightweight Asynchronous Snapshots for Distributed Dataflows, 2015), a variant of Chandy and Lamport’s distributed snapshots (1985): the coordinator triggers the sources, which snapshot their offsets and inject barriers; aligned barriers block a channel until all inputs' barriers arrive; unaligned barriers (FLIP-76) overtake buffered data, which is persisted as channel state; tasks acknowledge after the asynchronous upload; the coordinator completes the checkpoint and notifies every task.

Recovery: a lost TaskManager is detected after the heartbeat timeout; the whole pipelined region restarts after the restart delay; state, Kafka offsets and (for unaligned checkpoints) in-flight data are restored from the last completed checkpoint. A plain sink then writes replayed results again; the exactly-once KafkaSink (two-phase commit: pre-commit on the barrier, commit on notifyCheckpointComplete, commit-or-abort on recovery) never exposes them to read_committed consumers.

Savepoints and rescaling: stop with a savepoint, restart with another parallelism; key groups are redistributed in contiguous ranges; a parallelism above the max parallelism is refused.

One pipeline shape: Kafka source (with an optional filter) → keyBy → one window or keyed process function → sink. No joins, CEP, SQL, broadcast state, iterations or async I/O operators (the backpressure scenario models async I/O as a lower per-record cost).

Records are individual events at modest rates; costs are fixed per record; buffers hold a fixed number of records; the network has a fixed latency; checkpoint uploads have a fixed speed.

Timeouts are much shorter than Flink’s defaults (heartbeat 50 s, restart delay 1 s, checkpoint timeout 10 min — each setting shows the default). No high availability, no region failover for disconnected graphs, no checkpoint storage, no changelog state backend, no buffer debloating.

Duplicates are judged by result identity (key, window, firing). A new job started by a scenario step writes to a fresh view of the output.