Apache Flink Architecture in Depth

Why architecture matters here

Streaming is unusually correctness-sensitive, and Flink's architecture reads best as a set of mechanisms that must cooperate for a result to be right: the process and slot model that decides where work runs, watermarks that define what time means, keyed state that has to survive a machine dying, and a snapshot protocol that ties those into a consistent cut of a running system.

Every guarantee below has a configuration prerequisite, and most Flink incidents are one of them quietly unmet — a watermark that never advances, a checkpoint that cannot finish under backpressure, a transaction that expires while the job is down. Each section works through the mechanism first and the failure mode it produces second.

Processes, slots, and the deployment model

A Flink cluster is one JobManager process plus a set of TaskManagers. JobManager is a role split three ways: the Dispatcher owns the REST endpoint and web UI and spawns a JobMaster per job; the JobMaster owns one job's execution graph, schedules its subtasks and drives its checkpoints; the ResourceManager negotiates TaskManager containers with standalone, YARN or Kubernetes. Under HA only one JobManager is leader — leader election and the pointers to the JobGraph and the latest completed checkpoint live in ZooKeeper or Kubernetes ConfigMaps, while the state itself sits on the distributed filesystem. Losing the JobManager without HA does not lose state; it loses the job, and someone has to resubmit it against the right checkpoint.

A TaskManager is one JVM offering task slots. taskmanager.numberOfTaskSlots defaults to 1, not to the core count, and a slot is a memory reservation rather than a CPU reservation: slots divide managed memory and task heap evenly but nothing pins them to cores, so more slots than cores oversubscribes CPU silently. Memory is carved from taskmanager.memory.process.size into task heap, managed memory (RocksDB, sorting), network buffers, metaspace and overhead — a Flink OOM is usually one of those regions, not "the heap". Application mode runs your main() on the JobManager and gives each application its own cluster, so failure domains follow the application; session mode shares a long-lived cluster, making submission cheap at the price that one TaskManager crash takes down every job with a subtask on it.

Operator chaining and slot sharing

Chaining fuses consecutive operators into a single task when parallelism matches and the exchange is one-to-one. A chained map -> filter -> map runs in one thread and passes records by direct method call: no serialization, no network buffer, no queue. The chain breaks at any keyBy or rebalance, at a parallelism change, and at an explicit disableChaining(). Breaking one deliberately is a diagnostic move — an operator in its own task gets its own metrics, so you can see whether it is the expensive one.

Slot sharing is the independent packing rule: subtasks of different operators may occupy the same slot, so a job needs as many slots as its maximum operator parallelism, not the sum. A source at parallelism 4 feeding a window at 32 needs 32 slots, not 36, and each slot ends up holding a full vertical slice of the pipeline. Push a memory-hungry operator into its own slotSharingGroup when you want it to stop competing for managed memory with everything else.

The architecture: every piece explained

Read the diagram top to bottom. The top row is the control plane: a job is compiled to a dataflow graph, the JobManager schedules it, and TaskManagers run the subtasks in slots. The middle rows are the data plane and its semantics — the operator graph and its parallelism, the state backend behind every keyed operator, the replayable sources and transactional sinks the guarantees depend on, and the windows and watermarks that give unbounded input a notion of completeness. The bottom row is the fault-tolerance machinery. Each is unpacked below.

From StreamGraph to ExecutionGraph

Your DataStream or Table calls build a StreamGraph in the client: a logical DAG of one node per operator, carrying parallelism, chaining hints and the serializers resolved from your types. It is lowered to a JobGraph, at which point chaining has been applied and each chain has collapsed into one JobVertex; the JobGraph plus your jars is what crosses the wire. On the JobManager the JobMaster expands it into the ExecutionGraph — one ExecutionVertex per subtask, with explicit intermediate result partitions for every exchange — and the scheduler places each vertex into a slot.

