[
https://issues.apache.org/jira/browse/FLINK-40615?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
Martijn Visser reassigned FLINK-40615:
--------------------------------------
Assignee: Sylwester Lachiewicz
> KafkaSourceEnumerator assigns duplicate splits when partition discovery
> interval is shorter than split initialization time
> --------------------------------------------------------------------------------------------------------------------------
>
> Key: FLINK-40615
> URL: https://issues.apache.org/jira/browse/FLINK-40615
> Project: Flink
> Issue Type: Bug
> Components: Connectors / Kafka
> Reporter: Sylwester Lachiewicz
> Assignee: Sylwester Lachiewicz
> Priority: Major
> Labels: pull-request-available
>
> {{KafkaSourceEnumerator.getPartitionChange()}} deduplicates discovered
> partitions only against {{assignedSplits}} and
> {{pendingPartitionSplitAssignment}}. It does *not* account for partitions
> whose split initialization ({{initializePartitionSplits}}) is currently
> in-flight on the worker thread.
> When periodic discovery fires faster than {{initializePartitionSplits}}
> completes (which performs blocking {{listOffsets}} RPCs via AdminClient),
> {{checkPartitionChanges}} treats the same partitions as "new" again and
> dispatches a second {{initializePartitionSplits}} call.
> *Impact*: Duplicate splits are assigned to readers, causing duplicate record
> consumption. Users with {{partition.discovery.interval.ms}} set lower than
> the broker's metadata response time can silently get duplicate records.
> *Observed in CI*:
> {{SourceTopicIntegrityTest.testTopicIntegritySuccess[PATTERN]}} with
> {{DISCOVERY_INTERVAL=50ms}} — found 120 records instead of 100 due to
> partitions being assigned twice. See [run 34200551453, job
> 101978105599|https://github.com/apache/flink-connector-kafka/actions/runs/34200551453/job/101978105599].
> *Root cause timeline*:
> # Discovery callable returns 10 partitions → {{checkPartitionChanges}}
> dispatches {{initializePartitionSplits}} to worker (call A)
> # 50ms later, discovery fires again → {{getPartitionChange}} finds same 10
> partitions absent from {{assignedSplits}}/{{pendingPartitionSplitAssignment}}
> (call A still running) → dispatches duplicate {{initializePartitionSplits}}
> (call B)
> # Both calls' handlers run → partitions assigned twice → readers produce
> duplicate records
> *Fix*: Track in-flight partitions in a {{Set<TopicPartition>
> partitionsBeingInitialized}} field. Populated before dispatching
> {{callAsync}} in {{checkPartitionChanges}}, cleared in the handler wrapper
> regardless of success/failure, and excluded from {{getPartitionChange}}
> deduplication.
--
This message was sent by Atlassian Jira
(v8.20.10#820010)
