Asynchronous Barrier Snapshots

Distributed systems lack a global clock reliable enough to coordinate consistent snapshots across hundreds of operators simultaneously. A naive approach to fault tolerance might involve a "stop" mechanism where all processing halts, state is flushed to disk, and processing resumes. In high-throughput streaming environments, this introduces unacceptable latency spikes and significantly degrades performance.

Apache Flink addresses this challenge using Asynchronous Barrier Snapshots (ABS). This mechanism allows the system to create a consistent image of the global state without halting continuous data processing. ABS is a variation of the Chandy-Lamport distributed snapshot algorithm, tailored specifically for the directed acyclic graph (DAG) topology of a streaming job.

The Role of Checkpoint Barriers

The core component of the ABS algorithm is the checkpoint barrier. Unlike standard data records, barriers are control messages injected into the data stream by the source operators. These barriers divide the unbounded stream into logical epochs. All records preceding barrier $n$ belong to the snapshot $n$, while records following the barrier belong to snapshot $n+1$.

Barriers flow downstream alongside data records, strictly preserving the order of events (FIFO) within each channel. Because barriers occupy a specific position in the stream, they trigger state persistence exactly when all modifications related to the pre-barrier data are complete.

Diagram of the checkpoint barrier propagation through a simple operator topology. The barrier separates data belonging to the current checkpoint epoch from the next.

Barrier Alignment and Exactly-Once Semantics

When an operator receives data from multiple input channels (for example, a CoProcessFunction or a KeyedWindow), it must ensure that the snapshot reflects a consistent view across all inputs. This necessitates a process known as barrier alignment.

The alignment process functions as follows for an operator with inputs $I_1$ and $I_2$:

1. Arrival: The operator receives barrier $n$ from input channel $I_1$.
2. Blocking: The operator stops processing data from $I_1$. It cannot process any data from epoch $n+1$ on this channel until the snapshot is taken, otherwise, the state would mix data from two different epochs.
3. Buffering: While $I_1$ is blocked, the operator continues to process data from input $I_2$.
4. Completion: Once barrier $n$ arrives from $I_2$, both inputs are aligned.
5. Snapshotting: The operator emits the barrier to its downstream outputs and triggers the asynchronous snapshot of its local state.
6. Resumption: The operator unblocks $I_1$ and resumes processing data from both channels for epoch $n+1$.

The alignment phase guarantees exactly-once processing semantics. If the system failed and restored from this snapshot, every record before the barrier was processed exactly once, and no record after the barrier was processed prematurely.

Mathematically, let $R_{i}$ be the sequence of records received on channel $i$. The state $S$ at checkpoint $C_n$ satisfies:

$S(C_n) = f(\{r \in R_{i} \mid \text{index}(r) < \text{index}(\text{Barrier}_n) \forall i\})$

Where $f$ represents the state transition function. This implies that the state reflects strictly the consumption of the prefix of the streams up to the barrier.

The Trade-off: Latency and Backpressure

Barrier alignment introduces a latency trade-off. During the alignment phase, the operator effectively pauses consumption of the faster stream ($I_1$ in the previous example) to wait for the slower stream ($I_2$). In topologies with high data skew or significant network jitter, this waiting period manifests as backpressure that propagates upstream.

If the alignment time takes $t_{align}$, the total end-to-end latency for records in the blocked channel increases by at least $t_{align}$. For applications requiring ultra-low latency where occasional duplicate processing is acceptable upon recovery, Flink allows configuring Unaligned Checkpoints. In this mode, barriers overtake data records in the buffers, and the inflight data is persisted as part of the checkpoint state, bypassing the alignment delay at the cost of larger checkpoint sizes.

Asynchronous State Snapshotting

Once barriers are aligned (or immediately upon arrival for single-input operators), the operator performs the state snapshot. To prevent blocking the main processing thread, Flink separates the snapshot process into two phases:

1. Synchronous Phase: The operator performs a lightweight, local copy of the state. For heap-based backends, this might be a copy-on-write operation on the hash table. For RocksDB, this involves forcing a flush of the MemTable to disk and creating a hard link to the SSTables. This phase stops processing, but typically completes in milliseconds.
2. Asynchronous Phase: A background thread transfers the state files to durable remote storage (such as S3, HDFS, or GCS). The main stream processing thread resumes immediately after the synchronous phase is complete.

This separation ensures that the cost of transferring large state sets over the network does not penalize the throughput of the streaming job.

Visualization of the state size growth over time relative to barrier intervals. Efficient snapshotting minimizes the "Stop-Time" (synchronous phase).

Distributed Coordination

The checkpoint coordinator, running on the JobManager, orchestrates the entire lifecycle. It triggers the sources to inject barriers and collects acknowledgments from all tasks. A checkpoint is considered complete only when all operators in the DAG have successfully acknowledged their state handle for that specific barrier ID.

If an operator fails to acknowledge the snapshot (due to network failure or disk I/O errors), the coordinator marks checkpoint $n$ as failed and garbage collects any partial data. The stream processing continues uninterrupted, relying on the next successful checkpoint interval to establish a valid recovery point. This optimistic strategy ensures that transient infrastructure failures do not derail the real-time pipeline, provided that the interval between successful checkpoints remains within the limits defined by the service level agreement (SLA) for data loss recovery.