Parallelism only becomes concrete at that last expansion, which is why rescaling is a re-expansion rather than a recompile. It also explains a recurring confusion: the boxes in the web UI are JobGraph vertices, so an operator you named in code may not appear at all — it was fused into its neighbour. Name your operators and give every stateful one a uid(), and you get readable graphs and stable state mapping for free.

Event time, watermarks, and the idle-partition stall

Processing time is the wall clock of whichever machine ran the operator: cheap and non-deterministic, so a replay lands the windows differently. Event time is a timestamp carried by the record, so replaying the same input produces the same output — the property that makes backfills mean anything. A watermark of t is an assertion that no element at or below t will follow, and operators fire windows and event-time timers on it. forBoundedOutOfOrderness(Duration.ofSeconds(20)) emits largest timestamp seen minus 20 seconds; that bound is purely a completeness-versus-latency dial.

Propagation is where the pain lives. An operator's watermark is the minimum over its input channels, because it can assert only what its slowest input asserts. One Kafka partition with no traffic pins its channel, the minimum never advances, and every downstream window stops firing while state accumulates. The signature is unmistakable once seen: healthy throughput, zero output, climbing checkpoint size. .withIdleness(...) excludes a quiet channel from the minimum, explicitly trading away the guarantee for anything that arrives on it later. Records behind the watermark are dropped silently unless you set allowedLateness (window state survives to re-fire, so downstream must tolerate updates rather than appends) or sideOutputLateData. Assigners, triggers and sliding-window state amplification are covered in stream windowing.

Keyed state, key groups, and the backend choice

keyBy exists to make state partitionable. Inside a keyed operator, ValueState, ListState and MapState are implicitly scoped to the current key — you never pass a key, because the runtime sets it before every processElement call. Between key and subtask sits a level most people meet during an incident: keys hash into one of maxParallelism key groups, each subtask owns a contiguous range, and the key group is the atomic unit of redistribution on rescale. maxParallelism is baked into the first checkpoint and cannot change on restore (it defaults to 128 at parallelism 128 or below), so picking it carelessly permanently caps how far the job can scale.

HashMapStateBackend holds state as Java objects on the heap: no serialization on access, nanosecond reads, but it must fit in heap and GC behaviour degrades as it grows. EmbeddedRocksDBStateBackend holds it in an embedded LSM tree off-heap and on local disk: every access serializes and deserializes, roughly an order of magnitude more expensive per operation, in exchange for state larger than memory and incremental checkpoints that upload only newly written SST files. Heap while state per TaskManager is a few GB and per-record latency matters; RocksDB at tens of GB, or whenever the key space is unbounded. Where checkpoints are written is a separate setting from which backend holds the state.

End-to-end streaming job flow

Concretely: per-user hourly purchase totals from a Kafka topic. The submitted graph is KafkaSource -> keyBy(user_id) -> 1h tumbling window -> sum -> KafkaSink. At parallelism 8 the JobMaster deploys eight subtasks of each chain into slots. The source emits records and watermarks; keyBy hashes user_id into a key group, so a given user's running total lives in exactly one subtask's keyed state; each window emits when the watermark passes its end. Every checkpoint interval, barriers sweep the graph and each subtask uploads its state alongside its Kafka offsets. When a TaskManager dies the JobMaster notices via missed heartbeats, redeploys the affected subtasks, restores their state from the last completed checkpoint and rewinds the offsets stored with it — the totals are neither lost nor double-counted.

Asynchronous barrier snapshotting

Flink's checkpoint is Chandy-Lamport adapted to acyclic dataflows, with the markers carried in-band. The coordinator inside the JobMaster triggers checkpoint n; each source subtask records its reading position and injects barrier n into its output. Barriers travel with the records and never overtake them. An operator that has seen barrier n on every input channel snapshots its state, forwards the barrier downstream and acknowledges. Once every subtask has acknowledged, the coordinator writes the checkpoint metadata and calls notifyCheckpointComplete on all operators.

