Flink Fault Tolerance and Weakly Consistent Snapshots

This article discusses Flink’s fault tolerance, common problems, and weak-consistency use cases and solutions encountered in production.

Global State

Flink checkpointing adapts Chandy-Lamport with performance improvements described in Lightweight Asynchronous Snapshots for Distributed Dataflows.

The paper frequently mentions global state. In theory, this is the union of all nodes’ states at one moment. In a distributed system, networks, loads, I/O, and even clocks differ, so literally capturing every node at the same instant is impossible.

Flink Checkpointing

Like Chandy-Lamport markers, Flink uses barriers. JobMaster triggers source checkpoints by RPC. Sources snapshot themselves and broadcast barriers downstream; each task snapshots after receiving all input barriers.

Processing Guarantees

There are two scopes:

• Flink processing semantics: Computation within Flink, concerning processing-operator state.
• End-to-end semantics: Source ingestion, Flink computation, and sink writes, concerning all three kinds of state.

Exactly-Once

Exactly-once: Each input affects the result once. A count() job receiving ten events should produce count=10. The mechanisms differ by scope:

• Flink processing: After its first barrier, a task begins alignment. It buffers and stops consuming inputs whose barriers have arrived, while continuing to consume others until all barriers arrive. Its snapshot then aligns logically with the source snapshots.

• End-to-end: Two-phase commit extends processing guarantees across sources and sinks:

  1. Begin transaction: Each checkpoint commits a transaction, so the interval between checkpoints forms one transaction.
  2. Pre-commit: Source and sink snapshots prepare the transaction.
  3. Commit: Commit when JobMaster sends its notification callback.
  4. Abort: Abort on failure.

At-Least-Once

At-least-once: An input may affect results more than once. Ten inputs may produce count>10. The mechanisms are:

• Flink processing: Unlike exactly-once, a task continues consuming an input after its barrier arrives. Suppose Kafka source index 1 emits a barrier at offset 100, with downstream count=100. Continuing to consume before other barriers arrive can produce a snapshot with count>100. Recovery restores the source at offset 100 but the downstream task at the larger count, duplicating data.
• End-to-end: Exactly-once requires transactional sources and sinks. Transactions are common in OLTP, but support in big-data components, including Kafka and other AP stores, is limited in the context discussed here. Many accuracy-sensitive applications combine Flink exactly-once processing with idempotent sinks to approximate end-to-end exactly-once.

Weakly Consistent Snapshots

Barriers elegantly align distributed task snapshots. In practice, however, strong consistency involves tradeoffs.

Motivation

End-to-end exactly-once is difficult because both ends need transactions, potentially reducing performance at high traffic volumes. At-least-once introduces duplicates unless the sink provides idempotent writes.

At-most-once does not guarantee that every record is processed. Disabling checkpoints approximates this behavior, but long-window applications can then lose an unacceptable amount of data.

Consider three scenarios:

• Scenario 1: Forward behavior events into separate Kafka topics for downstream consumers. Real-time deduplication is difficult, so consumers prefer exact data. When exactly-once is unavailable, minimizing both loss and duplication reduces later modeling inaccuracies.
• Scenario 2: Join features and user behavior into positive/negative training samples. Excess duplicates bias real-time models; excess loss leaves them undertrained. Again, both should be minimized when exactly-once is unavailable.
• Scenario 3: Show video view counts to creators. A Lambda architecture often gives approximate live figures corrected the next day. Duplicates distort earnings expectations; downward corrections are worse for creators than upward corrections, so duplication can cost more than loss.

These applications tolerate some loss but neither extensive loss nor extensive duplication. At high volumes, existing checkpoints create problems:

• Replay: Minute-scale checkpoint intervals mean a failed checkpoint can force several minutes of replay. Repeated failures under unstable conditions greatly increase duplicate volume.
• Checkpoint success rate: One task’s snapshot failure fails the global snapshot. Overall success is roughly Math.pow(single-task success rate, task count), so large jobs become more vulnerable to environmental fluctuations.
• Performance: Even mild skew under exactly-once creates alignment backpressure and briefly idles many tasks. CPU utilization can fluctuate periodically with checkpoints. Community optimizations trade other resources for shorter checkpoints, not always appropriately, for example:
  • FLIP-76 Unaligned Checkpoints: Space for time.
  • FLIP-158 Generalized Incremental Checkpoints: I/O resources for time.

Strong consistency makes connected operators advance and fail together: snapshots must succeed together, and failover is coordinated. This makes low duplication difficult. When ideal end-to-end exactly-once is unattainable and limited loss is acceptable, could we instead optimize for minimal loss? Could relaxing snapshot consistency improve performance and stability while avoiding replay?

The Weakly Consistent Approach

Moving from strong to weak consistency is more than changing exactly-once to at-most-once. It trades consistency for flexibility, while still needing to solve the original operational problems.

Strong snapshots use barriers as physical markers of a logical moment. Weak snapshots can remove them. Imagine this mechanism:

JobMaster directly triggers snapshots on stateful non-source operators. The rest resembles native checkpointing: tasks finish snapshots and acknowledge by RPC, and JobMaster stores checkpoint metadata in the distributed filesystem.

Removing barriers solves alignment, but introduces a question: How do we minimize data loss? There are two parts:

• Reduce loss per operator: Without source snapshots, loss is approximately the input received between the last successful checkpoint and failure. Shorter intervals and higher checkpoint success rates help.
• Reduce the number of affected operators: Strong consistency requires coordinated restarts. Weak consistency removes that requirement, allowing a nonglobal restart strategy.

Implementation Challenges

The simplified mechanism is easy to describe but difficult to make comprehensive:

• Records can depend on one another. Losing a record that triggers a window may prevent the entire window from emitting, effectively losing all its data downstream.
• Some operators depend on timers and events. How should those mechanisms be supported?
• How can a task fail over without affecting its neighbors? FLIP-135 addresses this, also for model-training scenarios.
• How can one failed operator snapshot avoid failing the others? Independent snapshots need not all succeed together, but partial success complicates restoration and JobMaster-task coordination.

Supporting fixed execution patterns such as map-only or two-stream joins is manageable in production. Supporting every operator and DAG would require a much more fundamental redesign.

Summary

This article connected Flink’s fault tolerance with processing guarantees and explained why strong snapshots may not meet some production needs. Relaxing consistency can provide useful flexibility. The proposed weak-snapshot approach and its challenges come from practical production experience.