一.引言
使用 Flink - Kafka 接数据 Source 时程序报错：
org.apache.flink.runtime.JobException: Recovery is suppressed by FixedDelayRestartBackoffTimeStrategy任务每次启动后持续10min左右，然后 RUNNING -> FAILED，如此重启失败了多次。
二.问题现象
1.任务 URL 界面
对应任务界面可以看到有一个 Source 的 3 个 Task 在任务启动的时间内一直处于 INITIALIZING 状态，直到任务结束。
2.yarn 界面
上述报错情况下 yarn 界面如下，任务重启后大约 7min 失败
三.问题分析与解决
1.Source 持续?INITIALIZING 与?周期性 Failed
查看异常栈日志：
A.持续?INITIALIZING?
从下往上看，显示任务被远程的 taskManager close，结合上面 Source 端有 3 个 task 一直处于?INITIALIZING 状态，大概率是因为某?broker?crash ，从而导致无法 ping 通该节点，导致超时无法启动，当超时时间大于 flink 规定的心跳周期，任务?INITIALIZING -> FAILED，所以出现URL 的持续?INITIALIZING。
B.周期性 Failed
再往上是：Flink Recovery is suppressed by FixedDelayRestartBackoffTime，其中 maxNumberRestartAttempt 为 3，此时我们可以到提交客户端的?flink-conf.yaml 查看对应配置：
restart-strategy: failure-rate
restart-strategy.failure-rate.failure-rate-interval: 2 min可以看到重启策略为 failure-rate 即周期性的重启，其中周期为 2min，结合上面?maxNumberRestartAttempt = 3，这也找到了为什么程序 7min 左右循环退出的原因了，程序有1 min 左右的申请资源和初始化启动的时间，运行期间出现故障，按照?failure-rate 策略重启共计耗时 3 x 2min = 6min，3次尝试后任务仍然无法运行( taskManger 持续 ping 不通导致故障) 达到最大重启次数，任务退出。
filure-rate 重启策略如下：
重启策略在出现故障后重新启动作业，但当超过故障率（每个时间间隔的故障数）时，作业最终会失败。在两次连续的重启尝试之间，重启策略会等待固定的时间。
也可以在代码中配置：
val env = StreamExecutionEnvironment.getExecutionEnvironment()
env.setRestartStrategy(RestartStrategies.failureRateRestart(
3, // max failures per unit
Time.of(5, TimeUnit.MINUTES), //time interval for measuring failure rate
Time.of(10, TimeUnit.SECONDS) // delay
))查看更多的重启策略信息可以查看官方 API：Flink 重启策略。
2.问题解决
上面分析了很多，又是看日志，又是看官方 API，问题解决的方法其实很简单:
A.获取持续处于?INITIALIZING 的 task 对应的 Broker 地址
找到对应 source 对应 kafka 的 properties 配置中的 bootstrap.servers 参数即可，一般形式为 "broker1:port,broker2:port,broker3:port,..."
B.Ping 各个 broker
ping $broker分别 ping 对应 broker 对应机器，如果该 broker 正常会快速显示下述类似信息：
如果该 broker 异常，则 ping 后 shell 界面处于 _ 的等待状态，此时说明对应 broker 连接异常。
C.将异常 broker 从??bootstrap.servers 参数中剔除重启
一般情况下 kafka 都会有机器设置的冗余且实现互备，所以正常情况下去掉单台 broker 重启后任务不受影响
3.其他系统问题
Recovery is suppressed by FixedDelayRestartBackoffTimeStrategy 该报错主要是告知对应任务故障重启，主要异常分析还是要依靠最开始的异常栈确定异常源头，且大部分是与 flink 相关的系统参数，配置有关，例如本例就是 flink 的 source 端无法完全启动导致。除了上述 broker crash 掉点导致的 connection refused 之外
A.时间语义不匹配
还可能因为任务的时间语义与env设置不匹配导致任务重启并最终故障：
// 事件时间
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
// 处理时间
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
B.TimeWindowAll 并行度
TimeWindowAll 的并行度默认为1，不可以修改，如果代码内 setParallelism > 1 则会在任务 submit 时直接报错
C.non serializable field 序列化
如果在 Source 函数或者其他需要序列化的类内初始化了不可序列化的变量会在任务启动时报错，需要使该变量支持序列化或采用其他方式解决