Engineering Atlas › System Design Interviews

Apache Flink for system design interviews

13 min read · last reviewed 2026-10-17

Open in the interactive app: practice, quiz, mock interview →

In one minute

How it works

Jobs, operators and slots

You write a pipeline (DataStream API or Flink SQL), and Flink compiles it into a JobGraph of operators: sources, maps, keyBy, windows, joins, sinks. Each operator runs with some parallelism, and the runtime turns the graph into an ExecutionGraph of parallel subtasks. A JobManager schedules subtasks into slots on TaskManagers, which are JVM processes. Adjacent operators with the same parallelism and no repartition between them are chained into one task, so records pass by function call instead of serialisation. Flink 2.0 (March 2025) removed the old DataSet API, the Scala APIs and the legacy SourceFunction/SinkFunction interfaces, so batch jobs now run on the same DataStream and SQL APIs as streams (Flink 2.0 announcement).

Keyed state

keyBy(userId) hash-partitions the stream so all events for a key reach the same subtask. That subtask owns the key's state: a ValueState, ListState, MapState, or window contents. Keys hash into key groups, whose count equals the job's max parallelism: by default about 1.5 times the operator's parallelism, at least 128 and at most 32,768 (parallel execution docs). It cannot change without losing state compatibility, so set it deliberately before the first savepoint. Key groups are the unit of rescaling: changing parallelism reassigns contiguous ranges of key groups to subtasks and restores their state from the last checkpoint or savepoint.

Event time and watermarks

Each event carries a timestamp. A watermark is a special record meaning "no more events with timestamp at or below W are expected". Sources typically generate watermarks as the maximum timestamp seen minus a bounded out-of-orderness delay. An operator with several inputs tracks the minimum watermark across them, so the slowest input holds time back. When the watermark passes the end of a window, the window fires. Events that arrive later than that are late: dropped by default, or kept for allowedLateness and re-fired, or sent to a side output (timely stream processing).

Checkpoints and savepoints

Periodically the coordinator snapshots all state plus source positions; on failure, Flink restores the latest checkpoint and replays sources from those offsets. A savepoint is a user-triggered checkpoint in a portable format, used for upgrades, rescaling and migrations.

Down the stack

Layer 1The contract

Given replayable sources and checkpointing, Flink guarantees that each event affects managed state exactly once, even across failures: after recovery, state equals what it would be if the failure never happened. Output to external systems is exactly-once only with transactional or idempotent sinks; otherwise it is at-least-once, because records emitted after the last checkpoint are emitted again on replay. Event-time results are deterministic given the same input and watermark strategy, regardless of arrival order within the bound. Latency is the processing path (usually milliseconds) plus the watermark delay for windowed results; with transactional sinks, consumers using read_committed see results only after the checkpoint commits, so the checkpoint interval becomes output latency.

Layer 2The mechanism

Asynchronous barrier snapshotting. A variant of Chandy-Lamport that needs no channel logging for acyclic graphs (Carbone et al.). The coordinator tells each source to inject a checkpoint barrier n, which flows in-band with records. An operator with several inputs that sees barrier n on one input aligns: it stops reading that input until barrier n arrives on all inputs, so no post-barrier record leaks into the snapshot. It then snapshots state, forwards the barrier and resumes. When every sink has acknowledged, checkpoint n is complete (stateful stream processing). Unaligned checkpoints forward the barrier at once and store the overtaken in-flight buffers in the snapshot, keeping checkpoints fast under backpressure at the cost of size.

State backends. HashMapStateBackend keeps Java objects on the heap: fast, but memory-bound and GC-sensitive. EmbeddedRocksDBStateBackend keeps state in a RocksDB LSM tree on local disk, keyed by key-group prefix, key and namespace (such as the window). A snapshot flushes the memtable, hard-links the immutable SST files, and uploads them in the background. Incremental checkpoints (execution.checkpointing.incremental: true) upload only new SST files, with a shared registry reference-counting files used by several checkpoints. With nothing configured, Flink uses the heap backend. Flink 2.0 added an experimental third option, ForStStateBackend (state.backend.type: forst), which keeps state primarily in a remote file system with local disk as a cache and always snapshots incrementally, so TaskManagers no longer need disks sized for all their state (state backends, Flink 2.0 announcement).

Watermarks and timers. Each subtask tracks the latest watermark per input channel and emits the minimum when it advances; idle inputs can be excluded. Window operators register an event-time timer at window.maxTimestamp() (end minus 1 ms) per key and window, which fires when the watermark reaches it.

Two-phase commit sinks. On each checkpoint the sink pre-commits (for Kafka, flushes a transaction) and commits when notified that the checkpoint completed. The Kafka producer's transaction.timeout.ms must exceed the longest checkpoint plus restart time, or Kafka aborts the open transaction and its records are lost (Flink Kafka connector); brokers also cap it with transaction.max.timeout.ms, 15 minutes by default (Kafka broker configs). JobManager HA uses ZooKeeper or Kubernetes leader election and stores the pointer to the latest checkpoint metadata.

