Flink and stream processing

Apache Flink is a distributed engine for stateful computation over streams. You write a dataflow of sources, operators and sinks; Flink runs it in parallel, keeps per-key state beside the code, tracks progress in event time with watermarks, and checkpoints everything consistently, so a crash neither loses nor double-counts state. It is the default interview answer to “aggregate, join or detect patterns in a Kafka stream within seconds”, provided you can say where its exactly-once ends.

Picture it first: the festival order-counting room

During Diwali week, a sweet-shop chain with 40 branches wants live totals of boxes sold per sweet, per hour. Slips carry the time of sale, but bike riders deliver them in bundles, late and out of order.

A sorter routes each slip by sweet, so every kaju katli slip reaches clerk 3. Clerk 3 closes the 6–7 pm tally row not when the wall clock says 7, but when every rider’s latest slip shows 7:10, since no slip runs more than ten minutes behind. A 6:55 slip arriving at 7:40 goes into a “late” tray.

Every fifteen minutes a bell rings: riders note their place, and clerks finish the slips ahead of the bell and photocopy their sheets. Totals are posted only after a round of copies is filed. If a clerk faints, everyone restarts from the copies and riders re-deliver from their noted places.

Map it back: a clerk is a parallel subtask, routing by sweet is keyBy, the tally sheet is keyed state, the slip’s time is event time, the wall clock is processing time, “every latest slip shows 7:10” is a watermark, an hourly row is a window, the late tray is a side output, the bell is a checkpoint barrier, a rider’s place is a Kafka offset, the copies are a checkpoint, and posting only after filing is a transactional sink.

Data model: a dataflow of parallel operators

A Flink job is a directed graph: sources read streams (Kafka, database change logs, files), operators transform them, sinks write results. Streams are unbounded or bounded; batch is the bounded special case.

Each operator runs as n parallel subtasks. Records move between them in three ways:

• Forward: subtask i feeds subtask i, and Flink chains such operators into one task on one thread.
• keyBy: hash the key, so any upstream subtask may send to any downstream one: a network shuffle that brings one key’s events together.
• Rebalance and broadcast, for spreading load or shipping small rule sets.

The JobManager schedules tasks, coordinates checkpoints and drives recovery; for high availability, standbys take over through ZooKeeper or Kubernetes leader election. TaskManagers are worker JVMs offering task slots, which reserve managed memory, not CPU. A slot can hold one slice of every operator, so a job needs as many slots as its highest parallelism.

Most new pipelines start in Flink SQL, which compiles to the same dataflows; the DataStream API gives you explicit state, timers and windows. Flink 2.0 (March 2025) removed the DataSet and Scala APIs.

Time: event time, processing time and watermarks

Processing time is the operator machine’s clock. It is simple, but results depend on arrival: a job running ten minutes behind files clicks into the wrong windows, and a replay gives a different answer. Event time is the timestamp in the record, when the click happened, so results depend only on the data and replays repeat; billing, fraud features and corrections need it. (Ingestion time, stamped at the source, is a rarely used middle ground.)

Event time raises a new question: when is the 10:00–10:01 window complete? A phone in a metro tunnel uploads its clicks minutes late.

Watermarks

A watermark answers it. Watermark(t) flows in the stream and asserts that no more events with a timestamp ≤ t should arrive; when it passes a window’s end, the window fires.

The usual generator is bounded out-of-orderness: highest timestamp seen − a bound, emitted every 200 ms by default. The bound is your latency-versus-completeness dial: 5 s fires each window 5 s after it ends and calls anything later late; 2 minutes catches more stragglers but delays every result by 2 minutes.

Each source subtask (with Kafka, each partition) produces its own watermark, and an operator with several inputs takes the minimum. That keeps results correct, and it is why one silent partition can freeze a whole job.

Late data

An event behind the watermark is late and, with the default allowed lateness of 0, dropped. Your options: allowed lateness (keep window state longer; each straggler re-fires the window, so sinks must overwrite), a side output for anything later still, or a larger bound, which delays every window.

The Dataflow Model: what, where, when, how

Flink’s event-time design follows Google’s Dataflow Model (Akidau et al., VLDB 2015), which splits a streaming computation into four questions: what is computed (the aggregation), where in event time (the windows), when in processing time results are emitted (triggers: at the watermark, plus early or late firings), and how successive results relate (discarding, accumulating, or accumulating and retracting). Its argument: watermarks are sometimes too fast (late data still arrives) and sometimes too slow (one lagging input holds everything back), so triggers and refinements, not a perfect watermark, are how you trade correctness against latency and cost. Flink’s default trigger plus allowed lateness is the accumulating case.

