Flink是如何儲存Offset的

摘要： Flink對Offset的管理，有兩種方式： 1.Checkpointing disabled 完全依賴於kafka自身的API 2.Checkpointing enabled 當checkpoint做完的時候，會將offset提交給kafka or zk ...

Flink對Offset的管理，有兩種方式：

1.Checkpointing disabled 完全依賴於kafka自身的API

2.Checkpointing enabled 當checkpoint做完的時候，會將offset提交給kafka or zk

本文只針對於第二種，Checkpointing enabled

FlinkKafkaConsumerBase中的 notifyCheckpointComplete

@Override
//當checkpoint完成的時候，此方法會被呼叫
public final void notifyCheckpointComplete(long checkpointId) throws Exception {
if (!running) {
LOG.debug("notifyCheckpointComplete() called on closed source");
return;
}

final AbstractFetcher<?, ?> fetcher = this.kafkaFetcher;
if (fetcher == null) {
LOG.debug("notifyCheckpointComplete() called on uninitialized source");
return;
}

if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
// only one commit operation must be in progress
if (LOG.isDebugEnabled()) {
LOG.debug("Committing offsets to Kafka/ZooKeeper for checkpoint " + checkpointId);
}

try {
final int posInMap = pendingOffsetsToCommit.indexOf(checkpointId);
if (posInMap == -1) {
LOG.warn("Received confirmation for unknown checkpoint id {}", checkpointId);
return;
}

@SuppressWarnings("unchecked")
Map<KafkaTopicPartition, Long> offsets =
(Map<KafkaTopicPartition, Long>) pendingOffsetsToCommit.remove(posInMap);

// remove older checkpoints in map
for (int i = 0; i < posInMap; i++) {
pendingOffsetsToCommit.remove(0);
}

if (offsets == null || offsets.size() == 0) {
LOG.debug("Checkpoint state was empty.");
return;
}

//通過kafkaFetcher提交offsetfetcher.commitInternalOffsetsToKafka(offsets, offsetCommitCallback);
} catch (Exception e) {
if (running) {
throw e;
}
// else ignore exception if we are no longer running
}
}
}

跳轉到kafkaFetcher

@Override
protected void doCommitInternalOffsetsToKafka(
Map<KafkaTopicPartition, Long> offsets,
@Nonnull KafkaCommitCallback commitCallback) throws Exception {

@SuppressWarnings("unchecked")
List<KafkaTopicPartitionState<TopicPartition>> partitions = subscribedPartitionStates();

Map<TopicPartition, OffsetAndMetadata> offsetsToCommit = new HashMap<>(partitions.size());

for (KafkaTopicPartitionState<TopicPartition> partition : partitions) {
Long lastProcessedOffset = offsets.get(partition.getKafkaTopicPartition());
if (lastProcessedOffset != null) {
checkState(lastProcessedOffset >= 0, "Illegal offset value to commit");

// committed offsets through the KafkaConsumer need to be 1 more than the last processed offset.
// This does not affect Flink's checkpoints/saved state.
long offsetToCommit = lastProcessedOffset + 1;

offsetsToCommit.put(partition.getKafkaPartitionHandle(), new OffsetAndMetadata(offsetToCommit));
partition.setCommittedOffset(offsetToCommit);
}
}

// record the work to be committed by the main consumer thread and make sure the consumer notices that
consumerThread.setOffsetsToCommit(offsetsToCommit, commitCallback);
}

可以看到呼叫consumerThread.setOffsetsToCommit方法

void setOffsetsToCommit(
Map<TopicPartition, OffsetAndMetadata> offsetsToCommit,
@Nonnull KafkaCommitCallback commitCallback) {

// record the work to be committed by the main consumer thread and make sure the consumer notices that
/*
!=null的時候，說明kafkaConsumerThread更新的太慢了，新的將會覆蓋old
當此處執行的時候，kafkaconsumerThread中consumer.commitAsync()

這個方法還是關鍵的方法，直接給nextOffsetsToCommit賦值了
nextOffsetsToCommit，我們可以看到是AtomicReference，可以原子更新物件的引用
*/

if (nextOffsetsToCommit.getAndSet(Tuple2.of(offsetsToCommit, commitCallback)) != 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.");
}

// if the consumer is blocked in a poll() or handover operation, wake it up to commit soon
handover.wakeupProducer();

synchronized (consumerReassignmentLock) {
if (consumer != null) {
consumer.wakeup();
} else {
// the consumer is currently isolated for partition reassignment;
// set this flag so that the wakeup state is restored once the reassignment is complete
hasBufferedWakeup = true;
}
}
}

nextOffsetsToCommit已經有值了，接下我們來看一下KafkaConsumerThread的run方法

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

......
// main fetch loop
while (running) {

// check if there is something to commit
//default false
if (!commitInProgress) {
// get and reset the work-to-be committed, so we don't repeatedly commit the same
//setCommittedOffset方法已經給nextOffsetsToCommit賦值了，這裡進行獲取，所以commitOffsetsAndCallback is not null
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));
}
}

....
}

至此offset就更新完畢了，我們可以很清楚的看到，當checkpoint完成時，呼叫相關的commit方法，將kafka offset提交至kafka broker