Layer 3The machine

Records move between TaskManagers over Netty TCP in network buffers, 32 KiB memory segments (taskmanager.memory.segment-size). With credit-based flow control a receiver announces its free buffers and the sender sends only that many, so a slow operator causes backpressure that reaches the source, which simply reads Kafka more slowly. Buffers flush when full or after execution.buffer-timeout.interval, 100 ms by default (configuration); the 1.x key was execution.buffer-timeout.

A RocksDB state access is a JNI call plus serialisation, a few microseconds, versus tens to hundreds of nanoseconds on heap; a block cache miss adds an SSD read of about 100 microseconds, which is why local NVMe matters. A node with hundreds of GB of state uploads only changed SST files per checkpoint, often a few GB, which takes seconds at a few hundred MB/s to S3.

Rough throughput (hedge: highly workload dependent): simple keyed aggregations reach hundreds of thousands to a few million events per second per TaskManager, and large deployments run tens of millions per second. Checkpoint intervals are typically 10 seconds to a few minutes; recovery is restart plus state download plus replay since the last checkpoint.

Why it's built this way

  1. Stateful streaming needs state co-located with compute, because a network call per event to a database caps throughput and adds milliseconds.
  2. Co-located state must survive machine loss, so it is snapshotted to durable object storage; doing that without pausing the stream requires barriers rather than a global stop.
  3. Large state makes full snapshots too slow, so RocksDB's immutable SST files enable incremental checkpoints that upload only what changed.
  4. Real data arrives late and out of order, so time must come from events, and some signal (the watermark) must say when a window is complete enough to emit.
  5. Bounded buffers and credit flow make a slow consumer slow the producer instead of exhausting memory, which keeps the system stable under load spikes.

Diagram

flowchart LR
    K["Kafka partitions"] --> S["Source (offsets in state)"]
    S -->|"records + watermarks + barrier n"| KB["keyBy(user)"]
    KB --> W["Window operator (RocksDB state)"]
    W --> SK["2PC Kafka sink"]
    CC["Checkpoint coordinator (JobManager)"] -->|"trigger n"| S
    W -->|"async SST upload"| OS["S3 checkpoint dir"]
    SK -->|"ack n"| CC
    CC -->|"notify complete n: commit"| SK

Worked example

A payments company wants per-merchant fraud features: count and sum of transactions per merchant per 1-minute tumbling window, from a Kafka topic with 200,000 events per second and 2 million active merchants. Events can arrive up to 10 seconds late due to mobile retries.

With 10 seconds of bounded out-of-orderness, each window fires about 10 seconds after it ends, triggering up to 2 million timers as the watermark crosses the minute. Window state is a count and a sum, about 50 bytes per merchant, so 100 MB per open window. The team also keeps a 30-day profile of about 2 KB per merchant, 4 GB in total, so RocksDB with incremental checkpoints is the safer backend.

Parallelism 32 gives about 6,250 events per second per subtask, comfortable at a few microseconds per state access. Checkpoints every 30 seconds upload perhaps 200 MB of changed files. A transactional Kafka sink would delay visible results by up to 30 seconds, too slow for fraud, so the sink does idempotent upserts keyed by merchant and window start instead. Events more than 10 seconds late go to a side output that a nightly batch uses for corrections.

In the wild

Trade-offs

OptionStrengthsWeaknessesChoose when
FlinkEvent time, large state, exactly-once, low latencyOperational complexity, JVM tuning, state migrationsStateful real-time pipelines at scale
Spark Structured StreamingSame engine as batch, mature ecosystemMicro-batch latency (often 100 ms to seconds), weaker event-time featuresTeams already on Spark, near-real-time is enough
Kafka StreamsA library, no cluster, state in RocksDB plus changelog topicsKafka-only, rebalances move state slowlyKafka-to-Kafka services owned by app teams
Plain consumers + databaseSimple, familiarNo event time, state round trips, manual recoveryStateless or lightly stateful transforms
Batch jobsSimplest correctnessLatency of minutes to hoursResults are needed daily or hourly

Pitfalls in production

Leadership lens

Check yourself

  1. Why does a two-input operator wait for the barrier on both inputs before snapshotting in aligned mode?

    Show answer

    Otherwise post-checkpoint records from the faster input would be in the snapshot, and replay after recovery would count them twice.

  2. A job's windows stopped firing after a deploy, but events are flowing. What is the likely cause?

    Show answer

    An idle or lagging input holds back the minimum watermark; idleness detection or fixing that input lets time advance.

  3. Why are RocksDB incremental checkpoints cheap even with terabytes of state?

    Show answer

    SST files are immutable, so a checkpoint uploads only new files and references the rest.

  4. Why can a transactional Kafka sink increase end-to-end latency?

    Show answer

    Output becomes visible to read_committed consumers only when the checkpoint completes and the transaction commits.

Sources and further study

Primary

Systems docs

Books

Other curricula

Watch

Free resources

Open in the interactive app: practice, quiz, mock interview →