This article is adapted from my post on ByteDance’s engineering account, Practicing Single-Task Recovery in Flink at ByteDance.

At ByteDance, Flink joins user features and behavior in real time. The resulting instances update downstream recommendation models and affect users’ recommendations. The latency and stability of this joining service directly influence product quality. Implemented as something like a two-stream join, it traditionally triggers failover of the entire job when any task or node fails, affecting recommendations for all users or articles in the product.

To address this, we proposed single-task recovery. Enhancements to the network layer allow the following behavior after a machine or task failure, at the cost of losing some data associated with the failed task:

• Only failed tasks restart; the job does not restart globally.
• Healthy tasks continue supplying joined data downstream without interruption.

Approach

We asked whether a machine failure could restart only the tasks on that machine, while their upstream and downstream neighbors detected the failure and responded appropriately:

• Upstream: Discard data destined for failed tasks until recovery completes, then resume sending.
• Downstream: Clear incomplete data from failed tasks, then reconnect and receive data after recovery.

This led to three key questions:

• How do neighboring tasks detect a task failure?
• How is incomplete downstream data cleared?
• How are connections reestablished after recovery?

Existing Design

Note: Our implementation is based on Flink 1.9. The network model may differ from 1.11 onward, when unaligned checkpoints were introduced.

First, let us simplify the communication model between upstream and downstream tasks:

There are three cases for detecting neighboring task state changes:

• A task fails due to a logic error or OOM: It proactively releases network resources, sends a channel-close message upstream, and an exception downstream.
• YARN kills the TaskManager process: The operating system closes TCP connections normally, allowing upstream Netty servers and downstream Netty clients to detect the change.
• A machine loses power: The operating system cannot close TCP connections properly, so Netty clients and servers may not detect the failure. We force certain updates after deploying the replacement task.

In most cases, tasks can directly detect neighboring failures. With the communication model established, we can examine the sending- and receiving-side changes needed for single-task recovery.

Upstream Sending-Side Changes

Here is the sending side in more detail:

1. After receiving a client’s partition request, the Netty server registers a SubpartitionView and reader on the corresponding subpartition.
2. RecordWriter sends data to subpartitions. Each maintains a buffer queue and puts its reader in Readers Queue. (Task thread.)
3. The Netty thread takes a reader from Readers Queue, reads buffers from its subpartition, and sends them downstream. (Netty thread.)

We want upstream tasks to discard data destined for a failed downstream task. When a subpartition receives the failure notification through the Netty server, it marks itself Unavailable. RecordWriter checks this flag and discards records targeting it. After the recovered task reconnects, the subpartition becomes Available and consumption resumes.

Flink’s subpartition abstraction and configurable initialization make this straightforward. Following PipelinedSubpartition, we implemented a custom subpartition and corresponding view with this behavior.

Downstream Receiving-Side Changes

Now consider the receiving side:

Its threading model is similar to the sending side:

1. InputGate initializes all channels and connects to upstream servers through Netty clients.
2. InputChannel queues received buffers and puts a reference to itself in Channels Queue. (Netty thread.)
3. and 4. When called by InputProcessor, InputGate takes a channel from the queue and reads its buffered data. Incomplete data, such as half a record, is retained temporarily in InputProcessor. (Task thread.)

After an upstream task fails, we want to remove incomplete data from its InputChannel. In the diagram, this means clearing the specified channel’s buffer in InputProcessor.

When the Netty client detects an upstream failure, it notifies the InputChannel, which appends an Unavailable Event to its buffer queue. When InputProcessor reaches that event, it clears the incomplete buffer it is holding.

JobManager Restart-Strategy Changes

The restart strategy can follow the community’s RestartIndividualStrategy. The key difference is that, after deploying the replacement task, we use ExecutionGraph’s topology to find downstream tasks and invoke RPCs telling them to reconnect to the new upstream task.

ByteDance also implemented reserved TaskManagers. Failed tasks can immediately use these resources, substantially reducing data loss during recovery.

Implementation Details

The overall idea is straightforward, but implementation introduced several subtleties:

• When JobManager updates downstream channel information through RPC, the old channel’s Unavailable Event may not yet have reached InputProcessor, so the channel cannot close immediately. We cache the new partition information, drain the old channel, close it, and then initialize the replacement using the cached information.
• The network layer has one of Flink’s most complex threading models. To limit complexity and risk, we reused existing communication between Netty and task threads. However, adding availability flags to subpartitions and channels requires particular care with memory visibility, preventing threads from observing inconsistent states.

A quick recruiting note ^_^: ByteDance’s stream-processing engine team is hiring. Beyond network optimizations, we have developed improvements for our workloads in SQL, scheduling, checkpointing, and unified batch/stream processing. Come join us!