The load-bearing word is asynchronous. The synchronous phase is deliberately tiny — a copy-on-write flip on the heap backend, a native snapshot of immutable SST files on RocksDB — while the expensive upload to S3 or HDFS runs on a background thread and records keep flowing. Reported duration is alignment plus sync plus async, broken out per subtask, and reading that split is the diagnosis: a large sync phase means state is being copied badly, a large async phase means the object store is the constraint, a large alignment phase means backpressure.

Alignment, unaligned checkpoints, and backpressure

Alignment is what makes the snapshot a consistent cut. An operator holding barrier n from channel A but not from B stops consuming A, buffering its records, until B's barrier arrives — otherwise post-barrier records from A get folded into a pre-barrier snapshot and counted twice on restore. Alignment time therefore equals the skew between channels, and under backpressure barriers advance only as fast as the slowest queue drains. Checkpoint duration explodes precisely when the job is already unhealthy, hits its timeout (10 minutes by default), and a job that cannot complete a checkpoint cannot fail over without replaying from a very old one. That feedback loop is the most common Flink incident shape.

Unaligned checkpoints invert the tradeoff: the barrier overtakes queued buffers, and the in-flight data it jumped over is written into the checkpoint as part of the state, decoupling duration from backpressure. You pay with larger checkpoints, more snapshot IO, and a restore that has to re-inject those buffers. The usual production setting enables unaligned mode behind an aligned-checkpoint timeout, so healthy runs stay aligned and only a stalling one switches. Credit-based flow control itself is covered in backpressure architecture; the Flink-specific part is diagnosis, where backPressuredTimeMsPerSecond and busyTimeMsPerSecond per subtask identify the bottleneck as the first task that is busy but not backpressured. Everything behind it is a victim.

What exactly-once actually guarantees

Stated precisely: given a replayable source, every record affects operator state exactly once. The implementation is rewind and replay — restore state from checkpoint n, reset the sources to the offsets captured in it, reprocess. Records genuinely are read more than once; what is not duplicated is their effect on state. Side effects your code performs outside state, such as an HTTP call in a map or a plain JDBC write, happen again, and no checkpoint mechanism can prevent that.

End to end, the sink must join the same two-phase commit: on the barrier it flushes and pre-commits — the Kafka sink into an open transaction, the file sink into an in-progress file — and only notifyCheckpointComplete triggers the commit. Three consequences to design around. Nothing is visible to a read_committed consumer until the producing checkpoint completes, so the checkpoint interval is a floor on visible latency regardless of per-record processing time. The broker's transaction.max.timeout.ms must exceed the sink's transaction timeout, because a job down longer than that timeout has its pending transaction aborted — silent data loss, the opposite of the failure most people brace for. And a sink that upserts on a primary key reaches the same observable result under at-least-once delivery with none of that coupling, which is why idempotence beats transactions whenever the destination allows it. The complementary knob is checkpoint mode: at-least-once skips alignment for lower latency and accepts double-counting after a restore. The barrier-and-commit protocol in general is treated in exactly-once stream processing.

Failure modes you will actually hit

Distinct from checkpoints: savepoints are user-triggered, retained and owned by you rather than by the job, which is what makes them the mechanism for stateful upgrades, parallelism changes and version migrations — see Flink savepoints architecture.

When Flink is the right tool

Flink earns its operational cost when the computation is continuous, stateful and correctness-sensitive under out-of-order input: joins across unbounded streams, sessionization, online feature computation, pattern detection over time. Event-time semantics plus large keyed state plus a snapshot mechanism that survives node loss is the combination that is genuinely hard to rebuild.

If processing is stateless per record, a plain consumer with a thread pool is far cheaper to run and reason about. If the scope is one team's single input topic to single output topic, Kafka Streams is a library rather than a cluster — no JobManager, no slots, scaling tied to the consumer group. Flink can also run bounded input in batch mode through the same API, but choosing it for batch alone means operating a checkpointing system you never needed. The real price is an always-on distributed system with a durable filesystem dependency, checkpoint duration and state size as first-class alerting signals, and a deliberate procedure for every stateful upgrade.