Windows

Flink’s built-in window assigners:

A window is created by its first event, fires when the watermark passes its end, and is deleted once the watermark passes its end plus the allowed lateness. Prefer incremental functions (reduce, aggregate), which keep one accumulator per window instead of every event. The classic interview job, clicks per ad per minute (Flink 2.x):

OutputTag<Click> lateTag = new OutputTag<>("late-clicks") {};

SingleOutputStreamOperator<AdMinuteCount> counts = env
    .fromSource(kafkaClicks,
        WatermarkStrategy.<Click>forBoundedOutOfOrderness(Duration.ofSeconds(5))
            .withTimestampAssigner((click, ts) -> click.clickTimeMillis())
            .withIdleness(Duration.ofMinutes(1)),             // quiet partitions must not stall time
        "clicks")
    .keyBy(Click::adId)
    .window(TumblingEventTimeWindows.of(Duration.ofMinutes(1)))
    .allowedLateness(Duration.ofMinutes(1))
    .sideOutputLateData(lateTag)
    .aggregate(new CountClicks(), new AttachWindowStart())  // one accumulator per ad per window
    .uid("ad-minute-counts");                                 // stable ID for savepoints

counts.getSideOutput(lateTag).sinkTo(lateClicksSink);

Follow a click on ad 42 stamped 10:00:58. Arriving with the watermark at 10:00:57, it is on time; the window fires once a click stamped 10:01:05 or later lifts the watermark to 10:01:00. Arriving at 10:01:40, with the watermark at 10:01:30, it is late but inside the allowed minute: the window re-fires with the count plus one and the sink overwrites (ad 42, 10:00). After the watermark passes 10:02:00, the click goes to the side output.

The same job in Flink SQL, where the watermark is part of the table:

CREATE TABLE clicks (
  ad_id      BIGINT,
  click_id   STRING,
  click_time TIMESTAMP(3),
  WATERMARK FOR click_time AS click_time - INTERVAL '5' SECOND
) WITH ('connector' = 'kafka', 'topic' = 'clicks', 'format' = 'json' /* plus brokers */);

SELECT ad_id, window_start, COUNT(*) AS clicks
FROM TABLE(TUMBLE(TABLE clicks, DESCRIPTOR(click_time), INTERVAL '1' MINUTE))
GROUP BY ad_id, window_start, window_end;

Sliding windows hide cost. Flink keeps one accumulator, or one copy of each event, per window an event belongs to. A 1-hour window sliding every minute updates 60 accumulators per click, and 50,000 active ads hold 50,000 × 60 = 3 million of them. For “trending in the last hour, refreshed every minute”, count in 1-minute tumbling windows and keep a rolling sum of the last 60 per key.

State: keyed state, backends and key groups

Keyed state (ValueState, ListState, MapState and friends) is scoped to the current key, like a local key-value store sharded exactly as the stream is, so every update is a local read-modify-write with no locks or remote calls. Timers let a key wake itself: “no payment 15 minutes after checkout, so alert.” Operator state belongs to a subtask (sources keep Kafka offsets there), and broadcast state puts the same small map, such as fraud rules, on every subtask.

State backends

ForSt is Flink 2.0’s disaggregated state, aimed at containers with small disks and state in the hundreds of terabytes. In 2.0’s Nexmark runs, I/O-heavy queries with a 1 GB local cache reached 75–120% of local-state throughput.

State TTL, and sizing state up front

Unexpiring state grows forever. State TTL expires entries after a set time, though only in processing time, and costs 8 bytes per value on RocksDB. Flink SQL’s table.exec.state.ttl defaults to 0, meaning never: a streaming join or GROUP BY user_id keeps every key it has ever seen.

Size state before the interviewer asks. Deduplicating clicks by click_id for 10 minutes at 50,000/s holds 50,000 × 600 = 30 million IDs; at an assumed 60 bytes each, about 1.8 GB across the job. A 24-hour window is 144 times that, about 260 GB: time for a Bloom filter or an idempotent sink instead.

Key groups and max parallelism

