彻底搞懂 Flink Kafka OffsetState 存储

// 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);
					}
				}

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

//从 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()));
				}