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

本文主要讲解关于如何解决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），学习愉快哦！