Sylwester Lachiewicz created FLINK-40615:
--------------------------------------------
Summary: 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
{{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)
