Learn Apache Flink - State Management & Checkpointing

#Introduction

Episode 5 introduced windowing that stores temporary aggregations. Episode 6 brings that to the surface: state is one of Flink's biggest differentiators compared to other streaming engines. Almost all useful analytics — totals per user, pattern detection, recommendations — require the ability to remember what has already happened.

We'll dissect the types of state, how to store and access it via ValueState, ListState, MapState, and AggregatingState, then configure checkpointing to guarantee exactly-once. Finally, restart strategies make jobs resilient: if a TaskManager dies, the job recovers on its own.

#Stateful Operators and Keyed State

#When an Operator Stores State

An operator is considered stateful if it stores data between incoming elements. The simplest example: a sum operator inside a window, or an operator that computes a running total per user. Without state, both are impossible — every event is processed without any memory.

#Keyed State vs Operator State

Flink distinguishes two families of state:

• Keyed state: state scoped to one key, used after a keyBy operator. For example, the total transactions per userId.
• Operator state: state attached to one operator subtask, regardless of key. For example, the buffer list a source uses while tracking consumption positions.

keyed state   → per key, accessed from a keyed context
operator state → per subtask, shared among related operators

#State Types in Flink

#ValueState

ValueState stores a single value per key. Suitable for tracking the latest value or a simple accumulator:

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
 
ValueStateDescriptor<Long> desc =
    new ValueStateDescriptor<Long>("total", Long.class);
 
ValueState<Long> total = getRuntimeContext().getState(desc);
Long lama = total.value();
total.update((lama == null ? 0L : lama) + event.getAmount());

total.value() retrieves the current value, and total.update overwrites it. This read-write-update pattern is the main idiom of keyed state.

#ListState and MapState

ListState stores a collection of elements; MapState stores key-value pairs. Both are used when state must hold more than one entity:

ListStateDescriptor<String> riwayatDesc =
    new ListStateDescriptor<String>("riwayat", String.class);
 
ListState<String> riwayat = getRuntimeContext().getListState(riwayatDesc);
riwayat.add(event.getId());

MapStateDescriptor<String, Double> hargaDesc =
    new MapStateDescriptor<String, Double>(
        "harga", String.class, Double.class);
 
MapState<String, Double> harga = getRuntimeContext().getMapState(hargaDesc);
harga.put("apel", 15000.0);
double apel = harga.get("apel");

getListState and getMapState follow the same pattern: declare a descriptor, then get the handle from the runtime context. Choose ListState when order matters, MapState when fast lookup by sub-key is needed.

#AggregatingState

AggregatingState merges all values into a single aggregate result without storing the original elements — efficient for sums, averages, or counts:

AggregatingStateDescriptor<Order, Long, Long> aggDesc =
    new AggregatingStateDescriptor<Order, Long, Long>(
        "total-agg",
        new AggregateFunction<Order, Long, Long>() {
            public Long createAccumulator() { return 0L; }
            public Long add(Order value, Long acc) { return acc + value.getAmount(); }
            public Long getResult(Long acc) { return acc; }
            public Long merge(Long a, Long b) { return a + b; }
        },
        Long.class);
 
AggregatingState<Order, Long> agg =
    getRuntimeContext().getAggregatingState(aggDesc);

The AggregateFunction inside the descriptor defines how to initialize, add, get the result, and merge accumulators — this function determines the aggregation logic.

#Checkpointing Configuration

#Automatic Checkpoints

A checkpoint is a periodic snapshot of all state plus the source consumption position. To make it automatic, enable an interval in config.yaml:

execution.checkpointing.interval: 5min
execution.checkpointing.mode: exactly-once
execution.checkpointing.min-pause: 1min
execution.checkpointing.timeout: 10min
execution.checkpointing.tolerable-failed-checkpoints: 2

execution.checkpointing.mode chooses between exactly-once and at-least-once. min-pause guarantees checkpoints don't stack on top of each other, and tolerable-failed-checkpoints tolerates minor failures before the job is considered failed.

#Consistency Guarantees

Flink's checkpointing is based on distributed snapshots (an adapted Chandy-Lamport algorithm). When a process dies, Flink restores state to the last checkpoint and replays events from the stored source position. Because events are reprocessed, the exactly-once guarantee is achieved as long as the sink also supports two-phase commits (like the Kafka sink) or is idempotent.

#Choosing a State Backend

State is stored in a state backend:

state.backend.type: rocksdb
state.backend.incremental: true
state.backend.local-recovery: true

The hashmap backend stores state in heap memory — fast but limited. The rocksdb backend stores it on local disk with an in-memory cache — capable of holding huge state. Incremental checkpoints make snapshots lighter because they only send changes since the last checkpoint.

#Restart Strategies and Fault Tolerance

#Restart Strategies

The restart strategy determines how a job recovers after failure. In config.yaml:

restart-strategy.type: fixed-delay
restart-strategy.fixed-delay.attempts: 3
restart-strategy.fixed-delay.delay: 10s

There's also exponential-delay mode, which delays restarts progressively longer on each attempt — suitable for production so downstream systems aren't flooded with retries. The combination of restart strategy and checkpointing is what makes Flink self-healing.

#Monitoring Checkpoints

Monitor checkpoint status from the REST API or the dashboard:

curl -s http://localhost:8081/jobs/overview
curl -s http://localhost:8081/jobs | python3 -m json.tool

The curl -s http://localhost:8081/jobs command shows the job list as JSON. In the dashboard, the Checkpoints tab shows snapshot history: state size, duration, and SUCCESS or FAILED status.

#Conclusion

Episode 6 equipped you with the foundation of reliability: understanding keyed state and operator state, mastering ValueState, ListState, MapState, and AggregatingState, configuring checkpointing with exactly-once mode, choosing the right state backend, and setting up a restart strategy so jobs recover on their own.

The key takeaways:

• Keyed state is attached to a key after keyBy; operator state is attached to a subtask.
• ValueState, ListState, MapState, and AggregatingState are chosen according to the shape of the state data.
• Automatic checkpointing guarantees exactly-once with state snapshots and source positions.
• The hashmap backend uses heap memory; the rocksdb backend uses disk and suits large state.
• A restart strategy determines how a job recovers after failure, complementing checkpointing.

In the next episode, episode 7, we'll discuss error handling and debugging stream jobs — handling exceptions in operators and sources, monitoring job status and logs, using savepoints for recovery, and debugging with a local cluster and flink run -d. These skills are what separate engineers who write code from those who operate systems.