[ https://issues.apache.org/jira/browse/FLINK-21219?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17276277#comment-17276277 ]
enrico agnoli commented on FLINK-21219: --------------------------------------- Should this be a new separate configuration, similar to `flink.partition-discovery.interval-millis` from https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kafka.html#topic-discovery ? Something like `flink.partition-discovery.policy` or `flink.partition-discovery.start-position` > FlinkKafkaConsumer ignores offset overrides for new topics when restoring > from savepoint. > ----------------------------------------------------------------------------------------- > > Key: FLINK-21219 > URL: https://issues.apache.org/jira/browse/FLINK-21219 > Project: Flink > Issue Type: Bug > Components: Connectors / Kafka > Affects Versions: 1.12.1 > Reporter: Dominik Wosiński > Priority: Major > > 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): > {noformat} > if (restoredState != null) { for (KafkaTopicPartition partition : > allPartitions) { if (!restoredState.containsKey(partition)) { > restoredState.put(partition, > KafkaTopicPartitionStateSentinel.EARLIEST_OFFSET); } }{noformat} > > 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. -- This message was sent by Atlassian Jira (v8.3.4#803005)