[FLINK-10455] Potential Kafka producer leak in case of failures

• Type: Bug
• Status: Resolved
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 1.5.2
• Fix Version/s: 1.5.6, 1.7.0, 1.7.3, 1.8.1, 1.9.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available
• Assignee: Jiangjie Qin
• Reporter: Nico Kruber
• Created: 27/Sep/18 15:51
• Updated: 12/Jun/19 09:27
• Resolved: 12/Jun/19 09:27

Description

If the Kafka brokers' timeout is too low for our checkpoint interval [1], we may get an ProducerFencedException. Documentation around ProducerFencedException explicitly states that we should close the producer after encountering it.

By looking at the code, it doesn't seem like this is actually done in FlinkKafkaProducer011. Also, in case one transaction's commit in TwoPhaseCommitSinkFunction#notifyCheckpointComplete fails with an exception, we don't clean up (nor try to commit) any other transaction.
-> from what I see, TwoPhaseCommitSinkFunction#notifyCheckpointComplete simply iterates over the pendingCommitTransactions which is not touched during close()

Now if we restart the failing job on the same Flink cluster, any resources from the previous attempt will still linger around.

[1] https://ci.apache.org/projects/flink/flink-docs-release-1.5/dev/connectors/kafka.html#kafka-011

Issue Links

• is related to: FLINK-10876 Deadlock if closing firstly pending transactions in FlinkKafkaProducer(011).close()
• links to: GitHub Pull Request #6989
• links to: GitHub Pull Request #7107
• links to: GitHub Pull Request #7108
• links to: GitHub Pull Request #8653
• links to: GitHub Pull Request #8699
• links to: GitHub Pull Request #8701