性能文章>记录一次Flink作业异常的排查过程>

记录一次Flink作业异常的排查过程原创

2年前
8930625

最近2周开始接手apache flink全链路监控数据的作业,包括指标统计,业务规则匹配等逻辑,计算结果实时写入elasticsearch. 昨天遇到生产环境有作业无法正常重启的问题,我负责对这个问题进行排查跟进。

第一步,基础排查

首先拿到jobmanager和taskmanager的日志,我从taskmanager日志中很快发现2个基础类型的报错,一个是npe,一个是索引找不到的异常

elasticsearch sinker在执行写入数据的前后提供回调接口让作业开发人员对异常或者成功写入进行处理,如果在处理异常过程中有异常抛出,那么框架会让该task失败,导致作业重启。

npe很容易修复,索引找不到是创建索引的服务中的一个小bug,这些都是小问题。

重点是在日志中我看到另一个错误:

java.lang.OutOfMemoryError: unable to create new native thread
	at java.lang.Thread.start0(Native Method)
	at java.lang.Thread.start(Unknown Source)
	at org.apache.flink.runtime.io.network.api.writer.RecordWriter.<init>(RecordWriter.java:122)
	at org.apache.flink.runtime.io.network.api.writer.RecordWriter.createRecordWriter(RecordWriter.java:321)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.createRecordWriter(StreamTask.java:1202)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.createRecordWriters(StreamTask.java:1170)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.<init>(StreamTask.java:212)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.<init>(StreamTask.java:190)
	at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.<init>(OneInputStreamTask.java:52)
	at sun.reflect.GeneratedConstructorAccessor4.newInstance(Unknown Source)
	at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(Unknown Source)
	at java.lang.reflect.Constructor.newInstance(Unknown Source)
	at org.apache.flink.runtime.taskmanager.Task.loadAndInstantiateInvokable(Task.java:1405)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:689)
	at java.lang.Thread.run(Unknown Source)

这种异常,一般是nproc设置太小导致的,或者物理内存耗尽,检查完ulimit和内存,发现都很正常,这就比较奇怪了。

第二步、分析jstack和jmap

perfma有一个产品叫xland,我也是第一次使用,不得不说,确实牛逼,好用!
首先把出问题的taskmanager的线程栈信息和内存dump出来,具体命令:

jstatck pid > 生成的文件名
jmap -dump:format=b,file=生成的文件名 进程号

接着把这两个文件导入xland,xland可以直接看到线程总数,可以方便搜索统计线程数、实例个数等等

最先发现的问题是这个taskmanager 线程总数竟然有17000+,这个数字显然有点大,这个时候我想看一下,哪一种类型的线程比较大,xland可以很方便的搜索,统计,这时候我注意到有一种类型的线程非常多,总数15520

image.png
更上层的调用信息看不到了,只看到来自apache http client,根据作业流程,首先想到的就是es sinker的RestHighLevelClient用到这个东西

那么我们在xland中统计RestHighLevelClient对象个数,发现有几百个,很显然这里有问题

第三步、定位具体问题

有了前面xland的帮助,我们很容易定位到是esclient出了问题
在我们的作业里面有2个地方用到了es client,一个是es sinker,es sinker使用的就是RestHighLevelClient,另一个是我们同学自己写的一个es client,同样是使用RestHighLevelClient,在es sinker的ElasticsearchSinkFunction中单独构造,用于在写入es前,先搜索一些东西拿来合并,还做了cache

1、怀疑RestHighLevelClient bug

我们通过一个测试,来验证是不是RestHighLevelClient的问题

启动一个单纯使用es sinker的job,调整并发度,观察前面出现较多的
I/O dispatcher线程的个数,最后发现单个es sinker也会有240+个
I/O dispatcher线程,通过调整并发,所有taskmanager的
I/O dispatcher线程总数基本和并发成正向比例
停掉写es作业,此时所有taskmanager是不存在I/O dispatcher线程的

看起来I/O dispatcher那种线程数量大,似乎是“正常的”

2、杀掉作业,观察线程是否被正常回收
杀掉作业,I/O dispatcher线程变成0了,看起来es sinker使用是正常的

这时候基本上可以判断是我们自己写的es client的问题。到底是什么问题呢?

我们再做一个测试进一步确认

