rerorero created FLINK-25293: -------------------------------- Summary: Option to let fail if KafkaSource keeps failing to commit offset Key: FLINK-25293 URL: https://issues.apache.org/jira/browse/FLINK-25293 Project: Flink Issue Type: Improvement Components: Connectors / Kafka Affects Versions: 1.14.0 Environment: Flink 1.14.0 Reporter: rerorero
Is it possible to let KafkaSource fail if it keeps failing to commit offset? I faced an issue where KafkaSource keeps failing and never recover, while it's logging like these logs: {code:java} 2021-12-08 22:18:34,155 INFO org.apache.kafka.clients.consumer.internals.AbstractCoordinator [] - [Consumer clientId=dbz-mercari-contact-tool-jp-cg-1, groupId=dbz-mercari-contact-tool-jp-cg] Group coordinator b4-pkc-xmj7g.asia-northeast1.gcp.confluent.cloud:9092 (id: 2147483643 rack: null) is unavailable or invalid due to cause: null.isDisconnected: true. Rediscovery will be attempted. 2021-12-08 22:18:34,157 WARN org.apache.flink.connector.kafka.source.reader.KafkaSourceReader [] - Failed to commit consumer offsets for checkpoint 13 {code} This is happening not just once, but a couple of times a week. I found [other people reporting the same thing](https://lists.apache.org/thread/8l4f2yb4qwysdn1cj1wjk99tfb79kgs2). This could possibly be a problem with the Kafka client. It can be resolved by restarting the Flink Job. However, Flink Kafka connector doesn't provide an automatic way to save this situation. KafkaSource [keeps retrying forever when a retriable error occurs](https://github.com/apache/flink/blob/afb29d92c4e76ec6a453459c3d8a08304efec549/flink-connectors/flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/reader/KafkaSourceReader.java#L144-L148], even if it is not retriable actually. Since it sends metrics of the number of times a commit fails, it could be automated by monitoring it and restarting the job, but that would mean we need to have a new process to manage. Does it make sense to have KafkaSource have the option like, let the source task fail if it keeps failing to commit an offset more than X times? -- This message was sent by Atlassian Jira (v8.20.1#820001)