spuru9 commented on PR #305:
URL:
https://github.com/apache/flink-connector-kafka/pull/305#issuecomment-5605684400
**major** — `KafkaSourceEnumerator.getPartitionChange()`, migration-path
loop (lines 612–617, just outside this PR's diff so I can't anchor inline):
```java
// migration path, ensure that partitions without offset are properly
initialized
for (KafkaPartitionSplit split : unassignedSplits.values()) {
if (split.isMigrated()) {
initialPartitions.add(split.getTopicPartition());
}
}
```
Same race this PR fixes, still unguarded here. A migrated split stays
`isMigrated()` in `unassignedSplits` until `handlePartitionSplitChanges`
overwrites it, so if discovery fires again while its
`initializePartitionSplits` is in flight, this loop re-adds it to
`initialPartitions` and dispatches a second time → duplicate assignment via the
state-migration path. Narrower trigger than the main bug (needs restore from
offset-less legacy state + short discovery interval), but same defect class.
I reproduced the double-dispatch in a unit test (two discovery cycles with
the init callable still pending → `getOneTimeCallables()` was 2, expected 1).
Adding the `partitionsBeingInitialized` guard fixes it and keeps
`shouldLazilyInitializeSplitOffsetsOnMigration` green:
```java
for (KafkaPartitionSplit split : unassignedSplits.values()) {
if (split.isMigrated()
&&
!partitionsBeingInitialized.contains(split.getTopicPartition())) {
initialPartitions.add(split.getTopicPartition());
}
}
```
--
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]