Flink hashes keys into key groups, as many as the operator’s max parallelism, and gives each subtask a contiguous range (as in the first figure). Rescaling moves whole key groups instead of rehashing keys. Max parallelism is the rescaling ceiling, and changing it discards state. The default is 128 for parallelism up to 128, otherwise min(nextPowerOfTwo(p + p/2), 32768). With 128, parallelism 4 gives each subtask 128 / 4 = 32 groups; 6 gives some 22 and some 21 (about 3% imbalance); 200 is impossible. If you expect growth, set it on day one, to a number with many divisors such as 720; its metadata cost grows linearly.

Guarantees, precisely

Checkpoints: barriers instead of stopping the world

Flink’s fault tolerance is replay plus consistent snapshots, using asynchronous barrier snapshotting (Carbone et al., 2015), a variant of the Chandy–Lamport snapshot:

1. The JobManager tells every source to start checkpoint n. Each source records its position (for Kafka, the offset per partition) and injects a barrier n into its output.
2. Barriers flow with the records and never overtake them: records before barrier n belong to snapshot n, records after it to n + 1.
3. An operator with several inputs aligns, holding each input whose barrier has arrived until barrier n arrives on all of them.
4. It snapshots its state, forwards the barrier and carries on, while the state uploads asynchronously to S3 or HDFS.
5. When every sink has acknowledged, checkpoint n is complete: source offsets plus matching operator state.

Chandy–Lamport also records messages in flight on each channel. Because barriers travel in-band and inputs align, ABS needs only operator state on acyclic graphs; the replayable source stands in for the in-flight records.

Recovery is the reverse: restart, load every operator’s state from the latest completed checkpoint, and seek each source to the offsets stored with it. Records after those offsets run again against state that has never seen them, so their effect counts once. (Offsets committed to Kafka are only for lag monitoring.)

Three refinements matter at scale:

• Unaligned checkpoints (since 1.11) let barriers overtake buffered records and save those buffers in the checkpoint, so checkpoints stay quick under backpressure, at the cost of more checkpoint I/O.
• Incremental checkpoints (RocksDB, ForSt) upload only files changed since the last one: a 500 GB state that changes a few gigabytes uploads a few gigabytes.
• At-least-once mode skips alignment for slightly lower latency; after a restore some records affect state twice.

Checkpointing is off by default, and by default one failed checkpoint fails the job over.

Savepoints

A savepoint is a checkpoint you trigger by hand, in a portable format, that never expires: stop with a savepoint, then start new code, a new parallelism or a new Flink version from it. State is matched to operators by UID, so give every stateful operator a stable uid(...), or a refactor that changes the graph orphans its state. Even then, state is not guaranteed compatible from 1.x to 2.x.

Exactly-once state is not exactly-once output

Exactly-once in Flink means each event affects Flink’s state once, across failures. It does not mean each event is processed once: after a crash, records since the last checkpoint run through your code again, and anything they did outside Flink can happen twice. End to end you need a replayable source whose position is checkpointed, exactly-once checkpointing, and a sink that is transactional (two-phase commit tied to checkpoints) or idempotent (rewriting a result changes nothing).

The Kafka sink is the standard two-phase commit. With DeliveryGuarantee.EXACTLY_ONCE it writes each interval’s output in a Kafka transaction, pre-commits when the barrier arrives and opens a new one, and commits only when the JobManager reports the checkpoint complete. A failure before that aborts the transaction and replays its input; a failure after completion retries the commit.

The price is latency: read_committed consumers (see Kafka) see output only after the commit, roughly one checkpoint interval plus checkpoint duration later. Also say:

• The Kafka sink defaults to DeliveryGuarantee.NONE.
• transaction.timeout.ms must exceed the longest checkpoint plus restart, or Kafka aborts a transaction Flink still means to commit; brokers cap it at 15 minutes by default.
• Each job needs its own transactionalIdPrefix, or jobs fence each other.

Idempotent sinks are often better for aggregates: upsert rows keyed by (ad_id, window_start) into Cassandra, ClickHouse or PostgreSQL. Replays rewrite the same values, results appear as soon as windows fire, and late firings overwrite, provided the output is deterministic.

Choosing the checkpoint interval

The interval trades steady overhead against replay after a failure and, with transactional sinks, output latency. Take 50,000 events/s into a job that can do 200,000 events/s when catching up:

