解决Flink Kafka Connector 提交位点失败问题

后端 潘老师 3个月前 (12-10) 60 ℃ (0) 扫码查看

本文主要讲解关于如何解决Flink Kafka Connector 提交位点失败问题 相关内容,让我们来一起学习下吧!

现象:Flink 版本为 1.15.1,消费 Kafka 时提交位点失败,外部监控 Grafana 报警,但是由于 Flink 内部位点更新成功,而且 Checkpoint 成功,因此不影响正常的数据处理,只会造成从外部监控系统看 group 消息堆积。也是因为位点提交到 Flink 外部broker 是在 notifyCheckpointCompleted 方法中,该方法是在 Checkpoint 成功之后才会调用,即使失败也不会影响到 Checkpoint。

原因:第一次因为超时提交失败后,便把 Consumer Coordinator 置为 null,后面的每一次提交都会由于COORDINATOR_NOT_AVAILABLE 失败,而不会恢复。因此是 Kafka Client 的问题,而不是Flink的问题。

解决方法

  1. 调大 request.time.out,尽量避免 client 到 broker 之间的超时。默认30s,但是生产环境调到60s依然会出现请求超时。假如日志中存在DisconnectionException其实也是请求超时(Note that there is no* need to check for disconnects explicitly on the {@link ClientResponse} object;* instead, the future will be failed with a {@link DisconnectException}.
  2. 重启 Job,take savepoint 然后从 savepoint 恢复,恢复之后会用新的 Consumer Coordinator 将正确的位点(savepoint中的位点)提交到 broker,外部监控堆积可以直接降为正常。
  3. 升级 Flink / Kafka Connector,避免使用有问题的 Kafka Client(2.6.1, 3.1.0, 2.7.2, 2.8.1, 3.0.0)。注意如果 Flink 升级到 1.16.0 且使用阿里云全托管 Kafka 云盘存储,需要修改 Kafka 配置 enable.idempotence=false,否则报错:org.apache.flink.kafka.shaded.org.apache.kafka.clients.producer.internals.Sender [] - [Producer clientId=producer-23] Got error produce response with correlation id 2675 on topic-partition tapdb_event_confluence-1, retrying (2147482824 attempts left). Error: CORRUPT_MESSAGE

以上就是关于如何解决Flink Kafka Connector 提交位点失败问题相关的全部内容,希望对你有帮助。欢迎持续关注潘子夜个人博客(www.panziye.com),学习愉快哦!


版权声明:本站文章,如无说明,均为本站原创,转载请注明文章来源。如有侵权,请联系博主删除。
本文链接:https://www.panziye.com/back/12416.html
喜欢 (0)
请潘老师喝杯Coffee吧!】
分享 (0)
用户头像
发表我的评论
取消评论
表情 贴图 签到 代码

Hi,您需要填写昵称和邮箱!

  • 昵称【必填】
  • 邮箱【必填】
  • 网址【可选】