Apache Flink: State, Checkpoints & Exactly-Once

Apache Flink is a stream engine. It reads events one at a time, keeps state, and writes results. Many teams run it on Kafka.

The chapter "Stream Processing" showed the ideas: state, windows, watermarks and checkpoints. This chapter shows how Flink does them, because senior interviews ask how it works inside.

What you already know

• State is what a job remembers between events, such as a running total.
• A watermark decides when a window of event time is done.
• A checkpoint saves the state and the input position together.
• Exactly-once means the result is as if each event was handled once. The sink must help.

What you will learn

1. How a Flink job runs on a cluster.
2. Where state lives, and how to choose the state backend.
3. How Flink takes a checkpoint while data keeps flowing.
4. How it restarts after a crash.
5. Savepoints: how to change a job without losing its state.
6. Two-phase commit: exactly-once into Kafka.
7. Backpressure: what a slow sink does to the whole job.
8. When to choose Flink, and when Spark is enough.

We use the seven shop orders from "Streaming Architecture Fundamentals". The job sums revenue for each customer:

-- For each customer: the total of all their orders so far.

You write a Flink job in SQL, Java or Python. This is the revenue job in Flink SQL:

CREATE TABLE orders (
  order_id    BIGINT,
  customer_id INT,
  amount      DECIMAL(10, 2),
  event_time  TIMESTAMP(3),
  WATERMARK FOR event_time AS event_time - INTERVAL '3' MINUTE
) WITH (
  'connector' = 'kafka',
  'topic' = 'orders',
  'properties.bootstrap.servers' = 'broker:9092',
  'properties.group.id' = 'revenue',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
);

SELECT customer_id, SUM(amount) AS revenue
FROM orders
GROUP BY customer_id;

Flink turns the job into a graph of operators. An operator is one step, such as "read Kafka" or "sum by customer".

Each operator runs as several copies in parallel, called subtasks. The number of copies is the parallelism.

A Flink cluster has two kinds of process:

GROUP BY customer_id sends every order of one customer to the same subtask. That subtask keeps the total for that customer. This is keyed state.

With parallelism 2, the two subtasks share the customers. We use a made-up rule, odd and even, to keep the example easy to check:

Real Flink hashes the key into a key group, and gives each subtask a range of key groups. The number of key groups is the max parallelism.

You can later run with more subtasks, but never more than the max parallelism. Set it on purpose, before the job holds much state.

Flink handles each event as it arrives. Spark Structured Streaming collects events into small batches by default. That is why Flink is often chosen when every millisecond counts.

The state backend decides where a subtask keeps its state while it runs. You choose it per job. Flink 2.3 offers three:

An incremental checkpoint saves only what changed since the last one. With a large state, this makes each checkpoint much smaller.

Say the revenue job has 50 million customers. Their totals may not fit in memory, so RocksDB with incremental checkpoints is the usual choice.

State that never ends

The revenue query keeps one total for every customer, for ever. A new customer adds state, and nothing removes it.

A time to live, TTL, deletes state that has not changed for a while. In Flink SQL you set it for the whole job:

SET 'table.exec.state.ttl' = '30 d';

Flink does not stop the job to take a checkpoint. It sends a marker through the data, called a checkpoint barrier.

The JobManager asks the sources to start checkpoint number n. Each source saves its Kafka offsets and sends barrier n down the stream, between two events.

When an operator gets barrier n, it saves its state and passes the barrier on. Events before the barrier are in checkpoint n. Events after it are not.

Here the source reads the one log of orders. Barrier 1 goes out after offset 2:

source reads:  110  111  112  | barrier 1 |  113  114  150  151
               (offsets 0-2)                 (offsets 3-6)

When every operator has saved its part, the JobManager marks checkpoint 1 complete. Only a complete checkpoint is used for a restart.

Aligned and unaligned checkpoints

An operator with two inputs gets the barrier twice, once on each input. By default it waits for both before it saves. This wait is alignment.

While it waits, events pile up on the faster input. Under backpressure, alignment can take minutes.

An unaligned checkpoint does not wait. It also saves the events that are still in flight. Checkpoints finish faster, but they are larger.

The Kafka source keeps its offsets in the checkpoint. It also commits offsets to Kafka, but only so tools can show lag. Flink does not use them to restart.

