[FLINK-18276] NullPointerException (and potential leak) when closing KafkaConsumer after exception in reassignPartitions

• Type: Bug
• Status: Closed
• Priority: Minor
• Resolution: Abandoned
• Affects Version/s: 1.8.3, 1.9.3, 1.10.1, 1.11.0
• Component/s: Connectors / Kafka
• Labels: auto-deprioritized-major
• Assignee: Unassigned
• Reporter: Nico Kruber
• Created: 12/Jun/20 08:47
• Updated: 29/Sep/21 02:12
• Resolved: 29/Sep/21 02:12

Description

WARN  org.apache.flink.streaming.connectors.kafka.internal.KafkaFetcher  - Error while closing Kafka consumer
java.lang.NullPointerException
	at org.apache.flink.streaming.connectors.kafka.internal.KafkaConsumerThread.run(KafkaConsumerThread.java:282)

KafkaConsumerThread#reassignPartitions is temporarily setting consumer to null and if there is an exception (in this case, it was a timeout), the finally block in KafkaConsumerThread.run would fail with an NPE. Even more so, KafkaConsumerThread#reassignPartitions put the original consumer into consumerTmp which is not closed now and may leak underlying (Kafka) resources.