• 60 s interval. A crash just before a checkpoint replays 60 × 50,000 = 3 million events. Spare capacity is 200,000 − 50,000 = 150,000 events/s, so catching up takes 3,000,000 / 150,000 = 20 s, plus restore time; read_committed readers wait up to a minute.
• 10 s interval. Replay 500,000 events, about 3.3 s of catch-up, and readers wait about 10 s, for six times as many checkpoints.

With small state, 10–30 s is cheap; with hundreds of gigabytes, go incremental and keep checkpoint duration well below the interval. As Flink’s production checklist puts it, the interval is an SLA on how much you will reprocess.

How it scales

Parallelism and shuffles

Scale by raising operator parallelism and adding TaskManagers, up to three ceilings. Source parallelism stops at the partition count; extra Kafka source subtasks sit idle and, without an idleness timeout, hold back the watermark. Keyed parallelism stops at max parallelism. And every keyBy serialises each record and usually ships it across the network, so filter and project first. Network buffers flush when full or every 100 ms, the knob between latency and throughput.

Hot keys and skew

keyBy sends every event for a key to one subtask. If one ad takes 30% of the clicks during an IPL final, its subtask does 30% of the work and its backpressure slows everything. Fixes, cheapest first:

• Pre-aggregate before the shuffle and send partial counts; Flink SQL calls this local-global aggregation (mini-batch on).
• Salt and aggregate twice: key by (ad_id, hash(click_id) % 8), then by ad_id to add the 8 partials.
• For distinct counts, SQL’s split-distinct rewrite works the same way, or use HyperLogLog.

Backpressure, and how to read it

A slow subtask stops granting credits (one per free network buffer, since Flink 1.5), so its upstream stops sending, and the pressure travels back to the source, which simply reads Kafka more slowly: lag grows, nothing is dropped. Per subtask, busyTimeMsPerSecond, backPressuredTimeMsPerSecond and idleTimeMsPerSecond add up to about 1,000, and the UI marks a task HIGH when it is back-pressured over half the time. The bottleneck is the first busy task downstream of the back-pressured ones; those are victims, not culprits. See backpressure and load shedding.

Rescaling

Classic rescaling is stop with a savepoint, restart with a new parallelism. The adaptive scheduler rescales as slots come and go, reactive mode uses every TaskManager in the cluster, restoring from the latest checkpoint, and the Kubernetes Operator’s autoscaler sets each operator’s parallelism from its busy time and source backlog. Every rescale is still a restart, so huge state means slow scaling.

Limits and numbers

Defaults are from the Flink 2.3 documentation (2.3.0 shipped in June 2026; 1.20 remains the long-term-support 1.x line).

When to pick it, and when not

Pick Flink when state is large or long-lived, event-time correctness with late data matters, you join or enrich several streams, or one job must cover both history and live data.

Pick something else when it is Kafka in and Kafka out with modest state (Kafka Streams needs no cluster), minutes of latency are fine (a scheduled batch job; see batch vs stream processing), the work is stateless per event (a plain consumer group), or you need to serve queries: Flink computes, so write results to an OLAP store, Redis or a search index.

Common interview usage

Going deeper: failure modes

No output at all: a stalled watermark

One idle Kafka partition (a quiet region at night, or more source subtasks than partitions) pins the minimum watermark, so no window fires while records still flow. Use withIdleness and alert on watermark lag. The reverse, one partition racing ahead in a backfill, bloats buffered state; watermark alignment throttles it.

Late data vanishing quietly

With allowed lateness 0, late events disappear and only the numLateRecordsDropped metric notices, yet offline phones can be hours late. Choose the bound, lateness and a side output deliberately, and reconcile money with a daily batch.

State that grows until the job falls over

Keys that never return (session or click IDs), global windows without a purging trigger, and SQL joins without a state TTL grow forever, slowing checkpoints and recovery until disks fill. Set TTLs, prefer windowed joins, and alert on checkpoint size.

Checkpoints timing out under backpressure

A slow sink fills buffers, barriers queue, checkpoints hit the 10-minute timeout, and the job restarts and falls further behind. Fix the bottleneck, enable unaligned checkpoints and buffer debloating (since 1.14), and tolerate a few failed checkpoints.

Replays that change the answer

Replay assumes the same input gives the same output. Wall-clock reads, random IDs or lookups against a changing table break that: an upsert sink may overwrite a correct row with a different one, and an at-least-once sink (a JDBC insert, Elasticsearch with generated IDs) duplicates everything since the last checkpoint. Derive IDs from the input and upsert on natural keys.

Savepoints that won’t restore, and slow recovery

