slachiewicz opened a new pull request, #305: URL: https://github.com/apache/flink-connector-kafka/pull/305
## What is the purpose of the change Fix a race condition in `KafkaSourceEnumerator` where duplicate splits are assigned to readers when the partition discovery interval is shorter than the time `initializePartitionSplits` takes to complete on the worker thread. `getPartitionChange()` only deduplicates against `assignedSplits` and `pendingPartitionSplitAssignment` — it does **not** account for partitions whose split initialization is currently in-flight. When periodic discovery fires before `handlePartitionSplitChanges` runs, the same partitions are treated as "new" again, causing a second `initializePartitionSplits` dispatch and duplicate assignments. **Impact**: This is a correctness bug — users with `partition.discovery.interval.ms` set lower than broker 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 7 and 8 being assigned twice ([run 34200551453](https://github.com/apache/flink-connector-kafka/actions/runs/34200551453/job/101978105599)). ## Brief change log - **`KafkaSourceEnumerator.java`**: Added `Set<TopicPartition> partitionsBeingInitialized` field. Populated in `checkPartitionChanges` before dispatching `callAsync`, cleared in handler wrapper (both success and failure paths), and excluded from deduplication in `getPartitionChange`. - **`KafkaSourceEnumeratorTest.java`**: Added `testInFlightPartitionsNotRediscoveredAsNew` — simulates two consecutive discovery cycles before the first `initializePartitionSplits` completes, verifies no duplicate dispatch and exactly one assignment batch. - **`SourceTopicIntegrityTest.java`**: Raised `DISCOVERY_INTERVAL` from 50ms to 500ms (the test only needs periodic discovery enabled, not sub-100ms) and `ERROR_DISCOVERY_TIMEOUT` from 2s to 20s (aligning with FLINK-40589). ## Verifying this change - New unit test `testInFlightPartitionsNotRediscoveredAsNew` directly reproduces the race by running two periodic discovery cycles without executing the pending `initializePartitionSplits` callable between them. - All existing `KafkaSourceEnumeratorTest` tests pass without modification. ## Does this pull request potentially affect one of the following parts? - Dependencies (does it add or upgrade a dependency): no - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: no - The serializers: no - The runtime per-record code paths (performance sensitive): no ## Documentation - Does this pull request introduce a new feature? no - If yes, how is the feature documented? not applicable -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
