Efrat19 commented on code in PR #305:
URL:
https://github.com/apache/flink-connector-kafka/pull/305#discussion_r4262358955
##########
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:
Since the coordinator is single threaded, wdyt about simply deferring
`initializePartitionSplits` while partitions are being initiliazed?
It would only require a boolean like `partitionInitializationInFlight`,
avoiding both `partitionsBeingInitialized` and `partitionsInThisBatch`
hashsets, wdyt?
```
private boolean partitionInitializationInFlight;
private void checkPartitionChanges(Set<TopicPartition> fetchedPartitions,
Throwable t) {
if (t != null) {
throw new FlinkRuntimeException(
"Failed to list subscribed topic partitions due to ",
t);
}
if (partitionInitializationInFlight) {
LOG.debug("Skipping discovery result because split
initialization is in flight.");
return;
}
final PartitionChange partitionChange =
getPartitionChange(fetchedPartitions,
!initialDiscoveryFinished);
if (partitionChange.isEmpty()) {
return;
}
partitionInitializationInFlight = true;
context.callAsync(
() -> initializePartitionSplits(partitionChange),
(result, error) -> {
// Clear before handling so a throwing handler cannot
leave the flag set.
partitionInitializationInFlight = false;
handlePartitionSplitChanges(result, error);
});
}
```
--
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]
