字节跳动技术团队

背景

在字节跳动的实时计算场景中，我们有很多任务（数量 2k+）会直接服务于线上，其输出时延和稳定性会直接影响线上产品的用户体验，这类任务通常具有如下特点：

  • 流量大，并发高（最大的任务并行度超过 1w）
  • 拓扑类似于多流 Join，将各个数据源做整合输出给下游，不依赖 Checkpoint
  • 没有使用 Checkpoint 并且对短时间内的小部分数据丢失不敏感（如 0.5%），但对数据输出的持续性要求极高

在 Flink 现有的架构设计中，多流 Join 拓扑下单个 Task 失败会导致所有 Task 重新部署，耗时可能会持续几分钟，导致作业的输出断流，这对于线上业务来说是不可接受的。

针对这一痛点，我们提出单点恢复的方案，通过对 network 层的增强，使得在机器下线或者 Task 失败的情况下，以短时间内故障 Task 的部分数据丢失为代价，达成以下目标：

  • 作业不发生全局重启，只有故障 Task 发生 Failover
  • 非故障 Task 不受影响，正常为线上提供服务

解决思路

当初遇到这些问题的时候，我们提出的想法是说能不能在机器故障下线的时候，只让在这台机器上的 Tasks 进行 Failover，而这些 Tasks 的上下游 Tasks 能恰好感知到这些失败的 Tasks，并作出对应的措施：

  • 上游：将原本输出到 Failed Tasks 的数据直接丢弃，等待 Failover 完成后再开始发送数据。
  • 下游：清空 Failed Tasks 产生的不完整数据，等待 Failover 完成后再重新建立连接并接受数据

根据这些想法我们思考得出几个比较关键点在于：

  • 如何让上下游感知 Task Failed ?
  • 如何清空下游不完整的数据 ?
  • Failover 完成后如何与上下游重新建立连接 ?

基于以上考虑我们决定基于已有的 Network 层线程模型，修改上下游对于 Task Failed 后的处理逻辑，让非故障的 Tasks 短时间内完成对失败 Task 的感知操作，从而使得作业持续稳定地输出。

当前架构

注：我们的实现基于 Flink-1.9，1.11 后的网络模型加入了 Unaligned Checkpoint 的特性，可能会有所变化。

我们先将 Flink 的上下游 Task 通信模型简单抽象一下：

Figure. 上下游通信模型上下游 Task 感知彼此状态的逻辑，分三种情况考虑：

  • Task 因为逻辑错误或 OOM 等原因 Fail，Task 自身会主动释放 network resources，给上游发送 channel close 信息，给下游发送 Exception。
  • TaskManager 进程被 Yarn Kill，TCP 连接会被操作系统正常关闭，上游 Netty Server 和下游 Netty Client 可以感知到连接状态变化。
  • 机器断电宕机，这个情况下操作系统不会正确关闭 TCP 连接，所以 Netty 的 Server 和 Client 可能互相感知不到，这个时候我们在 deploy 新的 Task 后需要做一些强制更新的处理。

可以看到，在大部分情况下，Task 是可以直接感知到上下游 Task 的状态变化。了解了基础的通信模型之后，我们可以按照之前的解决思路继续深入一下，分别在上游发送端和下游接收端可以做什么样改进来实现单点恢复。

优化方案

根据我们的解决思路，我们来绘制一下单个 Task 挂了之后，整个 Job 的通信流程：

Figure. 单点恢复流程Map(1) 失败之后：

  1. 将 Map(1) 失败的信息通知 Source(1) 、Sink(1) 和 JobManager。
  2. JobManager 开始申请新的资源准备 Failover，同时上游 Source(1) 和下游 Sink(1) 切断和 Map(1) 的数据通道，但是 Source(1) 和 Sink(1) 和其他 Task 的数据传输仍正常进行。
  3. Map(1)’ 被成功调度，和上游建立连接，JobManager 通知 Sink(1) 和 Map(1)’ 建立连接，数据传输通道被恢复。

