背景

突然间钉钉群出现告警，发现Flink集群中跑的所有Job进入Restarting。业务中断，影响比较大，需要迅速定位恢复。

问题处理

1、查看flink集群、yarn集群、hdfs集群的运行状态，发现运行正常。

2、查看taskmanager日志文件发下异常:

2021-12-29 09:07:11,465 WARN  org.apache.flink.runtime.taskmanager.Task                    [] - Source: Custom Source -> Filter (1/1)#35 (ba2fe0076320a3ee8f7e7b2f471feab5) switched from RUNNING to FAILED with failure cause: java.lang.NullPointerException

此异常是在task运行的时候抛出的，可以从异常日志看到问题出现在Source的Filter方法中，也就是我们自己写的filter代码，查看Filter的具体代码发现存在空指针异常问题。对存在空值的地方进行非空判断，重启后Job恢复。

原理分析

Job恢复后，我们进行了深入分析：由于最近我们也没有新协议的接入，而且数据格式都是规整过的数据。排除是消息格式引起的问题。深入看下代码，发现报错的地方，算子filter方法中变量参数dataAvailableTime是直接使用的主方法的变量。代码如下：

main...

String dataAvailableTime=...;

DataStream<Tuple2<String, String>> kafkaDataStream = env.addSource(myConsumer).filter(new FilterFunction<Tuple2<String, String>>() {

                @Override

                public boolean filter(Tuple2<String, String> stringStringTuple2) throws Exception {

                    //校验数据request_time是否还在有效期

                    boolean res = DtoTransferUtil.validateBasic(stringStringTuple2.f1,dataAvailableTime);

                    return res;

                }

    });

    内部类中dataAvailableTime变量直接使用的主类的变量。在flink中有job manager和task manager之分。其中，job manager负责job的调度，task manager负责任务的执行。我们通过Flink Api编写流式代码的时候，flink会将你编写的这些流式代码构建成StreamGraph。StreamGraph是基于有向无环图来实现的，有向无环图能很好的解决任务调度间的依赖。然后再将StreamGraph进行执行优化，生成JobGraph。最后将生成的JobGraph提交给JobManager，由JobManager负责任务的分发和调度。   

任务执行的有向无环图

任务执行流程

JobManager和TaskManager在flink中是单独的进程，所有分布式任务执行系统，基本都是通过RPC（Remote Procedure Call）远程过程调用来进行进程间方法调用，类似调用本地的方法。flink是基于Akka框架来实现RPC。

Task执行过程

类路径：

https://github.com/apache/flink/blob/master/flink-runtime/src/main/java/org/apache/flink/runtime/taskmanager/Task.java

   // Make sure the user code classloader is accessible thread-locally.

    // We are setting the correct context class loader before instantiating the invokable

    // so that it is available to the invokable during its entire lifetime.

    executingThread.setContextClassLoader(userCodeClassLoader.asClassLoader());

    // When constructing invokable, separate threads can be constructed and thus should be

    // monitored for system exit (in addition to invoking thread itself monitored below).

    FlinkSecurityManager.monitorUserSystemExitForCurrentThread();

    try {

        // now load and instantiate the task's invokable code

        invokable =

                loadAndInstantiateInvokable(

                        userCodeClassLoader.asClassLoader(), nameOfInvokableClass, env);

    } finally {

        FlinkSecurityManager.unmonitorUserSystemExitForCurrentThread();

    }

    // ----------------------------------------------------------------

    //  actual task core work

    // ----------------------------------------------------------------

    // we must make strictly sure that the invokable is accessible to the cancel() call

    // by the time we switched to running.

    this.invokable = invokable;

通过userCodeClassLoade加载user code，然后进行执行。从这一点上可以看出，如果在算子中直接使用外部的变量，这样执行是有问题的。Job提交后，job manager接收到提交的job，开始执行job main方法。相应的算子task会被调度到不同的task manager上运行。都是不同的进程，task manager在执行算子的时候，是获取不到job main方法中的参数的。所以我们需要采用正确的方式传值。如何正确的传参数，可以参考下我另一篇文章。