最新在线上更新了代码之后导致了任务在消费kafka数据的时候,突然就不消费数据了,发现原因在公司的可视化界面中,看不到数据的更新,进入flink监控页面中看到任务没有failover过的记录
虽然任务在正常的运行中,但实际情况是已经不消费数据了,最开始以为代码有问题,经过检查发现代码没有问题,然后查看checkpoint的情况,发现了问题,flink在做checkpoint的时候出现了失败
在查看tm和jm的日志,tm日志中没有任何报错,但是在jm日志中看到一个有用的信息,说在做checkpoint的时候checkpoint超时,导致了checkpoint失败,后来我通过
因为新任务增加了state的使用,之前checkpoint的超时时间设置为1s,我以为是因为状态大了导致checkpoint超时,然后我更改了超时时间,设置成了1分钟,但运行一段时间后也发现同样的错误,在错误中发现了一个问题,在KeyedProcess算子中,有一个subtask acknowledgement Time为N/A,导致了checkpoint的超时,因为他的ack Time没有,使barrier一直处于对齐的过程中,导致checkpoint超时而失败
在解决这个问题之前,我看一下kafka的分区的数据出现了不平均的情况,以为是因为kafka分区的数据原因,导致了flink source消费不到数据,无法发送barrier,但是后来仔细想了一下,checkpoint的barrier是自动生成的然后发送到下游,也就是跟消费数据是没有关系的,那么也就是我flink的任务链整个都堵塞了,使barrier无法正常发送到下游,使checkpoint失败,那么也就是因为整个任务链的堵塞导致了无法在消费新的数据,
根据上面的线索我开始排查代码,和测试代码一共找到两个位置出现了问题
1.在异步IO的时候,因为我添加了过滤条件,导致asyncInvoke方法不能正常的返回结果,导致一直处于堵塞的状态,最终使超过了capacity(容量),从而导致整个链路堵塞,即不消费kafka数据
红色圈圈之前没有注释导致了数据来了无法正常的发送下游,最终导致整个链路堵塞
灰色圈圈是需要注意的地方 这是我后加上去的,因为在这里有一个异常,如果出现了异常也无法正常调用到complete方法,也会导致链路的堵塞,所以当出现异常通过completeExceptionally方法来异常结束掉
在这里提一下,异步IO中 resultFuture.completeExceptionally方法和resultFuture.complete,也是通过这次问题学习到的地方
这两个方法任何一个被调用的时候都会生成一个 triggered 标志,代表这个CompletableFuture状态变为完成,在源码中可以看到,这里顺便看一下源码
调用过程
1.asyncInvoke方法中调用 resultFuture.completeExceptionally - > StreamRecordQueueEntry.completeExceptionally -> CompletableFuture.completeExceptionally , 完成最终的触发
public boolean completeExceptionally(Throwable ex) {
if (ex == null) throw new NullPointerException();
boolean triggered = internalComplete(new AltResult(ex));
postComplete();
return triggered;
}
2.asyncInvoke方法中调用 resultFuture.complete -> StreamRecordQueueEntry.complete -> CompletableFuture.complete , 完成最终的触发
public boolean complete(T value) {
boolean triggered = completeValue(value);
postComplete();
return triggered;
}
flink的异步IO 最终需要调用complete 或者completeExceptionally来完成一次触发,否则会因为超过capacity而阻塞整个链路
2.在我的keyedProcess的onTimer方法中,错误的删除了所有定时器,导致无法在触发新的定时器,是数据无法正常发送到下游,也间接的导致了checkpoint的失败
改完了代码之后重新执行任务已经可以正常的运行了,并且checkpoint也没有失败过
