前言
这是很多Flink用户不太注意的一个隐藏得比较深的坑（严格来讲不算bug），近期组内同学频繁踩坑，故十分有必要快速记录一下，以提请注意。
复现问题
用户意图：将有状态Flink流作业的部分或全部Source Topic重置消费位点，实现数据补录。
操作步骤：
- 停止作业，在消息队列管理平台设定新位点；
- 改变新位点对应的Kafka Consumer的UID，使状态记录的旧位点失效；
- 从最新Checkpoint恢复作业。
预期现象：作业成功恢复，并从预定的位点开始消费。
实际现象：作业成功恢复，但是从对应topic的EARLIEST位点开始消费，数据大量积压。
下面图表中橙色折线为消费者的积压量，蓝色箭头为用户手动重置位点的时间，红色箭头为改动UID的任务恢复并完成第一次Checkpoint的时间，此时积压量几乎等于Topic消息总量。
另外，在TaskManager中也可发现类似如下的日志，说明实际恢复的位点是EARLIEST，而非用户指定的位点。
2025-02-12 18:49:32.643 INFO [Source:: waybill_rou:r (7/20)#0] o.a.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase - Consumer subtask 6 restored state: {}.
2025-02-13 18:49:34.714 INFO [Legacy Source Thread: (7/20)#0] o.a.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase - Consumer subtask 6 creating fetcher with offsets {KafkaTopicPartition{topic='my_topic', partition=38}=-915623761775}.
分析原因
我们知道，FlinkKafkaConsumer将位点信息储存在内部的OperatorState中，当FlinkKafkaConsumer实例初始化时，会一并初始化状态，并填充KafkaTopicPartition及其对应的Offset。代码如下。
// org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase
@Override
public final void initializeState(FunctionInitializationContext context) throws Exception {
OperatorStateStore stateStore = context.getOperatorStateStore();
this.unionOffsetStates =
stateStore.getUnionListState(
new ListStateDescriptor<>(
OFFSETS_STATE_NAME,
createStateSerializer(getRuntimeContext().getExecutionConfig())));
if (context.isRestored()) {
restoredState = new TreeMap<>(new KafkaTopicPartition.Comparator());
// populate actual holder for restored state
for (Tuple2<KafkaTopicPartition, Long> kafkaOffset : unionOffsetStates.get()) {
restoredState.put(kafkaOffset.f0, kafkaOffset.f1);
}
LOG.info(
"Consumer subtask {} restored state: {}.",
getRuntimeContext().getIndexOfThisSubtask(),
restoredState);
} else {
LOG.info(
"Consumer subtask {} has no restore state.",
getRuntimeContext().getIndexOfThisSubtask());
}
}
当用户修改了FlinkKafkaConsumer算子的UID后，再从状态恢复作业，此时无法再取得unionOffsetStates，故恢复的状态容器restoredState是一个空的Map。这样又会导致Kafka Consumer启动时无法从restoredState取得任何Partition Offset信息，故将所有分区都填充为EARLIEST_OFFSET（见如下代码中标注感叹号的部分），引发问题。
@Override
public void open(Configuration configuration) throws Exception {
// determine the offset commit mode
this.offsetCommitMode =
OffsetCommitModes.fromConfiguration(
getIsAutoCommitEnabled(),
enableCommitOnCheckpoints,
((StreamingRuntimeContext) getRuntimeContext()).isCheckpointingEnabled());
// create the partition discoverer
this.partitionDiscoverer =
createPartitionDiscoverer(
topicsDescriptor,
getRuntimeContext().getIndexOfThisSubtask(),
getRuntimeContext().getNumberOfParallelSubtasks());
this.partitionDiscoverer.open();
subscribedPartitionsToStartOffsets = new HashMap<>();
final List<KafkaTopicPartition> allPartitions = partitionDiscoverer.discoverPartitions();
if (restoredState != null) {
for (KafkaTopicPartition partition : allPartitions) {
// [!]
if (!restoredState.containsKey(partition)) {
restoredState.put(partition, KafkaTopicPartitionStateSentinel.EARLIEST_OFFSET);
}
}
for (Map.Entry<KafkaTopicPartition, Long> restoredStateEntry :
restoredState.entrySet()) {
// seed the partition discoverer with the union state while filtering out
// restored partitions that should not be subscribed by this subtask
if (KafkaTopicPartitionAssigner.assign(
restoredStateEntry.getKey(),
getRuntimeContext().getNumberOfParallelSubtasks())
== getRuntimeContext().getIndexOfThisSubtask()) {
subscribedPartitionsToStartOffsets.put(
restoredStateEntry.getKey(), restoredStateEntry.getValue());
}
}
if (filterRestoredPartitionsWithCurrentTopicsDescriptor) {
// 移除此Sub-Task不需要再订阅的Partition，代码略
}
LOG.info(
"Consumer subtask {} will start reading {} partitions with offsets in restored state: {}",
getRuntimeContext().getIndexOfThisSubtask(),
subscribedPartitionsToStartOffsets.size(),
subscribedPartitionsToStartOffsets);
} else {
// use the partition discoverer to fetch the initial seed partitions,
// and set their initial offsets depending on the startup mode.
// for SPECIFIC_OFFSETS and TIMESTAMP modes, we set the specific offsets now;
// for other modes (EARLIEST, LATEST, and GROUP_OFFSETS), the offset is lazily
// determined
// when the partition is actually read.
// 以下根据不同的startupMode参数处理，代码略去
}
this.deserializer.open(
RuntimeContextInitializationContextAdapters.deserializationAdapter(
getRuntimeContext(), metricGroup -> metricGroup.addGroup("user")));
}
处理方法
最稳妥的方法当然是不要更改任何有状态作业的Source UID。
如果有强需求的话，我们可以新增一个Kafka Property参数（例如properties.restore-from-group-offsets），并修改FlinkKafkaConsumerBase#open()方法的逻辑，在UID改变且参数为true时，使用GROUP_OFFSET标记填充restoredState，而非默认的EARLIEST_OFFSET。简单方便。
