[
https://issues.apache.org/jira/browse/FLINK-41100?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=18131242#comment-18131242
]
yongfu.gao commented on FLINK-41100:
------------------------------------
Hi, I'd like to take this one.
The id is still built as clusterId + "-" + kafkaPartitionSplit.splitId() in four
places: DynamicKafkaSourceSplit#splitId(), DynamicKafkaSourceEnumerator
(~L1294),
and KafkaPartitionSplitReaderWrapper (splitId() and finishedSplits()). Nothing
parses these ids, and DynamicKafkaSourceSplitSerializer stores clusterId +
KafkaPartitionSplit, so the format can change without breaking checkpoint state.
Plan: one formatting helper with a collision-free encoding used at all four
sites, plus a mock-based reader test with (cluster "a", topic "b-c") and
(cluster "a-b", topic "c") asserting distinct outputs, no premature output
release, and per-partition watermarks.
Do you want GlobalSplitOwnerAssigner moved to the (clusterId, TopicPartition)
key from FLINK-41088 in the same change, or left as is?
> DynamicKafkaSource shares one split output between partitions whose split ids
> collide
> -------------------------------------------------------------------------------------
>
> Key: FLINK-41100
> URL: https://issues.apache.org/jira/browse/FLINK-41100
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Affects Versions: kafka-3.2.0, kafka-4.0.1, kafka-5.0.0
> Reporter: Martijn Visser
> Priority: Major
>
> {{DynamicKafkaSourceSplit#splitId}} is {{<cluster>-<topic>-<partition>}}, so
> cluster {{a}} with topic {{b-c}} and cluster {{a-b}} with topic {{c}} both
> get {{a-b-c-0}}. The sub-readers of {{DynamicKafkaSourceReader}} emit into
> the same {{ReaderOutput}} under that id ({{KafkaPartitionSplitReaderWrapper}}
> prefixes the cluster id), and {{createOutputForSplit}} returns the existing
> output for an id it already knows. When one reader holds both partitions,
> they share one split-local watermark generator, and when one of them
> finishes, {{SourceReaderBase}} releases the output the other one still writes
> to. From then on that partition no longer holds back the combined watermark,
> so its records can be late. Watermark alignment pauses and resumes both
> partitions together for the same reason.
> The split ids used towards Flink should be unique per physical partition.
> FLINK-41088 fixes the same collision in the enumerator's recovery paths.
--
This message was sent by Atlassian Jira
(v10.3.26#10030026)
