slachiewicz commented on code in PR #305:
URL:
https://github.com/apache/flink-connector-kafka/pull/305#discussion_r4504556188
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java:
##########
@@ -368,9 +374,19 @@ private void checkPartitionChanges(Set<TopicPartition>
fetchedPartitions, Throwa
if (partitionChange.isEmpty()) {
return;
}
+ // Track in-flight partitions to avoid duplicate discovery.
+ final Set<TopicPartition> partitionsInThisBatch = new HashSet<>();
+ partitionsInThisBatch.addAll(partitionChange.getInitialPartitions());
+ partitionsInThisBatch.addAll(partitionChange.getNewPartitions());
+ partitionsBeingInitialized.addAll(partitionsInThisBatch);
+
context.callAsync(
() -> initializePartitionSplits(partitionChange),
- this::handlePartitionSplitChanges);
+ (result, error) -> {
+ // Clear in-flight state on the success and failure paths.
+
partitionsBeingInitialized.removeAll(partitionsInThisBatch);
Review Comment:
You are right: `StoppableKafkaEnumContextProxy` drops the handler call for a
swallowed failure, so the partitions stayed marked until the enumerator was
recreated. If `initializePartitionSplits` throws, the callable now releases the
in-flight state through `runInCoordinatorThread`, and
`testPartitionsReleasedWhenInitializationFailureIsSuppressed` covers fail,
swallow, retry and a single assignment; it fails without the release.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/source/enumerator/KafkaSourceEnumerator.java:
##########
@@ -368,9 +374,19 @@ private void checkPartitionChanges(Set<TopicPartition>
fetchedPartitions, Throwa
if (partitionChange.isEmpty()) {
return;
}
+ // Track in-flight partitions to avoid duplicate discovery.
+ final Set<TopicPartition> partitionsInThisBatch = new HashSet<>();
Review Comment:
Adopted: `partitionInitializationInFlight` replaces both sets and the extra
migration-path condition, because the coordinator is single-threaded. The two
existing regression tests pass unchanged, and the flag is cleared before the
handler runs, as suggested.
--
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]
