生产事故：flink任务造成kafka连接泄漏导致kafka集群挂了！！！

1、问题背景

为保障系统大促期间稳定运行，计划进行全链路生产压测。

2、问题现象

1》压测期间产生大量事后数据流向flink实时计算环节，flink任务消费的kafka出现堆积而产生告警。

2》通过flink监控平台查看日志发现flink任务频繁重启失败，checkpoint save失败。

3》通过kafka平台监控发现，flink任务连接kafka的连接数不断攀升，即kafka连接泄漏。

4》短时间内所有flink任务都出现频繁的重启失败，导致kafka集群夯住，最终kafka集群挂了。

3、临时解决方案

停止所有的flink任务，重启kafka集群，然后分批重启flink任务，不断观察kafka连接数情况。针对个别flink任务重启后连接数依然高的情况，通过暂时剔除某业务计算逻辑后重启，通过跳过测试数据，把阻塞的数据消费掉来临时解决，等堆积消失后再添加之前剔除掉的业务计算逻辑后重启。

4、问题原因分析

flink job在发送数据的时候会检查是否有外部异常，若此时发现有外部异常(压测把磁盘IO打满，导致kafka partition leader切换), 此时会触发任务重启，但不会释放container资源。在关闭各TaskManager的FlinkKafkaProducer调用close时出现了NoSuchMethodError，导致KafkaProducer连接无法正常关闭，进而引发Kafka Producer连接泄露。

根本原因：使用jar包版本不匹配，即flink版本为1.11.1，依赖的kafka客户端版本为2.4.1，而实际项目中使用的kafka客户端版本为1.0.0，造成版本冲突，FlinkKafkaProducer在调用close时产生：java.lang.NoSuchMethodError，导致flink producer与kafka服务端连接没有释放，导致连接泄露。

5、结论与修复方案

结论：千里之堤毁于蚁穴。
修复：更换依赖版本一致的jar包。