一个task异常后，一个job是如何failed的

1. 算子异常后，所在的task会捕获异常，并将task进行cancel

   1. 位置：org.apache.flink.runtime.taskmanager.Task#cancelInvokable
      2.

2. task进行cancel后，会通知JM当前task已经failed，并把异常告诉给JM

   1. 位置：org.apache.flink.runtime.taskmanager.Task#notifyFinalState
   2. 

3. JM收到TM的通知（有个task failed了）

   1. 位置：org.apache.flink.runtime.jobmaster.JobMaster#updateTaskExecutionState

   2. 

   3. schedulerNG.updateTaskExecutionState(taskExecutionState)执行，会将executionGraph图上的所有task遍历进行cancel

      1. SchedulerBase#updateTaskExecutionStateInternal--》
         DefaultScheduler#maybeHandleTaskFailure--》
         DefaultScheduler#handleTaskFailure--》
         SchedulerBase#failJob--》
         ExecutionGraph#cancelVerticesAsync--》
         ExecutionJobVertex#cancelWithFuture--》
         org.apache.flink.runtime.executiongraph.ExecutionVertex#cancel--》
         org.apache.flink.runtime.executiongraph.Execution#startCancelling--》
         org.apache.flink.runtime.executiongraph.Execution#sendCancelRpcCall--》
         org.apache.flink.runtime.jobmanager.slots.TaskManagerGateway#cancelTask
         

      2. TM收到JM发送的cancel请求

         1. org.apache.flink.runtime.taskexecutor.TaskExecutor#cancelTask--》
            org.apache.flink.runtime.taskmanager.Task#cancelExecution--》
            启动一个cancelThread线程调用TaskCanceler（Runnable）进行task的cancel--》
            org.apache.flink.runtime.jobgraph.tasks.AbstractInvokable#cancel--》
            //将streamTask循环发送邮件的循环条件设置为false
            org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor#allActionsCompleted
            

         2. TM会将task进行cancel

            1. 将streamTask的循环收收发邮件的条件破坏
            2. streamTask跳出循环进行收尾
            3. streamTask判断是否cancel，是的话就抛出cancelTaskException
            4. 然后会被task捕获，然后也会调用org.apache.flink.runtime.taskmanager.Task#notifyFinalState，通知JM（当前task已经cancel了），但是不同的是JM收到此task的cancel，不需要通知executionGraph上其他的task进行相关操作（task的Failed就需要通知其他task）
               1. 判断位置：org.apache.flink.runtime.scheduler.SchedulerBase#isNotifiable