Flink Kafka Source Operator无法自动重连问题排查求助

问题场景 #

• Flink版本：1.16.1，Kafka Connector版本与Flink匹配
• 作业包含2个Kafka Source Operator，分别消费不同Topic，已启用每分钟一次的Checkpoint
• 测试操作：将Kafka Pod缩容至0，10分钟后恢复原规模
• 预期行为：Kafka恢复后Source自动重连并继续消费
• 实际行为：Source既不重连也不触发作业失败，作业状态无变化陷入循环，仅删除JM/TM Pod后才恢复
• Kafka恢复后关键错误日志：

  org.apache.kafka.common.errors.TimeoutException: Failed to send request after 30000 ms.
  2023-08-29 07:32:01.945 [Source Data Fetcher for Source: Kafka Data source (3/3)#0] INFO  org.apache.kafka.clients.NetworkClient  - [Consumer clientId=data-group-2, groupId=data-group] Disconnecting from node 1 due to socket connection setup timeout. The timeout value is 30096 ms.
  2023-08-29 07:32:01.945 [Source Data Fetcher for Source: Kafka Data source (3/3)#0] INFO  org.apache.kafka.clients.FetchSessionHandler  - [Consumer clientId=data-group-2, groupId=data-group] Error sending fetch request (sessionId=1361712539, epoch=INITIAL) to node 1:
  org.apache.kafka.common.errors.DisconnectException: null
  

问题原因 #

默认配置无法应对长时间Kafka集群不可用的场景：

• Kafka Consumer默认的重连退避上限（reconnect.backoff.max.ms）为10秒，当Kafka不可用超过该时间，Consumer会进入低频次无意义重连循环，且Flink Source未检测到该状态触发重启
• Flink默认重启策略为固定延迟重试，仅重试3次后停止，若Kafka不可用期间重试次数耗尽，后续不会再触发重试
• Kafka集群恢复后，Consumer未主动刷新元数据，仍指向失效节点，无法建立新连接

解决方案 #

1. 调整Kafka Consumer重连与元数据配置 #

在Kafka Source的配置中添加以下参数，确保Consumer能主动刷新元数据并合理重试：

properties.setProperty("reconnect.backoff.max.ms", "60000"); // 最大重连间隔设为1分钟
properties.setProperty("reconnect.backoff.ms", "1000"); // 初始重连间隔1秒
properties.setProperty("metadata.max.age.ms", "30000"); // 每30秒强制刷新元数据
properties.setProperty("fetch.max.wait.ms", "5000"); // 缩短拉取超时，更快感知连接异常

2. 配置Flink作业无限重试策略 #

修改flink-conf.yaml或在代码中设置，确保作业在Kafka恢复后能自动重启：

配置文件方式

restart-strategy: fixed-delay
restart-strategy.fixed-delay.attempts: -1 # -1表示无限重试
restart-strategy.fixed-delay.delay: 10s # 每次重试间隔10秒

代码方式

env.setRestartStrategy(RestartStrategies.fixedDelayRestart(
        RestartStrategies.RESTART_ATTEMPTS_UNLIMITED,
        org.apache.flink.api.common.time.Time.seconds(10)
));

3. 优化Kafka连接检测配置 #

添加心跳与会话超时配置，加快连接失效检测：

properties.setProperty("heartbeat.interval.ms", "3000");
properties.setProperty("session.timeout.ms", "30000");

验证步骤 #

1. 应用上述配置后重新提交作业
2. 重复测试场景：缩容Kafka Pod至0，10分钟后恢复
3. 检查Flink作业日志，确认Source自动重连并恢复消费

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