[FLINK-17327] Kafka unavailability could cause Flink TM shutdown

• Type: Bug
• Status: Closed
• Priority: Critical
• Resolution: Fixed
• Affects Version/s: 1.10.0
• Fix Version/s: 1.11.0
• Component/s: Connectors / Kafka
• Labels: pull-request-available
• Assignee: Aljoscha Krettek
• Reporter: Jun Qin
• Created: 22/Apr/20 14:51
• Updated: 26/Feb/21 13:42
• Resolved: 15/Jun/20 17:14

Description

Steps to reproduce:

1. Start a Flink 1.10 standalone cluster
2. Run a Flink job which reads from one Kafka topic and writes to another topic, with exactly-once checkpointing enabled
3. Stop all Kafka Brokers after a few successful checkpoints

When Kafka brokers are down:

1. org.apache.kafka.clients.NetworkClient reported connection to broker could not be established
2. Then, Flink could not complete snapshot due to Timeout expired while initializing transactional state in 60000ms
3. After several snapshot failures, Flink reported Too many ongoing snapshots. Increase kafka producers pool size or decrease number of concurrent checkpoints.
4. 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
5. 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:

1. org.apache.kafka.clients.NetworkClient reported connection to broker could not be established
2. 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
3. Flink tried to cancel the upstream tasks and created a new producer
4. The new producer obviously reported connectivity issue to brokers
5. This continues till Kafka brokers are back. 
6. Flink reported Too many ongoing snapshots. Increase kafka producers pool size or decrease number of concurrent checkpoints.
7. Flink cancelled the tasks and restarted them
8. The job continues, and new checkpoint succeeded. 
9. 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. 

 

 

 

Issue Links

• causes: FLINK-18311 StreamingKafkaITCase stalls indefinitely
• is caused by: FLINK-17350 StreamTask should always fail immediately on failures in synchronous part of a checkpoint
• is caused by: FLINK-17351 CheckpointCoordinator and CheckpointFailureManager ignores checkpoint timeouts
• relates to: FLINK-15362 Bump Kafka client version to 2.4.1 for universal Kafka connector
• links to: GitHub Pull Request #12589
• links to: GitHub Pull Request #12747