[
https://issues.apache.org/jira/browse/FLINK-38453?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18023822#comment-18023822
]
Arvid Heise commented on FLINK-38453:
-------------------------------------
Hi [~loserwang1024] , thank you for being active on the Kafka connector and
advancing the FLIP. I have opened a PR nonetheless because I already was pretty
far. But I will wait for your review in any case.
Things to consider:
* Your FLIP goes much farther than what my PR tries to do. I'm merely trying
to make the best use of the existing infrastructure and your are advanding the
framework.
* Hence your solution will only work with more recent Flink versions after the
FLIP is fully implemented.
* My proposal can be backported to earlier branches that support 1.X.
* I don't fundamentally change anything, so I'm sure everything that you
develop will be applicable as well.
* We also need a rather quick solution so that we can fully use the
HybridSource with KafkaSource, which is currently not possible because recovery
will loose offsets of unassigned splits.
Please consider advancing my proposal now and enhance everything with our
solution later.
> KafkaEnumerator doesn't restore offsets of owned split
> ------------------------------------------------------
>
> Key: FLINK-38453
> URL: https://issues.apache.org/jira/browse/FLINK-38453
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Affects Versions: kafka-5.0.0
> Reporter: Arvid Heise
> Assignee: Arvid Heise
> Priority: Major
> Labels: pull-request-available
>
> KafkaEnumerator's state contains the TopicPartitions only but not the
> offsets, so it doesn't contain the full split state contrary to the design
> intent.
> There are a couple of issues with that approach. It implicitly assumes that
> splits are fully assigned to readers before the first checkpoint. Else the
> enumerator will invoke the offset initializer again on recovery from such a
> checkpoint leading to inconsistencies (LATEST may be initialized during the
> first attempt for some partitions and initialized during second attempt for
> others).
> Through addSplitBack callback, you may also get these scenarios later for
> BATCH which actually leads to duplicate rows (in case of EARLIEST or
> SPECIFIC-OFFSETS) or data loss (in case of LATEST). Finally, it's not
> possible to safely use KafkaSource as part of a HybridSource because the
> offset initializer cannot even be recreated on recovery.
> All cases are solved by also retaining the offset in the enumerator state.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