Say a TaskManager dies after order 114, before checkpoint 2. Flink restarts the job from checkpoint 1:

Orders 113 and 114 were read twice, but the state counts them once. The state went back to before them in step 4.

The checkpoint interval is a trade. A short interval means less to read again after a crash, but more work all the time.

Many teams start between 30 seconds and a few minutes. You set it with execution.checkpointing.interval.

You will change the job: fix a bug, add a column, or add subtasks. If you start the new job empty, every customer total starts at 0.

A savepoint is a snapshot you take by hand, so the new job can start from it. This table shows how it differs from a checkpoint:

An upgrade has three steps. Stop with a savepoint, then start the new version from it:

# 1. Stop the job and write a savepoint
./bin/flink stop --savepointPath s3://shop-flink/savepoints <job-id>

# 2. Deploy the new version from that savepoint
./bin/flink run --fromSavepoint s3://shop-flink/savepoints/savepoint-<id> revenue-job-v2.jar

# 3. Check the totals: c2 should still be 48.00

Flink must match each saved piece of state to an operator in the new job. It matches by operator ID.

If you do not set the IDs, Flink makes them from the shape of the job. Then a small code change can change them. In the DataStream API, set uid() on every operator that has state.

The state is safe after a crash. The output is the hard part. Flink makes a Kafka sink exactly-once with a two-phase commit.

The sink writes results inside a Kafka transaction. Phase 1 happens when the barrier reaches the sink: it flushes the transaction but does not commit it.

Phase 2 happens when the whole checkpoint is complete. Then the sink commits the transaction. This table follows checkpoint 2:

If the job crashes before step 3, the checkpoint is not complete. Flink restarts from checkpoint 1, and T2 is aborted. Readers never see it.

If the job crashes after step 3 but before step 4, the checkpoint holds the ID of T2. On restart, Flink commits T2 again, so its results are not lost.

In code, the Kafka sink needs two settings. This is the Java DataStream API:

KafkaSink<String> sink = KafkaSink.<String>builder()
    .setBootstrapServers("broker:9092")
    .setRecordSerializer(KafkaRecordSerializationSchema.builder()
        .setTopic("revenue_by_customer")
        .setValueSerializationSchema(new SimpleStringSchema())
        .build())
    .setDeliveryGuarantee(DeliveryGuarantee.EXACTLY_ONCE)
    .setTransactionalIdPrefix("revenue-job")          // unique per job on the cluster
    .setProperty("transaction.timeout.ms", "900000")  // 15 minutes
    .build();

Two costs follow from this design:

• Readers see results only after each checkpoint. With a checkpoint every minute, results are up to a minute late.
• Kafka aborts a transaction that stays open longer than transaction.timeout.ms. Set it longer than your longest checkpoint plus your longest restart.

The broker caps the timeout with transaction.max.timeout.ms, 15 minutes by default. The producer default is only 60 seconds, so set it on purpose.

A sink outside Kafka, such as a database, cannot join this Kafka transaction. Write to it with an upsert on the key, so a repeat replaces the row.

Backpressure means a slow operator makes the operators before it slow down. In Flink it travels back to the source.

Each operator sends data only when the next one has room for it. So nothing is lost, but the whole job runs at the speed of the slowest step.

Here the sink writes to a database that answers slowly. The Flink web UI shows each operator as busy, backpressured or idle:

The slow step is the first operator that is busy and not backpressured. Here it is the sink.

Backpressure shows in two other places. Kafka lag grows, because the source reads slowly. Aligned checkpoints get slower, because barriers wait in full buffers.

Fix the slow step: batch the database writes, add sink subtasks, or make the database faster. Unaligned checkpoints help the checkpoints finish, but the job is still slow.

Both engines can be exactly-once inside, and both need the sink to help. The difference is in how they run and what your team knows:

Choose Flink when you need low delay, complex event-by-event logic, or very large state. Choose Spark when your team already runs Spark and seconds of delay are fine.

For Spark code, read "Structured Streaming Fundamentals" in the Apache Spark section.

Build each scenario on the pipeline canvas with the chapter closed. In each one, say where the state lives, how often you checkpoint, and how the sink stays exactly-once.