flink源码分析之kafka consumer的执行流程

WARN  org.apache.flink.streaming.connectors.kafka.internal.Kafka010Fetcher  - Committing offsets to Kafka takes longer than the checkpoint interval. Skipping commit of previous offsets because newer complete checkpoint offsets are available. This does not compromise Flink's checkpoint integrity.

    /**
     * The core work method that bootstraps the task and executes its code.
     */
    @Override
    public void run() {
        try {
            doRun();
        } finally {
            terminationFuture.complete(executionState);
        }
    }

try {
            beforeInvoke();
            // final check to exit early before starting to run
            if (canceled) {
                throw new CancelTaskException();
            }
            // let the task do its work
            runMailboxLoop();
            // if this left the run() method cleanly despite the fact that this was canceled,
            // make sure the "clean shutdown" is not attempted
            if (canceled) {
                throw new CancelTaskException();
            }
            afterInvoke();
        }
        //----------省略部分代码

      protected void initializeStateAndOpenOperators(StreamTaskStateInitializer streamTaskStateInitializer) throws Exception {
          for (StreamOperatorWrapper<?, ?> operatorWrapper : getAllOperators(true)) {
              StreamOperator<?> operator = operatorWrapper.getStreamOperator();
              operator.initializeState(streamTaskStateInitializer);
              operator.open();
          }
      }

      @Override
      public void open() throws Exception {
          super.open();
          FunctionUtils.openFunction(userFunction, new Configuration());
      }

  // determine the offset commit mode
          this.offsetCommitMode = OffsetCommitModes.fromConfiguration(
              getIsAutoCommitEnabled(),
              enableCommitOnCheckpoints,
              ((StreamingRuntimeContext) getRuntimeContext()).isCheckpointingEnabled());

  public void runMailboxLoop() throws Exception {
          mailboxProcessor.runMailboxLoop();
      }

  // 创建 mailboxProcessor
          this.mailboxProcessor = new MailboxProcessor(this::processInput, mailbox, actionExecutor);

      /**
       * Runs the mailbox processing loop. This is where the main work is done.
       */
      public void runMailboxLoop() throws Exception {
          // 邮箱
          final TaskMailbox localMailbox = mailbox;

          Preconditions.checkState(
              localMailbox.isMailboxThread(),
              "Method must be executed by declared mailbox thread!");

          assert localMailbox.getState() == TaskMailbox.State.OPEN : "Mailbox must be opened!";
          // 邮箱controller，与processor绑定
          final MailboxController defaultActionContext = new MailboxController(this);
          // 当mailbox循环处于运行状态时，会一直消费Mailbox中的message（实际上是一个FIFO的队列）
          while (isMailboxLoopRunning()) {
              // The blocking `processMail` call will not return until default action is available.
              processMail(localMailbox, false);
              if (isMailboxLoopRunning()) {
                  // 进行 task 的 default action，也就是调用 processInput()
                  // 这里的defaultAction是在StreamTask的构造方法中的this.mailboxProcessor = new MailboxProcessor(this::processInput, mailbox, actionExecutor)中的this::processInput。
                  mailboxDefaultAction.runDefaultAction(defaultActionContext); // lock is acquired inside default action as needed
              }
          }
      }

this.consumerThread = new KafkaConsumerThread(
            LOG,
            handover,
            kafkaProperties,
            unassignedPartitionsQueue,
            getFetcherName() + " for " + taskNameWithSubtasks,
            pollTimeout,
            useMetrics,
            consumerMetricGroup,
            subtaskMetricGroup);

   public void run() {
           // early exit check
           if (!running) {
               return;
           }

           // this is the means to talk to FlinkKafkaConsumer's main thread
           final Handover handover = this.handover;

           // This method initializes the KafkaConsumer and guarantees it is torn down properly.
           // This is important, because the consumer has multi-threading issues,
           // including concurrent 'close()' calls.
           try {
               // 获取consumer，每次来获取都是new一个新的KafkaConsumer
               this.consumer = getConsumer(kafkaProperties);
           }
           catch (Throwable t) {
               handover.reportError(t);
               return;
           }
           // -------------省略-------

       // -------省略代码---------------
       // main fetch loop
               while (running) {

                   // check if there is something to commit
                   if (!commitInProgress) {// 默认为false
                       // get and reset the work-to-be committed, so we don't repeatedly commit the same
                       final Tuple2<Map<TopicPartition, OffsetAndMetadata>, KafkaCommitCallback> commitOffsetsAndCallback =
                               nextOffsetsToCommit.getAndSet(null);

                       if (commitOffsetsAndCallback != null) {// 如果不为空，代表上次的提交没有完成，这里会继续进行异步提交
                           log.debug("Sending async offset commit request to Kafka broker");

                           // also record that a commit is already in progress
                           // the order here matters! first set the flag, then send the commit command.
                           commitInProgress = true;
                           consumer.commitAsync(commitOffsetsAndCallback.f0, new CommitCallback(commitOffsetsAndCallback.f1));
                       }
                   }
                   try {
                       if (hasAssignedPartitions) {
                           // 如果已经分配好分区，查看是否有未分配的partition，分配分区的
                           newPartitions = unassignedPartitionsQueue.pollBatch();
                       }
                       else {
                           // if no assigned partitions block until we get at least one
                           // instead of hot spinning this loop. We rely on a fact that
                           // unassignedPartitionsQueue will be closed on a shutdown, so
                           // we don't block indefinitely
                           newPartitions = unassignedPartitionsQueue.getBatchBlocking();
                       }
                       if (newPartitions != null) {
                           reassignPartitions(newPartitions);
                       }
                   } catch (AbortedReassignmentException e) {
                       continue;
                   }
                   if (!hasAssignedPartitions) {
                       // Without assigned partitions KafkaConsumer.poll will throw an exception
                       continue;
                   }
                   // get the next batch of records, unless we did not manage to hand the old batch over
                   if (records == null) {
                       try {
                           records = consumer.poll(pollTimeout);
                       }
                       catch (WakeupException we) {
                           continue;
                       }
                   }
                   try {
                       handover.produce(records);
                       records = null;
                   }
                   //---------省略代码

    // record the work to be committed by the main consumer thread and make sure the consumer notices that
        if (nextOffsetsToCommit.getAndSet(Tuple2.of(offsetsToCommit, commitCallback)) != null) {//设置新值，返回老值，老值是否为null
            log.warn("Committing offsets to Kafka takes longer than the checkpoint interval. " +
                    "Skipping commit of previous offsets because newer complete checkpoint offsets are available. " +
                    "This does not compromise Flink's checkpoint integrity.");
        }