写给大忙人看的Flink 消费 Kafka 已经对 Flink 消费 kafka 进行了源码级别的讲解。可是有一点没有说的很明白那就是 offset 是怎么存储到状态中的？
Kafka Offset 是如何存储在 state 中的
在 写给大忙人看的Flink 消费 Kafka 的基础上继续往下说。
代码语言：javascript复制// get the records for each topic partition
// 我们知道 partitionDiscoverer.discoverPartitions 已经保证了 subscribedPartitionStates 仅仅包含该 task 的 KafkaTopicPartition
for (KafkaTopicPartitionState<TopicPartition> partition : subscribedPartitionStates()) {
//仅仅取出属于该 task 的数据
List<ConsumerRecord<byte[], byte[]>> partitionRecords =
records.records(partition.getKafkaPartitionHandle());
for (ConsumerRecord<byte[], byte[]> record : partitionRecords) {
//传进来的 deserializer. 即自定义 deserializationSchema
final T value = deserializer.deserialize(record);
//当我们自定义 deserializationSchema isEndOfStream 设置为 true 的时候，整个流程序就停掉了
if (deserializer.isEndOfStream(value)) {
// end of stream signaled
running = false;
break;
}
// emit the actual record. this also updates offset state atomically
// and deals with timestamps and watermark generation
emitRecord(value, partition, record.offset(), record);
}
}其中 subscribedPartitionStates 方法实际上是获取属性 subscribedPartitionStates。
继续往下追踪，一直到
代码语言：javascript复制protected void emitRecordWithTimestamp(
T record, KafkaTopicPartitionState<KPH> partitionState, long offset, long timestamp) throws Exception {
if (record != null) {
// 没有 watermarks
if (timestampWatermarkMode == NO_TIMESTAMPS_WATERMARKS) {
// fast path logic, in case there are no watermarks generated in the fetcher
// emit the record, using the checkpoint lock to guarantee
// atomicity of record emission and offset state update
synchronized (checkpointLock) {
sourceContext.collectWithTimestamp(record, timestamp);
// 设置 state 中的 offset( 实际上设置 subscribedPartitionStates 而当 snapshotState 时，获取 subscribedPartitionStates 中的值进行 snapshotState)
partitionState.setOffset(offset);
}
} else if (timestampWatermarkMode == PERIODIC_WATERMARKS) {
emitRecordWithTimestampAndPeriodicWatermark(record, partitionState, offset, timestamp);
} else {
emitRecordWithTimestampAndPunctuatedWatermark(record, partitionState, offset, timestamp);
}
} else {
// if the record is null, simply just update the offset state for partition
synchronized (checkpointLock) {
partitionState.setOffset(offset);
}
}
}当 sourceContext 发送完这条消息的时候，才设置 offset 到 subscribedPartitionStates 中。
而当 FlinkKafkaConsumer 做 Snapshot 时，会从 fetcher 中获取 subscribedPartitionStates。
代码语言：javascript复制//从 fetcher subscribedPartitionStates 中获取相应的值
HashMap<KafkaTopicPartition, Long> currentOffsets = fetcher.snapshotCurrentState();
if (offsetCommitMode == OffsetCommitMode.ON_CHECKPOINTS) {
// the map cannot be asynchronously updated, because only one checkpoint call can happen
// on this function at a time: either snapshotState() or notifyCheckpointComplete()
pendingOffsetsToCommit.put(context.getCheckpointId(), currentOffsets);
}
for (Map.Entry<KafkaTopicPartition, Long> kafkaTopicPartitionLongEntry : currentOffsets.entrySet()) {
unionOffsetStates.add(
Tuple2.of(kafkaTopicPartitionLongEntry.getKey(), kafkaTopicPartitionLongEntry.getValue()));
}至此进行 checkpoint 时，相应的 offset 就存入了 state。
