Flink 流 - kafka.internals.Handover$ClosedException
Flink Streaming - kafka.internals.Handover$ClosedException
我正在使用 Apache Flink v1.12.3。
最近遇到这个错误，不知道具体是什么意思。错误与Kafka或Flink有关吗？
错误日志：
2021-07-02 21:32:50,149 WARN org.apache.kafka.clients.consumer.internals.ConsumerCoordinator [] - [Consumer clientId=consumer-myGroup-24, groupId=myGroup] Offset commit failed on partition my_topic-14 at offset 11328862986: The request timed out.
// ...
2021-07-02 21:32:50,150 INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator [] - [Consumer clientId=consumer-myGroup-24, groupId=myGroup] Group coordinator 1.2.3.4:9092 (id: 2147483616 rack: null) is unavailable or invalid, will attempt rediscovery
// ...
2021-07-02 21:33:20,553 INFO org.apache.kafka.clients.FetchSessionHandler [] - [Consumer clientId=consumer-myGroup-21, groupId=myGroup] Error sending fetch request (sessionId=1923351260, epoch=9902) to node 29: {}.
// ...
2021-07-02 21:33:19,457 INFO org.apache.kafka.clients.FetchSessionHandler [] - [Consumer clientId=consumer-myGroup-15, groupId=myGroup] Error sending fetch request (sessionId=1427185435, epoch=10820) to node 29: {}.
org.apache.kafka.common.errors.DisconnectException: null
// ...
2021-07-02 21:34:10,157 WARN org.apache.flink.runtime.taskmanager.Task [] - Source: my_topic_stream (4/15)#0 (2e2051d41edd606a093625783d844ba1) switched from RUNNING to FAILED.
org.apache.flink.streaming.connectors.kafka.internals.Handover$ClosedException: null
at org.apache.flink.streaming.connectors.kafka.internals.Handover.close(Handover.java:177) ~[blob_p-a7919582483974414f9c0d4744bab53199b880d7-d9edc9d0741b403b3931269bf42a4f6b:?]
at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.cancel(KafkaFetcher.java:164) ~[blob_p-a7919582483974414f9c0d4744bab53199b880d7-d9edc9d0741b403b3931269bf42a4f6b:?]
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.cancel(FlinkKafkaConsumerBase.java:945) ~[blob_p-a7919582483974414f9c0d4744bab53199b880d7-d9edc9d0741b403b3931269bf42a4f6b:?]
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.lambda$createAndStartDiscoveryLoop(FlinkKafkaConsumerBase.java:913) ~[blob_p-a7919582483974414f9c0d4744bab53199b880d7-d9edc9d0741b403b3931269bf42a4f6b:?]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_212]
这是卡夫卡的问题。 Kafka 消费者客户端在将偏移量提交给 Kafka 集群时抛出错误（超时）。一种可能的原因是Kafka集群繁忙，无法及时响应。此错误使任务管理器 运行 Kafka 消费者失败。
尝试在从 Kafka 创建源流时向属性添加参数。可能的参数是：request.timeout.ms，把时间调长一点试试。
参考文献：
我正在使用 Apache Flink v1.12.3。
最近遇到这个错误，不知道具体是什么意思。错误与Kafka或Flink有关吗？
错误日志：
2021-07-02 21:32:50,149 WARN org.apache.kafka.clients.consumer.internals.ConsumerCoordinator [] - [Consumer clientId=consumer-myGroup-24, groupId=myGroup] Offset commit failed on partition my_topic-14 at offset 11328862986: The request timed out.
// ...
2021-07-02 21:32:50,150 INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator [] - [Consumer clientId=consumer-myGroup-24, groupId=myGroup] Group coordinator 1.2.3.4:9092 (id: 2147483616 rack: null) is unavailable or invalid, will attempt rediscovery
// ...
2021-07-02 21:33:20,553 INFO org.apache.kafka.clients.FetchSessionHandler [] - [Consumer clientId=consumer-myGroup-21, groupId=myGroup] Error sending fetch request (sessionId=1923351260, epoch=9902) to node 29: {}.
// ...
2021-07-02 21:33:19,457 INFO org.apache.kafka.clients.FetchSessionHandler [] - [Consumer clientId=consumer-myGroup-15, groupId=myGroup] Error sending fetch request (sessionId=1427185435, epoch=10820) to node 29: {}.
org.apache.kafka.common.errors.DisconnectException: null
// ...
2021-07-02 21:34:10,157 WARN org.apache.flink.runtime.taskmanager.Task [] - Source: my_topic_stream (4/15)#0 (2e2051d41edd606a093625783d844ba1) switched from RUNNING to FAILED.
org.apache.flink.streaming.connectors.kafka.internals.Handover$ClosedException: null
at org.apache.flink.streaming.connectors.kafka.internals.Handover.close(Handover.java:177) ~[blob_p-a7919582483974414f9c0d4744bab53199b880d7-d9edc9d0741b403b3931269bf42a4f6b:?]
at org.apache.flink.streaming.connectors.kafka.internals.KafkaFetcher.cancel(KafkaFetcher.java:164) ~[blob_p-a7919582483974414f9c0d4744bab53199b880d7-d9edc9d0741b403b3931269bf42a4f6b:?]
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.cancel(FlinkKafkaConsumerBase.java:945) ~[blob_p-a7919582483974414f9c0d4744bab53199b880d7-d9edc9d0741b403b3931269bf42a4f6b:?]
at org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase.lambda$createAndStartDiscoveryLoop(FlinkKafkaConsumerBase.java:913) ~[blob_p-a7919582483974414f9c0d4744bab53199b880d7-d9edc9d0741b403b3931269bf42a4f6b:?]
at java.lang.Thread.run(Thread.java:748) [?:1.8.0_212]
这是卡夫卡的问题。 Kafka 消费者客户端在将偏移量提交给 Kafka 集群时抛出错误（超时）。一种可能的原因是Kafka集群繁忙，无法及时响应。此错误使任务管理器 运行 Kafka 消费者失败。
尝试在从 Kafka 创建源流时向属性添加参数。可能的参数是：request.timeout.ms，把时间调长一点试试。
参考文献：