3、启动问题作业,杀死job后,观察I/O dispatcher线程个数
重启flink的所有taskmanager,给一个“纯净”的环境,发现杀死作业后,还有I/O dispatcher线程。
这个测试可以判断是我们的es client存在线程泄漏

四、背后的原理

es sinker本质上是一个RichSinkFunction,RichSinkFunction带了open 和close 方法,在close方法中,es sinker正确关闭了http client

@Override
	public void close() throws Exception {
		if (bulkProcessor != null) {
			bulkProcessor.close();
			bulkProcessor = null;
		}

		if (client != null) {
			client.close();
			client = null;
		}

		callBridge.cleanup();

		// make sure any errors from callbacks are rethrown
		checkErrorAndRethrow();
	}

而我们的es client是没有被正确关闭的。

具体原理应该是是这样的,当es sinker出现npe或者写es rejected等异常时,job会被flink重启,es sinker这种RichSinkFunction类型的算子会被flink 调用close关闭释放掉一些资源,而我们写在ElasticsearchSinkFunction中es client,是不会被框架关照到的,而这种写法我们自己也无法预先定义重启后关闭client的逻辑.

如果在构造时使用单例,理论上应该是可以避免作业反复重启时es client不断被构造导致线程泄漏和内存泄漏的,但是编写单例写法有问题,虽然有double check,但是没加volatile,同时锁的是this, 而不是类。

五、小结

1、xland确实好用,排查问题帮助很大
2、flink作业用到的外部客户端不要单独构造,要使用类似RichFunction这种方式,提供open,close方法,确保让资源能够被flink正确释放掉。
3、用到的对象,创建的线程,线程池等等最好都起一个名字,方便使用xland事后排查问题,如果有经验的话,应该一开始就统计下用于构造es client的那个包装类对象个数。

请先登录,查看6条精彩评论吧
快去登录吧,你将获得
  • 浏览更多精彩评论
  • 和开发者讨论交流,共同进步

为你推荐

类初始化导致死锁
一张图简单描述死锁 如上图,Thread1 拿到了 object1,Thread2 拿到了 object2,但是现在 Thread1 需要拿到 object2 的锁才能继续往下,Thread2 又要拿到 object1 才能继续往下
关于内存溢出,咱再聊点有意思的?
概述 上篇文章讲了JVM在GC上的一个设计缺陷,揪出一个导致GC慢慢变长的JVM设计缺陷,可能有不少人还是没怎么看明白的,今天准备讲的大家应该都很容易看明白 本文其实很犹豫写不写,因为感觉没有
字符串字面量长度是有限制的
前言 偶然在一次单元测试中写了一个非常长的字符串字面量。 正文 在一次单元测试中,我写了一个很长的字符串字面量,大概10万个字符左右,编译时,编译器给出了异常告警 `java: constant
多次字符串相加一定要用StringBuilder而不用-吗?
今天在写一个读取Java class File并进行分析的Demo时,偶然发现了下面这个场景(基于oracle jdk 1.8.0_144): ``` package test; public c
如何通过反射获得方法的真实参数名(以及扩展研究)
前段时间,在做一个小的工程时,遇到了需要通过反射获得方法真实参数名的场景,在这里我遇到了一些小小的问题,后来在部门老大的指导下,我解决了这个问题。通过解决这个问题,附带着我了解到了很多新的知识,我觉得
谨防JDK8重复类定义造成的内存泄漏
概述 如今JDK8成了主流,大家都紧锣密鼓地进行着升级,享受着JDK8带来的各种便利,然而有时候升级并没有那么顺利?比如说今天要说的这个问题。我们都知道JDK8在内存模型上最大的改变是,放弃了Perm
高吞吐、低延迟 Java 应用的 GC 优化实践
本篇原文作者是 LinkedIn 的 Swapnil Ghike,这篇文章讲述了 LinkedIn 的 Feed 产品的 GC 优化过程,虽然文章写作于 April 8, 2014,但其中的很多内容和
「每日五分钟,玩转 JVM」:久识你名,初居我心
聊聊 JVMJVM,一个熟悉又陌生的名词,从认识Java的第一天起,我们就会听到这个名字,在参加工作的前一两年,面试的时候还会经常被问到JDK,JRE,JVM这三者的区别。JVM可以说和我们是老朋友了