Stream Processing (4) - Failover in Flink

Job restarts and failover are so familiar that restarting is many developers’ first response to trouble. Understanding failover is essential for deeper diagnosis and for recognizing Flink’s limitations.

Other articles in this stream-processing series:

• Stream Processing (1) - Why Not RocksDB in Streaming State
• Stream Processing (2) - Flink Fault Tolerance and Weakly Consistent Snapshots
• Stream Processing (3) - A Rule System Based on Flink CEP
• Stream Processing (4) - Failover in Flink

Preface

This article examines how failover arises and raises questions about it, rather than revisiting the well-covered topic of recovery guarantees. Failover may seem like a routine exception followed by automatic recovery. Usually it is, but consider:

• A Flink job drives real-time marketing notifications through CEP. Every failover delays campaigns and affects revenue.
• A midnight sales event relies on live dashboards. A failover freezes metrics for minutes, and checkpoint rollback may produce confusing changes.

These latency-sensitive cases go beyond the usual BI-dashboard workload. Jobs are highly sensitive to failover, yet developers often feel unable to prevent its impact.

How Should We View Failover?

Our attitude determines Flink’s application boundaries. If business losses from failover are accepted as inevitable, these applications may need another engine or hot-standby and degradation plans.

I prefer to view those losses as a stream-engine design problem. Can distributed streaming overcome them completely? There are two broad approaches:

• Accept failover and use high availability to protect downstream consumers.
• Reduce its impact until data interruption approaches zero.

HA

Hot standby is straightforward in principle: when one job fails over, another takes over so downstream data does not stop for long.

In practice, several questions make this difficult:

1. How are source offsets controlled?
2. How is failover detected quickly?
3. How are loss and duplication avoided during switching?
4. …

A more practical approach decouples the whole pipeline: two isolated Flink jobs write separate tables, and the serving layer reads both, using timestamps and values for resilience. This also has deployment challenges, beyond the scope here.

Reducing Failover Cost

Architecture changes can reduce the cost. For a map-only job with parallelism 10,000, splitting it into ten jobs of parallelism 1,000 reduces the failure exposure and recovery duration of each job substantially.

Can we find a more general solution beyond such special cases?

Mechanisms

First, we need to understand the internal failover process.

A Complete Flink Topology

The diagram shows a window-aggregation job with five tasks on three TaskManagers connected all-to-all. Any TaskManager failure triggers global failover.

Global failover redeploys all tasks and downloads and restores snapshots, which is expensive. Two common paths lead from failure to JobManager redeployment:

• A TaskManager detects a neighboring failure and notifies JobManager, often producing connection-lost errors.
• A task fails and notifies JobManager, where its exception stack can usually be found.

To understand how to reduce failover cost, first examine communication between Flink components and how a failure propagates.

Between TaskManagers

The simplified diagram shows the following; see A Deep-Dive into Flink’s Network Stack for details:

• Netty server: Pulls data from buffers populated by sending tasks and transmits it to clients.
• Netty client: Receives data and places it in each receiving task’s dedicated buffer queue.

TaskManagers communicate over TCP. Problems usually arise from full GC, unresponsive nodes, or process failures, and fall into two categories:

• Transport layer: TCP connection errors, including connection refusal. Netty propagates these to the application layer.
• Application layer: TaskManagers exchange messages such as PartiionRequest requests, which supply metadata needed for upstream transmission. PartitionNotFoundException is an example of this category.

Can every transport error be detected correctly? Not necessarily. Many ordinary process terminations release resources in order, including TCP connections, allowing Netty to surface the error.

Abrupt power loss does not release resources normally. TCP’s keep-alive mechanism can leave the peer retaining the connection and apparently hanging. TaskManager-JobManager communication must handle this case.

Between TaskManager and JobManager

Failover-related messages fall into two categories:

• Heartbeats
• Task-state updates

JobManager exchanges periodic heartbeats with every TaskManager. Flink uses Akka RPC, and many JobManager operations run on one thread, similarly to Spark’s EventLoop. Heavy work such as failover can therefore queue RPC messages in Akka and cause heartbeat timeouts.

Task-state synchronization is more involved. See the lifecycle in Jobs and Scheduling:

When a task moves from RUNNING to FAILED, its TaskManager sends JobManager the exception. JobManager identifies affected tasks under the configured strategy and cancels them. Only after all involved tasks reach FAILED or CANCELLED does it begin scheduling and deploying replacements.

Failover Strategy

Flink defaults to region failover, dividing tasks into independent regions based on topology. Any task failure redeploys the entire region.

This is safe but somewhat coarse. All tasks restore the last checkpoint and replay input. In a region of N tasks, one failure forces the other N-1 to repeat consumption and computation.

At large scale, failover can limit horizontal growth. As resources move to the cloud, unreliable scalability also creates substantial manual operational work.

Other Industry Approaches

Distributed systems must consider failover costs. MillWheel, Ray, and RisingWave offer useful comparisons:

MillWheel: Fault-Tolerant Stream Processing at Internet Scale

MillWheel, widely used at Google in earlier years, recovers at task or even record granularity, largely avoiding redundant computation from region-wide failover.

A record entering a MillWheel operator follows these steps:

1. Check its unique ID for duplicates and discard it if already processed.
2. Execute user logic, potentially changing state, timers, and downstream output.
3. Persist state changes and pending output records in state.
4. Downstream acknowledgments allow pending-output records to be removed.

Every record’s changes are reflected in the state store to support exactly-once and fine-grained recovery. In practice, checkpoints persist asynchronously at second-level intervals rather than writing every change immediately to remote storage.

Frequent state-store changes are expensive. MillWheel therefore offers weak productions for workloads that do not require exactly-once, with users responsible for idempotent processing.

Ray / Mobius

Ant built the Mobius AI engine on Ray. Its key features include:

1
Single Node Failover. We designed a special failover mechanism that only
2
needs to rollback the failed node it's own, in most cases, to recover the
3
job. This will be a huge benefit if your job is sensitive about failure
4
recovery time. In other frameworks like Flink, instead, the entire job
5
should be restarted once a node has failure.

Its approach resembles MillWheel: acknowledgments on the sender and state-buffered data on the receiver provide redundant storage for single-task recovery.

Other systems:

RisingWave, like Flink, uses Chandy-Lamport-style snapshots and can restart an entire job after one task fails.

I previously implemented single-task recovery in Flink by adjusting each stage and adding safeguards. It worked well, but more edge cases appeared over time:

• TaskManager A loses its connection to B while retaining its connection to JobManager.
• Too many simultaneous failures leave newly deployed tasks unable to connect upstream, producing cascading failures.

Patching these cases introduces more complexity and new cases. The underlying problem is the absence of a global arbiter: TaskManagers independently infer failures from neighboring TCP connections. JobManager could be redesigned to coordinate this, but it would require restructuring the entire failover path and significantly changing Flink’s architecture.

No Silver Bullet

There is no silver bullet for the performance-versus-recovery tradeoff. MillWheel’s reliance on state and backend storage sounds robust, but maintaining low latency, throughput, and horizontal scalability under frequent state access is also difficult. The paper does not resolve those questions, and practical deployments likely require substantial additional work.

Conclusion

Failover limits Flink’s application scope to some extent. This article examined TaskManager and JobManager communication through practical cases, compared other systems, and shared lessons from implementation.