[FLINK-29153] KafkaConsumerThread should catch WakeupException when committing offsets

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 1.16.0
• Fix Version/s: 1.16.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available
• Assignee: Qingsheng Ren
• Reporter: Qingsheng Ren
• Created: 31/Aug/22 07:01
• Updated: 08/Nov/23 06:40
• Resolved: 13/Sep/22 02:20

Description

KafkaConsumerThread in the legacy FlinkKafkaConsumer makes a wakeup on the KafkaConsumer on offset commit to wakeup the potential blocking KafkaConsumer.poll(). However the wakeup might happen when the consumer is not polling. The wakeup will be remembered by the consumer and re-examined while committing the offset asynchronously, which leads to an unnecessary WakeupException.

As the JavaDoc and method signature of KafkaConsumer does not show that the WakeupException could be thrown in KafkaConsumer#commitAsync, this could be considered as a bug on Kafka side (breaking the contract of public API), but we can still fix it by catching the WakeupException and retrying.

Issue Links

• causes: FLINK-24119 KafkaITCase.testTimestamps fails due to "Topic xxx already exist"
• causes: FLINK-29018 KafkaITCase.testMultipleSourcesOnePartition failed with TopicExistsException
• links to: GitHub Pull Request #20764