Flink AsyncFunction无法触发CheckpointedFunction方法及Checkpoint耗时过长问题

根因 #

Flink 中AsyncFunction不属于算子层实现，仅为用户自定义的业务逻辑类，你调用AsyncDataStream工具类提交异步逻辑时，框架会自动将AsyncFunction封装到内置的AsyncWaitOperator算子实例中。CheckpointedFunction的生命周期钩子只会作用于实际的算子实例，包裹层内部的AsyncFunction不会被注册到状态回调链中，因此你重写的snapshotState()、initializeState()方法永远不会被触发。

解决方案 #

• 状态逻辑上移：将需要持久化的状态放到调用AsyncDataStream的上游ProcessFunction中维护，将状态值随流数据一同传入AsyncFunction处理
• 自定义异步算子：若必须在异步算子层维护状态，可自行继承AsyncWaitOperator实现自定义算子类，在该类中实现CheckpointedFunction接口，再将自定义算子直接加入作业拓扑，不使用AsyncDataStream工具类生成算子

根因 #

你看到的日志Time from receiving all checkpoint barriers/RPC to executing it exceeded threshold已经明确了耗时点：算子已经收到了全部Checkpoint barrier，但无法立刻执行快照，需要等待前置条件满足。
恰好一次语义下，AsyncWaitOperator触发快照前必须等待所有在途的异步请求全部处理完成、结果输出完毕，才能执行快照操作，避免出现数据丢失或重复。如果你的异步请求RT高、超时时间设置不合理、并发队列堆积了大量未处理请求，哪怕状态仅为几个字节，也需要等待所有异步请求结束才能启动快照，这就是你Checkpoint耗时长达数分钟的核心原因，和本地存储介质无关。

解决方案 #

• 排查异步请求的平均RT、超时配置，设置合理的异步请求超时时间，及时清理超时的无效请求，避免无效等待
• 调整异步算子的async.operators.max.capacity参数与并行度，控制异步请求的并发上限，避免队列中堆积过多未处理请求
• 调大Checkpoint间隔，避免1秒的过短间隔频繁触发快照，进一步加剧请求堆积

内容的提问来源于stack exchange，提问作者diegoruizbarbero