Changing a UID, a state type or max parallelism makes old state unreadable, and forcing a start with --allowNonRestoredState silently drops it; set UIDs on day one and test restores in CI. Restoring terabytes after each failure also takes minutes: use incremental checkpoints, local recovery, more parallelism, or ForSt on 2.x.

Cheat sheet

How to say it in the interview

“Between the clicks topic and ClickHouse I’d run a Flink job. It reads clicks in event time with a 10-second watermark bound, drops duplicate click IDs using keyed state with a 10-minute TTL, then keys by ad and counts one-minute tumbling windows, with a minute of allowed lateness and a side output for later clicks. RocksDB state is checkpointed to S3 every 30 seconds with the Kafka offsets, so a crash means restore and replay, not double counts. The sink upserts by ad and minute, so replays overwrite. If one ad runs hot, I pre-aggregate before the keyBy.”

What does exactly-once mean in Flink? Each event affects Flink’s state once, because state and source offsets are snapshotted and restored together. Records still re-run after a failure, so output needs a transactional or idempotent sink, and side effects need idempotency keys.

How does Flink know a window is complete? Watermarks: each source asserts “event time has reached t”, usually the highest timestamp minus a bound, and operators take the minimum across inputs. Stragglers within allowed lateness re-fire the window; later ones go to a side output.

What happens when a TaskManager dies? The job restarts from the latest completed checkpoint: operators reload state, sources seek to the stored offsets, and up to one interval of input is replayed.

One ad gets half the clicks. What breaks? Its subtask becomes the bottleneck and backpressure slows the whole job. Pre-aggregate before the shuffle, or salt the key and sum the partials in a second stage.

Why not a consumer that increments Redis counters? Fine at low volume in processing time. It double-counts on replay, has no event-time windows and no consistent snapshot to rebuild from. Flink gives you those, and you pay with a cluster to run.

Where to go next

• Kafka: the log Flink reads and writes.
• Batch vs stream processing: when streaming is worth it.
• Delivery guarantees: at-least-once, exactly-once and DLQs: the semantics in general.
• Change data capture and event sourcing and CQRS: database changes as streams.
• Clocks, ordering and conflict resolution: why event time and wall clocks disagree.
• Design ad click aggregation and design top-K trending: Flink end to end.

References and further reading

1. Apache Flink 2.3 documentation: Stateful Stream Processingnightlies.apache.orgKeyed state and key groups, checkpoint barriers and alignment, unaligned checkpoints, recovery by replay, savepoints, and at-least-once mode.
2. Apache Flink 2.3 documentation: Timely Stream Processingnightlies.apache.orgEvent time versus processing time, the definition of Watermark(t), the minimum across parallel inputs, and lateness.
3. Apache Flink 2.3 documentation: Windowsnightlies.apache.orgWindow assigners, window lifecycle, allowed lateness (default 0), late firings, side outputs, and one copy of each element per window.
4. Apache Flink 2.3 documentation: State Backendsnightlies.apache.orgHashMap (default) versus embedded RocksDB versus the experimental ForSt backend, incremental checkpoints and the 2^31-byte key/value limit.
5. Apache Flink 2.3 documentation: Production Readiness Checklistnightlies.apache.orgThe default max-parallelism formula, why operator UIDs matter, checkpoint interval as an SLA, and JobManager high availability.
6. Apache Flink 2.3 documentation: Kafka connectornightlies.apache.orgKafkaSink delivery guarantees (default NONE), transactions committed on checkpoint, transaction timeouts, idleness and offset commits.
7. Apache Flink 2.0.0: A new Era of Real-Time Data Processing (Flink blog, March 2025)flink.apache.orgDisaggregated state with ForSt and the Nexmark results, plus the removed APIs and the 1.x-to-2.x state incompatibility.
8. The Dataflow Model (Akidau et al., VLDB 2015)vldb.orgWhat, where, when and how; why watermarks are too fast or too slow; triggers and accumulation modes.
9. Lightweight Asynchronous Snapshots for Distributed Dataflows (Carbone et al., 2015)arxiv.orgAsynchronous barrier snapshotting: a Chandy–Lamport variant that stores only operator state on acyclic graphs.
10. Apache Spark 4.2 Structured Streaming Programming Guidespark.apache.orgMicro-batch latency of about 100 ms with exactly-once, and the still-experimental continuous mode (about 1 ms, at-least-once, map-like queries only).