[jira] [Updated] (FLINK-40615) KafkaSourceEnumerator assigns duplicate splits when partition discovery interval is shorter than split initialization time

["ASF GitHub Bot (Jira)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/FLINK-40615?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

ASF GitHub Bot updated FLINK-40615:
-----------------------------------
    Labels: pull-request-available  (was: )

> 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
>            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)