从这个流程，我们可以将优化分为三个模块，分别为上游发送端、下游接收端和 JobManager。

上游发送端的优化

我们再细化一下上游发送端的相关细节，

Figure. 上游数据发送流程(1) Netty Server 收到 Client 发送的 Partition Request 后，在对应的 Subpartition 注册读取数据的 SubpartitionView 和 Reader。

(2) RecordWriter 发送数据到不同的 Subpartitions，每个 Subpartition 内部维护一个 buffer 队列，并将读取数据的 Reader 放入到 Readers Queue 中。（Task 线程）

(3) Netty 线程读取 Readers Queue，取出对应的 Reader 并读取对应 Subpartition 中的 buffer 数据，发送给下游。（Netty 线程）

我们的期望是上游发送端在感知到下游 Task 失败之后，直接将发送到对应 Task 的数据丢弃。那么我们的改动逻辑，在这个示意图中，就是 Subpartition 通过 Netty Server 收到下游 Task Fail 的消息后，将自己设置为 Unavailable，然后 RecordWriter 在发送数据到指定 Subpartition 时，判断是否可用，如果不可用则直接将数据丢弃。而当 Task Failover 完成后重新与上游建立连接后，再将该 Subpartition 置为 Available，则数据可以重新被消费。

发送端的改动比较简单，得益于 Flink 内部对 Subpartition 的逻辑做了很好的抽象，并且可以很容易的通过参数来切换 Subpartition 初始化的类型，我们在这里参考 PipelinedSubpartition 的实现，根据上述的逻辑，实现了我们自己的 Subpartition 和对应的 View。

下游接收端的优化

同样，我们来细化一下下游接收端的细节：

Figure. 下游数据接收流程仔细来看，其实和上游的线程模型颇有类似之处：

(1) InputGate 初始化所有的 Channel 并通过 Netty Client 和上游 Server 建立连接。

(2) InputChannel 接收到数据后，缓存到 buffer 队列中并将自己的引用放入到 Channels Queue 里。（Netty 线程）

(3) InputGate 通过 InputProcessor 的调用，从 Queue 里拉取 Channel 并读取 Channel 中缓存的 buffer 数据，如果 buffer 不完整（比如只有半条 record），那么则会将不完整的 buffer 暂存到 InputProcessor 中。（Task 线程）

这里我们期望下游接收端感知到上游 Task 失败之后，能将对应 InputChannel 的接收到的不完整的 buffer 直接清除。不完整的 buffer 存储在 InputProcessor 中，那么我们如何让 InputProcessor 知道哪个 Channel 出现了问题？

简单的方案是说，我们在 InputChannel 中直接调用 InputGate 或者 InputProcessor，做 buffer 清空的操作，但是这样引入一个问题，由于 InputChannel 收到 Error 是在 Netty 线程，而 InputProcessor 的操作是在 Task 线程，这样跨线程的调用打破了已有的线程模型，必然会引入锁和调用时间的不确定性，增加架构设计的复杂度，并且因为 InputProcessor 会对每一条 record 都有调用，稍有不慎就会带来性能的下降。

我们沿用已有的线程模型，Client 感知到上游 Task 失败的消息之后告知对应的 Channel，Channel 向自己维护的 receivedBuffers 的末尾插入一个 UnavailableEvent，并等待 InputProcessor 拉取并清空对应 Channel 的 buffer 数据。示意图如下所示，红色的模块是我们新增的部分：

Figure. 下游改动示意图### JobManager 重启策略的优化

JobManager 重启策略可以参考社区已有的 RestartIndividualStrategy，比较重要的区别是，在重新 deploy 这个失败的 Task 后，我们需要通过 ExecutionGraph 中的拓扑信息，找到该 Task 的下游 Tasks，并通过 Rpc 调用让下游 Tasks 和这个新的上游 Tasks 重新建立连接。

这里实现有一个难点是如果 JobManager 去 update