[ https://issues.apache.org/jira/browse/FLINK-17327?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17130745#comment-17130745 ]
Aljoscha Krettek commented on FLINK-17327: ------------------------------------------ For posterity, here's a summary of the issue: * our code calls {{close()}} on the {{KafkaProducer}} (Kafka code), which is equivalent to calling {{close}} with a timeout of {{Long.MAX_VALUE}} * this means threads will leak when a failure happens, for example because of Broker downtime * the Flink Task Watchdog will kill the Task Manager because of these threads after a timeout The fix is to always call {{close()}} with a reasonable timeout. The fix also requires a Kafka version bump because of KAFKA-6635/KAFKA-7763, which mean that resources still leak even when closing with a timeout. Additionally, we need to close with exactly zero as timeout, because otherwise in-flight transactions will be aborted. > Kafka unavailability could cause Flink TM shutdown > -------------------------------------------------- > > Key: FLINK-17327 > URL: https://issues.apache.org/jira/browse/FLINK-17327 > Project: Flink > Issue Type: Bug > Components: Connectors / Kafka > Affects Versions: 1.10.0 > Reporter: Jun Qin > Priority: Critical > Fix For: 1.12.0 > > Attachments: 0001-Change-version-to-2.4.2-ALJOSCHA.patch, > 0002-Don-t-abort-in-flight-transactions.patch, Standalonesession.log, TM.log, > TM_produer_only_task.log > > > Steps to reproduce: > # Start a Flink 1.10 standalone cluster > # Run a Flink job which reads from one Kafka topic and writes to another > topic, with exactly-once checkpointing enabled > # Stop all Kafka Brokers after a few successful checkpoints > When Kafka brokers are down: > # {{org.apache.kafka.clients.NetworkClient}} reported connection to broker > could not be established > # Then, Flink could not complete snapshot due to {{Timeout expired while > initializing transactional state in 60000ms}} > # After several snapshot failures, Flink reported {{Too many ongoing > snapshots. Increase kafka producers pool size or decrease number of > concurrent checkpoints.}} > # Eventually, Flink tried to cancel the task which did not succeed within 3 > min. According to logs, consumer was cancelled, but producer is still running > # Then {{Fatal error occurred while executing the TaskManager. Shutting it > down...}} > I will attach the logs to show the details. Worth to note that if there > would be no consumer but producer only in the task, the behavior is different: > # {{org.apache.kafka.clients.NetworkClient}} reported connection to broker > could not be established > # after {{delivery.timeout.ms}} (2min by default), producer reports: > {{FlinkKafkaException: Failed to send data to Kafka: Expiring 4 record(s) for > output-topic-0:120001 ms has passed since batch creation}} > # Flink tried to cancel the upstream tasks and created a new producer > # The new producer obviously reported connectivity issue to brokers > # This continues till Kafka brokers are back. > # Flink reported {{Too many ongoing snapshots. Increase kafka producers pool > size or decrease number of concurrent checkpoints.}} > # Flink cancelled the tasks and restarted them > # The job continues, and new checkpoint succeeded. > # TM runs all the time in this scenario > I set Kafka transaction time out to 1 hour just to avoid transaction timeout > during the test. > To get a producer only task, I called {{env.disableOperatorChaining();}} in > the second scenario. > > > -- This message was sent by Atlassian Jira (v8.3.4#803005)