简介

FlinkKafkaConsumer为Flink消费Kafka数据的连接器。在Flink中的角色为数据源。

FlinkKafkaConsumer的继承结构

如下图所示：

我们发现FlinkKafkaConsumer继承自FlinkKafkaConsumerBase。FlinkKafkaConsumerBase又实现了SourceFunction和RichFunction接口。接下来我们重点分析它的open和run方法。

FlinkKafkaConsumerBase的open方法

该方法包含的内容为FlinkKafkaConsumer的初始化逻辑。
首先设置提交offset的模式

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

OffsetCommitMode是一个枚举类型，具有如下三个值：

  • DISABLED：完全禁用offset提交。
  • ON_CHECKPOINTS：当checkpoint完成的时候再提交offset。
  • KAFKA_PERIODIC：周期性提交offset。

判断OffsetCommitMode的逻辑封装在了OffsetCommitModes.fromConfiguration方法中。该方法的代码如下：

  1. /**
  2.      * Determine the offset commit mode using several configuration values.
  3.      *
  4.      * @param enableAutoCommit whether or not auto committing is enabled in the provided Kafka properties.
  5.      * @param enableCommitOnCheckpoint whether or not committing on checkpoints is enabled.
  6.      * @param enableCheckpointing whether or not checkpoint is enabled for the consumer.
  7.      *
  8.      * @return the offset commit mode to use, based on the configuration values.
  9.      */
  10. public static OffsetCommitMode fromConfiguration(
  11.     boolean enableAutoCommit,
  12.     boolean enableCommitOnCheckpoint,
  13.     boolean enableCheckpointing) {
  15.     if (enableCheckpointing) {
  16.         // if checkpointing is enabled, the mode depends only on whether committing on checkpoints is enabled
  17.         return (enableCommitOnCheckpoint) ? OffsetCommitMode.ON_CHECKPOINTS : OffsetCommitMode.DISABLED;
  18.     } else {
  19.         // else, the mode depends only on whether auto committing is enabled in the provided Kafka properties
  20.         return (enableAutoCommit) ? OffsetCommitMode.KAFKA_PERIODIC : OffsetCommitMode.DISABLED;

这段代码逻辑可以总结为：

  1. 如果启用了checkpoint，并且启用了checkpoint完成时提交offset，返回ON_CHECKPOINTS。
  2. 如果未启用checkpoint，但是启用了自动提交，返回KAFKA_PERIODIC。
  3. 其他情况都返回DISABLED。 ```java // create the partition discoverer this.partitionDiscoverer = createPartitionDiscoverer( topicsDescriptor, getRuntimeContext().getIndexOfThisSubtask(), getRuntimeContext().getNumberOfParallelSubtasks()); this.partitionDiscoverer.open();

subscribedPartitionsToStartOffsets = new HashMap<>(); final List allPartitions = partitionDiscoverer.discoverPartitions();

  1. 这段代码中topicsDescriptor为fixedTopics和topicPattern的封装。其中fixedTopics明确指定了topic的名称，称为固定topic。topicPattern为匹配topic名称的正则表达式，用于分区发现。
  3. **createPartitionDiscoverer**方法创建了一个KafkaPartitionDiscoverer对象，主要负责Kafka分区发现。partitionDiscoverer.open()方法创建出一个KafkaConsumer。
  5. **subscribedPartitionsToStartOffsets** 为已订阅的分区列表，这里将它初始化。<br />partitionDiscoverer.discoverPartitions()用户获取所有fixedTopics和匹配topicPattern的Topic包含的所有分区信息。该部分代码稍后分析。
  7. 接下来**open**方法的代码结构如下：
  8. ```java
  9. if (restoredState != null) {
  10. // 从快照恢复逻辑...
  11. } else {
  12. // 直接启动逻辑...
  13. }

如果consumer是从快照恢复的，restoredState不为空。反之restoredState为空。
我们首先分析一下从快照恢复的逻辑。代码如下：

  1. // 如果restoredState没有存储某一分区的状态
  2. // 需要重头消费该分区
  3. for (KafkaTopicPartition partition : allPartitions) {
  4.     if (!restoredState.containsKey(partition)) {
  5.         restoredState.put(partition, KafkaTopicPartitionStateSentinel.EARLIEST_OFFSET);
  6.     }
  7. }
  9. for (Map.Entry<KafkaTopicPartition, Long> restoredStateEntry : restoredState.entrySet()) {
  10.     // seed the partition discoverer with the union state while filtering out
  11.     // restored partitions that should not be subscribed by this subtask
  12.     // 此处可过滤掉不归该task负责的kafka分区
  13.     if (KafkaTopicPartitionAssigner.assign(
  14.         restoredStateEntry.getKey(), getRuntimeContext().getNumberOfParallelSubtasks())
  15.         == getRuntimeContext().getIndexOfThisSubtask()){
  16.         subscribedPartitionsToStartOffsets.put(restoredStateEntry.getKey(), restoredStateEntry.getValue());
  17.     }
  18. }
  20. // 依照分区发现配置的topic正则表达式过滤分区
  21. if (filterRestoredPartitionsWithCurrentTopicsDescriptor) {
  22.     // 过滤掉topic名称不符合topicsDescriptor的topicPattern的分区
  23.     subscribedPartitionsToStartOffsets.entrySet().removeIf(entry -> {
  24.         if (!topicsDescriptor.isMatchingTopic(entry.getKey().getTopic())) {
  25.             LOG.warn(
  26.                 "{} is removed from subscribed partitions since it is no longer associated with topics descriptor of current execution.",
  27.                 entry.getKey());
  28.             return true;
  29.         }
  30.         return false;
  31.     });
  32. }

接下来我们分析下Consumer直接启动的逻辑（不从快照恢复）。
在此之前需要了解下StartupMode这个枚举类型。该枚举类型有5个值：

  • GROUP_OFFSETS：从保存在zookeeper或者是Kafka broker的对应消费者组提交的offset开始消费，这个是默认的配置
  • EARLIEST：尽可能从最早的offset开始消费
  • LATEST：从最近的offset开始消费
  • TIMESTAMP：从用户提供的timestamp处开始消费
  • SPECIFIC_OFFSETS：从用户提供的offset处开始消费

然后，Comsumer使用分区发现工具来获取初始的分区。根据StartupMode来设置它们的起始消费offset。

我们先看SPECIFIC_OFFSETS这种情况。

  1. case SPECIFIC_OFFSETS:
  2. // 如果没有配置具体从哪个offset开始消费，程序抛出异常
  3. if (specificStartupOffsets == null) {
  4.     throw new IllegalStateException(
  5.         "Startup mode for the consumer set to " + StartupMode.SPECIFIC_OFFSETS +
  6.         ", but no specific offsets were specified.");
  7. }
  9. for (KafkaTopicPartition seedPartition : allPartitions) {
  10.     // 获取每个分区指定的消费起始offset
  11.     Long specificOffset = specificStartupOffsets.get(seedPartition);
  12.     if (specificOffset != null) {
  13.         // since the specified offsets represent the next record to read, we subtract
  14.         // it by one so that the initial state of the consumer will be correct
  15.         // 如果分区配置了offset，设置从offset开始消费
  16.         subscribedPartitionsToStartOffsets.put(seedPartition, specificOffset - 1);
  17.     } else {
  18.         // default to group offset behaviour if the user-provided specific offsets
  19.         // do not contain a value for this partition
  20.         // 如果分区没有配置offset，设置从GROUP_OFFSET开始消费
  21.         subscribedPartitionsToStartOffsets.put(seedPartition, KafkaTopicPartitionStateSentinel.GROUP_OFFSET);
  22.     }
  23. }
  25. break;

如果采用TIMESTAMP模式，逻辑如下所示：

  1. case TIMESTAMP:
  2. // 如果没有配置timestamp，程序报错退出
  3. if (startupOffsetsTimestamp == null) {
  4.     throw new IllegalStateException(
  5.         "Startup mode for the consumer set to " + StartupMode.TIMESTAMP +
  6.         ", but no startup timestamp was specified.");
  7. }
  9. // 根据timestamp获取分区的offset
  10. // 遍历这些分区
  11. for (Map.Entry<KafkaTopicPartition, Long> partitionToOffset
  12.                             : fetchOffsetsWithTimestamp(allPartitions, startupOffsetsTimestamp).entrySet()) {
  13.                         // 如果无offset，使用LATEST_OFFSET
  14.                         // 如果获取到了offset，从这个offset开始消费
  15.                         subscribedPartitionsToStartOffsets.put(
  16.                             partitionToOffset.getKey(),
  17.                             (partitionToOffset.getValue() == null)
  18.                                     // if an offset cannot be retrieved for a partition with the given timestamp,
  19.                                     // we default to using the latest offset for the partition
  20.                                     ? KafkaTopicPartitionStateSentinel.LATEST_OFFSET
  21.                                     // since the specified offsets represent the next record to read, we subtract
  22.                                     // it by one so that the initial state of the consumer will be correct
  23.                                     : partitionToOffset.getValue() - 1);
  24.                     }
  26.                     break;
  27. // 其他情况，使用KafkaTopicPartitionStateSentinel类对应的值作为offset
  28. default:
  29. for (KafkaTopicPartition seedPartition : allPartitions) {
  30.     subscribedPartitionsToStartOffsets.put(seedPartition, startupMode.getStateSentinel());
  31. }

接下来的if语句段负责打印一些日志信息。这里就不再分析了。

FlinkKafkaConsumerBase的run方法

run方法包含了从KafkaConsumer消费数据，和向Flink下游发送数据的逻辑。
首先检查open方法中初始化的subscribedPartitionsToStartOffsets是否为null。

  1. if (subscribedPartitionsToStartOffsets == null) {
  2.     throw new Exception("The partitions were not set for the consumer");
  3. }

接下来配置成功commit和失败commit数量的监控。

  1. // initialize commit metrics and default offset callback method
  2. // 设置成功提交计数监控
  3. this.successfulCommits = this.getRuntimeContext().getMetricGroup().counter(COMMITS_SUCCEEDED_METRICS_COUNTER);
  4. // 设置失败提交计数监控
  5. this.failedCommits =  this.getRuntimeContext().getMetricGroup().counter(COMMITS_FAILED_METRICS_COUNTER);
  6. // 获取子任务index
  7. final int subtaskIndex = this.getRuntimeContext().getIndexOfThisSubtask();
  9. // 注册一个提交时的回调函数
  10. this.offsetCommitCallback = new KafkaCommitCallback() {
  11.     @Override
  12.     public void onSuccess() {
  13.         // 提交成功，成功提交计数器加一
  14.         successfulCommits.inc();
  15.     }
  17.     @Override
  18.     public void onException(Throwable cause) {
  19.         LOG.warn(String.format("Consumer subtask %d failed async Kafka commit.", subtaskIndex), cause);
  20.         // 提交失败，失败提交计数器加一
  21.         failedCommits.inc();
  22.     }
  23. };

接下来判断subscribedPartitionsToStartOffsets集合是否为空。如果为空，标记数据源的状态为暂时空闲。

  1. if (subscribedPartitionsToStartOffsets.isEmpty()) {
  2.     sourceContext.markAsTemporarilyIdle();
  3. }

下面是获取数据的过程。这里创建了一个KafkaFetcher，负责借助KafkaConsumer API从Kafka broker获取数据。

  1. this.kafkaFetcher = createFetcher(
  2.     sourceContext,
  3.     subscribedPartitionsToStartOffsets,
  4.     watermarkStrategy,
  5.     (StreamingRuntimeContext) getRuntimeContext(),
  6.     offsetCommitMode,
  7.     getRuntimeContext().getMetricGroup().addGroup(KAFKA_CONSUMER_METRICS_GROUP),
  8.     useMetrics);

接下来检测running变量的状态。如果没有running，直接返回。

  1. if (!running) {
  2.     return;
  3. }

最后是根据分区发现间隔时间的配置来确定是否启动分区的定时发现任务。

  1. if (discoveryIntervalMillis == PARTITION_DISCOVERY_DISABLED) {
  2.     // 直接启动获取数据任务
  3.     kafkaFetcher.runFetchLoop();
  4. } else {
  5.     // 否则，启动定期分区发现任务和数据获取任务
  6.     runWithPartitionDiscovery();
  7. }

最后我们分析下runWithPartitionDiscovery方法。代码如下：

  1. private void runWithPartitionDiscovery() throws Exception {
  2.     final AtomicReference<Exception> discoveryLoopErrorRef = new AtomicReference<>();
  3.     // 启动分区发现定时任务
  4.     createAndStartDiscoveryLoop(discoveryLoopErrorRef);
  6.     // 启动kafka broker数据获取任务
  7.     kafkaFetcher.runFetchLoop();
  9.     // make sure that the partition discoverer is waked up so that
  10.     // the discoveryLoopThread exits
  11.     // 使partitionDiscoverer.discoverPartitions()抛异常
  12.     // 能够从discoveryLoopThread 返回
  13.     partitionDiscoverer.wakeup();
  14.     // 等待discoveryLoopThread 执行完毕
  15.     joinDiscoveryLoopThread();
  17.     // rethrow any fetcher errors
  18.     final Exception discoveryLoopError = discoveryLoopErrorRef.get();
  19.     if (discoveryLoopError != null) {
  20.         throw new RuntimeException(discoveryLoopError);
  21.     }
  22. }

我们再跟踪下看看如何启动分区发现定时任务的。

  1. private void createAndStartDiscoveryLoop(AtomicReference<Exception> discoveryLoopErrorRef) {
  2.     // 创建一个discoveryLoop线程
  3.     discoveryLoopThread = new Thread(() -> {
  4.         try {
  5.             // --------------------- partition discovery loop ---------------------
  7.             // throughout the loop, we always eagerly check if we are still running before
  8.             // performing the next operation, so that we can escape the loop as soon as possible
  10.             while (running) {
  11.                 if (LOG.isDebugEnabled()) {
  12.                     LOG.debug("Consumer subtask {} is trying to discover new partitions ...", getRuntimeContext().getIndexOfThisSubtask());
  13.                 }
  15.                 final List<KafkaTopicPartition> discoveredPartitions;
  16.                 try {
  17.                     // 尝试发现新分区，如果方法抛出异常，退出循环
  18.                     discoveredPartitions = partitionDiscoverer.discoverPartitions();
  19.                 } catch (AbstractPartitionDiscoverer.WakeupException | AbstractPartitionDiscoverer.ClosedException e) {
  20.                     // the partition discoverer may have been closed or woken up before or during the discovery;
  21.                     // this would only happen if the consumer was canceled; simply escape the loop
  22.                     break;
  23.                 }
  25.                 // no need to add the discovered partitions if we were closed during the meantime
  26.                 // 如果没有发现新的分区，或者数据源已关闭之时，没必要再添加新分区
  27.                 if (running && !discoveredPartitions.isEmpty()) {
  28.                     kafkaFetcher.addDiscoveredPartitions(discoveredPartitions);
  29.                 }
  31.                 // do not waste any time sleeping if we're not running anymore
  32.                 if (running && discoveryIntervalMillis != 0) {
  33.                     try {
  34.                         // 睡眠discoveryIntervalMillis时间
  35.                         Thread.sleep(discoveryIntervalMillis);
  36.                     } catch (InterruptedException iex) {
  37.                         // may be interrupted if the consumer was canceled midway; simply escape the loop
  38.                         break;
  39.                     }
  40.                 }
  41.             }
  42.         } catch (Exception e) {
  43.             discoveryLoopErrorRef.set(e);
  44.         } finally {
  45.             // calling cancel will also let the fetcher loop escape
  46.             // (if not running, cancel() was already called)
  47.             if (running) {
  48.                 cancel();
  49.             }
  50.         }
  51.     }, "Kafka Partition Discovery for " + getRuntimeContext().getTaskNameWithSubtasks());
  52.     // 启动分区发现定时任务线程
  53.     discoveryLoopThread.start();
  54. }

我们再详细研究下上述方法中partitionDiscoverer.discoverPartitions()的调用，即发现分区的执行过程。代码如下：

  1. public List<KafkaTopicPartition> discoverPartitions() throws WakeupException, ClosedException {
  2.     // 确保没有关闭数据源，也没有wakeup
  3.     if (!closed && !wakeup) {
  4.         try {
  5.             List<KafkaTopicPartition> newDiscoveredPartitions;
  7.             // (1) get all possible partitions, based on whether we are subscribed to fixed topics or a topic pattern
  8.             // 如果配置了fixedTopic，获取这些topic的分区
  9.             if (topicsDescriptor.isFixedTopics()) {
  10.                 newDiscoveredPartitions = getAllPartitionsForTopics(topicsDescriptor.getFixedTopics());
  11.             } else {
  12.                 // 如果没有配置fixedTopic
  13.                 // 1. 获取所有topic
  14.                 List<String> matchedTopics = getAllTopics();
  16.                 // retain topics that match the pattern
  17.                 // 2. 逐个排除名字不是fixedTopic，或名字不匹配topicPattern的topic
  18.                 Iterator<String> iter = matchedTopics.iterator();
  19.                 while (iter.hasNext()) {
  20.                     if (!topicsDescriptor.isMatchingTopic(iter.next())) {
  21.                         iter.remove();
  22.                     }
  23.                 }
  25.                 if (matchedTopics.size() != 0) {
  26.                     // get partitions only for matched topics
  27.                     // 3. 如果有匹配的topic，获取他们的分区
  28.                     newDiscoveredPartitions = getAllPartitionsForTopics(matchedTopics);
  29.                 } else {
  30.                     // 否则newDiscoveredPartitions 设置为null
  31.                     newDiscoveredPartitions = null;
  32.                 }
  33.             }
  35.             // (2) eliminate partition that are old partitions or should not be subscribed by this subtask
  36.             if (newDiscoveredPartitions == null || newDiscoveredPartitions.isEmpty()) {
  37.                 throw new RuntimeException("Unable to retrieve any partitions with KafkaTopicsDescriptor: " + topicsDescriptor);
  38.             } else {
  39.                 Iterator<KafkaTopicPartition> iter = newDiscoveredPartitions.iterator();
  40.                 KafkaTopicPartition nextPartition;
  41.                 while (iter.hasNext()) {
  42.                     nextPartition = iter.next();
  43.                     // 分区存入discoveredPartitions集合中
  44.                     // 返回值为分区是否归当前task消费
  45.                     if (!setAndCheckDiscoveredPartition(nextPartition)) {
  46.                         iter.remove();
  47.                     }
  48.                 }
  49.             }
  51.             return newDiscoveredPartitions;
  52.         } catch (WakeupException e) {
  53.             // the actual topic / partition metadata fetching methods
  54.             // may be woken up midway; reset the wakeup flag and rethrow
  55.             wakeup = false;
  56.             throw e;
  57.         }
  58.     } else if (!closed && wakeup) {
  59.         // may have been woken up before the method call
  60.         wakeup = false;
  61.         throw new WakeupException();
  62.     } else {
  63.         throw new ClosedException();
  64.     }
  65. }

kafkaFetcher的runFetchLoop方法

此方法为FlinkKafkaConsumer获取数据的主入口，通过一个循环来不断获取kafka broker的数据。

  1. public void runFetchLoop() throws Exception {
  2.     try {
  3.         // kick off the actual Kafka consumer
  4.         // 启动kafka消费线程，定期从kafkaConsumer拉取数据并转交给handover对象
  5.         consumerThread.start();
  7.         while (running) {
  8.             // this blocks until we get the next records
  9.             // it automatically re-throws exceptions encountered in the consumer thread
  10.             // 获取handover中的数据
  11.             // 如果此时consumerThread尚未把数据交给handover，该方法会阻塞
  12.             final ConsumerRecords<byte[], byte[]> records = handover.pollNext();
  14.             // get the records for each topic partition
  15.             for (KafkaTopicPartitionState<T, TopicPartition> partition : subscribedPartitionStates()) {
  17.                 // 获取属于该分区的records
  18.                 List<ConsumerRecord<byte[], byte[]>> partitionRecords =
  19.                     records.records(partition.getKafkaPartitionHandle());
  21.                 partitionConsumerRecordsHandler(partitionRecords, partition);
  22.             }
  23.         }
  24.     }
  25.     finally {
  26.         // this signals the consumer thread that no more work is to be done
  27.         consumerThread.shutdown();
  28.     }
  30.     // on a clean exit, wait for the runner thread
  31.     try {
  32.         consumerThread.join();
  33.     }
  34.     catch (InterruptedException e) {
  35.         // may be the result of a wake-up interruption after an exception.
  36.         // we ignore this here and only restore the interruption state
  37.         Thread.currentThread().interrupt();
  38.     }
  39. }

此方法中的collect kafka数据的逻辑在partitionConsumerRecordsHandler中。我们查看下它的代码：

  1. protected void partitionConsumerRecordsHandler(
  2.     List<ConsumerRecord<byte[], byte[]>> partitionRecords,
  3.     KafkaTopicPartitionState<T, TopicPartition> partition) throws Exception {
  5.     // 反序列化Kafka record为bean
  6.     // 此deserializer需要实现KafkaDeserializationSchema接口
  7.     for (ConsumerRecord<byte[], byte[]> record : partitionRecords) {
  8.         deserializer.deserialize(record, kafkaCollector);
  10.         // emit the actual records. this also updates offset state atomically and emits
  11.         // watermarks
  12.         // 发送数据，更新offset，生成timestamp和watermark
  13.         emitRecordsWithTimestamps(
  14.             kafkaCollector.getRecords(),
  15.             partition,
  16.             record.offset(),
  17.             record.timestamp());
  19.         // 如果数据源已到末尾，停止fetcher循环
  20.         if (kafkaCollector.isEndOfStreamSignalled()) {
  21.             // end of stream signaled
  22.             running = false;
  23.             break;
  24.         }
  25.     }
  26. }

它调用了emitRecordWithTimestamp方法，继续查看。

  1. protected void emitRecordsWithTimestamps(
  2.     Queue<T> records,
  3.     KafkaTopicPartitionState<T, KPH> partitionState,
  4.     long offset,
  5.     long kafkaEventTimestamp) {
  6.     // emit the records, using the checkpoint lock to guarantee
  7.     // atomicity of record emission and offset state update
  8.     synchronized (checkpointLock) {
  9.         T record;
  10.         while ((record = records.poll()) != null) {
  11.             // 此处调用SourceFunction中的sourceContext
  12.             // 数据源收集元素逻辑在此
  13.             long timestamp = partitionState.extractTimestamp(record, kafkaEventTimestamp);
  14.             sourceContext.collectWithTimestamp(record, timestamp);
  16.             // this might emit a watermark, so do it after emitting the record
  17.             // 可能定时发送水位线
  18.             partitionState.onEvent(record, timestamp);
  19.         }
  20.         partitionState.setOffset(offset);
  21.     }
  22. }

KafkaConsumerThread

KafkaConsumerThread负责在单独的线程中从Kafka中拉取数据到handover。这里我们分析下它的run方法中获取数据的部分。

  1. // main fetch loop
  2. while (running) {
  4.     // check if there is something to commit
  5.     // 检查是否则commit过程中
  6.     if (!commitInProgress) {
  7.         // get and reset the work-to-be committed, so we don't repeatedly commit the same
  8.         // 获取需要提交的offset值，以及commit回调函数
  9.         // 获取完毕之后需要设置为null，防止反复提交
  10.         final Tuple2<Map<TopicPartition, OffsetAndMetadata>, KafkaCommitCallback> commitOffsetsAndCallback =
  11.             nextOffsetsToCommit.getAndSet(null);
  13.         if (commitOffsetsAndCallback != null) {
  14.             log.debug("Sending async offset commit request to Kafka broker");
  16.             // also record that a commit is already in progress
  17.             // the order here matters! first set the flag, then send the commit command.
  18.             // 开始提交过程
  19.             commitInProgress = true;
  20.             // 异步提交offset
  21.             consumer.commitAsync(commitOffsetsAndCallback.f0, new CommitCallback(commitOffsetsAndCallback.f1));
  22.         }
  23.     }
  25.     // 为consumer指定新的分区
  26.     // 由于分区发现功能的存在，consumer需要添加新发现的分区，否则poll数据会报错
  27.     try {
  28.         if (hasAssignedPartitions) {
  29.             newPartitions = unassignedPartitionsQueue.pollBatch();
  30.         }
  31.         else {
  32.             // if no assigned partitions block until we get at least one
  33.             // instead of hot spinning this loop. We rely on a fact that
  34.             // unassignedPartitionsQueue will be closed on a shutdown, so
  35.             // we don't block indefinitely
  36.             newPartitions = unassignedPartitionsQueue.getBatchBlocking();
  37.         }
  38.         if (newPartitions != null) {
  39.             reassignPartitions(newPartitions);
  40.         }
  41.     } catch (AbortedReassignmentException e) {
  42.         continue;
  43.     }
  45.     if (!hasAssignedPartitions) {
  46.         // Without assigned partitions KafkaConsumer.poll will throw an exception
  47.         continue;
  48.     }
  50.     // get the next batch of records, unless we did not manage to hand the old batch over
  51.     if (records == null) {
  52.         try {
  53.             // 从consumer拉取数据
  54.             // 这里的pollTimeout可以通过配置flink.poll-timeout参数修改
  55.             // pollTimeout默认值为100ms
  56.             records = consumer.poll(pollTimeout);
  57.         }
  58.         catch (WakeupException we) {
  59.             continue;
  60.         }
  61.     }
  63.     try {
  64.         // 将数据交给handover
  65.         handover.produce(records);
  66.         records = null;
  67.     }
  68.     catch (Handover.WakeupException e) {
  69.         // fall through the loop
  70.     }
  71. }
  72. // end main fetch loop
  73. }
  74. catch (Throwable t) {
  75.     // let the main thread know and exit
  76.     // it may be that this exception comes because the main thread closed the handover, in
  77.     // which case the below reporting is irrelevant, but does not hurt either
  78.     handover.reportError(t);
  79. }
  80. finally {
  81.     // make sure the handover is closed if it is not already closed or has an error
  82.     handover.close();
  84.     // make sure the KafkaConsumer is closed
  85.     try {
  86.         consumer.close();
  87.     }
  88.     catch (Throwable t) {
  89.         log.warn("Error while closing Kafka consumer", t);
  90.     }
  91. }

checkpoint流程

checkpoint流程大部分代码为状态的读写。这里为大家总结下主要的流程，不分析具体的代码。

snapshotState方法

FlinkKafkaConsumerBase的snapshotState方法包含snapshot的流程。包含如下：

  1. 如果KafkaFetcher尚未初始化完毕。需要保存已订阅的topic连同他们的初始offset。
  2. 如果KafkaFetcher已初始化完毕，调用fetcher的snapshotCurrentState方法。
  3. 如果offsetCommitMode为ON_CHECKPOINTS类型，还需要将topic和offset写入到pendingOffsetsToCommit集合中。该集合用于checkpoint成功的时候向Kafka broker提交offset。（offsetCommitMode不为ON_CHECKPOINTS和DISABLED的时候，使用的是自动提交offset的模式）

    notifyCheckpointComplete方法

    在所有的operator都快照成功的时候，会向JobManager的CheckpointCoordinator发送确认消息，然后coordinator会通知各个operator checkpoint已经完成。（详细请参见Flink 源码之快照
    ）为了保证保证数据不会被遗漏和重复消费，ON_CHECKPOINTS模式运行的FlinkKafkaConsumer只能在这个时候提交offset到kafka consumer。调用notifyCheckpointComplete的时候通知kafka consumer，将checkpoint之时保存的各个分区的offset提交给kafka broker。从而保证数据的一致性。