Google Dataflow遇RuntimeException时是否会自动重试DoFn？

针对这个Dataflow无界管道里外部服务数据增强偶发报错的场景，我这边有个适配性很强的方案——完全弃用Failsafe，改用Dataflow原生的容错机制来解决，具体逻辑和实现思路如下：

问题场景回顾 #

我们的管道基于无界数据源（Unbounded data source），在调用外部服务做数据增强（enrichment）时，偶尔会抛出RuntimeException。根本原因是Dataflow的处理速度快于外部服务的数据同步速度：当Dataflow处理某条数据时，外部服务还没完成这条数据的识别，但等待10秒后再尝试，外部服务就能正常响应，不会再报错。

为什么弃用Failsafe？ #

Failsafe是通用的重试框架，但它并没有和Dataflow的分布式运行模型、检查点机制、窗口处理等特性深度集成。如果强行在Dataflow管道里使用Failsafe，可能会出现重试逻辑和Dataflow自身容错机制冲突的情况，比如重复处理元素、破坏Exactly-Once语义等。而Dataflow原生机制是为自身的数据流处理场景量身打造的，适配性更好。

利用Dataflow原生机制解决问题的具体做法 #

• 配置内置重试策略：对于无界数据流的管道，Dataflow本身支持对失败的元素进行重试。我们可以通过StreamingPipelineOptions来配置重试参数：
  • 设置重试延迟：因为明确知道10秒后问题会解决，所以直接把初始重试延迟设为10秒，代码示例如下：

    StreamingPipelineOptions options = PipelineOptionsFactory.as(StreamingPipelineOptions.class);
    options.setRetryDelay(Duration.standardSeconds(10));
    

  • 设置重试次数：由于10秒后必然能成功，所以只需要设置1次重试即可，避免不必要的重复处理：

    options.setMaxRetryAttempts(1);
    

• 结合检查点保证一致性：Dataflow的检查点机制会记录元素的处理状态，重试时会从检查点恢复未成功处理的元素，确保数据处理的Exactly-Once语义，不会出现元素丢失或重复处理的问题。
• 异常处理的适配：在@ProcessElement方法中抛出的RuntimeException会被Dataflow的原生机制捕获，自动触发重试逻辑，不需要额外的异常封装。

这样配置后，当外部服务暂时无法识别数据时，Dataflow会自动等待10秒后重试该元素，既解决了报错问题，又完全贴合Dataflow的运行特性，比使用Failsafe更可靠。

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