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."); }