[FLINK-21219] FlinkKafkaConsumer ignores offset overrides for new topics when restoring from savepoint.

• Type: Bug
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 1.12.1
• Component/s: Connectors / Kafka
• Labels: auto-deprioritized-major auto-deprioritized-minor
• Assignee: Unassigned
• Reporter: Dominik Wosiński
• Created: 01/Feb/21 00:28
• Updated: 06/Nov/21 10:39

Description

Currently, when FlinkKafkaConsumer is restored from savepoint, the following code will handle topics that do not have offsets committed (for example if a new topic was added):

if (restoredState != null) { for (KafkaTopicPartition partition : allPartitions) { if (!restoredState.containsKey(partition)) { restoredState.put(partition, KafkaTopicPartitionStateSentinel.EARLIEST_OFFSET); } }

 

So if we have a KafkaConsumer with topicPattern and the pattern is changed, new topis will always start from earliest offset, even if originally the setting was different.