MartijnVisser commented on code in PR #323:
URL:
https://github.com/apache/flink-connector-kafka/pull/323#discussion_r4350687215
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java:
##########
@@ -638,8 +653,20 @@ private KafkaSourceEnumerator
createEnumeratorWithAssignedTopicPartitions(
kafkaClusterId,
kafkaMetadataService,
signalNoMoreSplitsCallback);
- KafkaSourceEnumerator.SplitOwnerSelector splitOwnerSelector =
+
context.setReaderReadyForAssignment(readerRecoveryGate::isReaderReadyForAssignment);
+ context.setSplitAssignmentListener(
+ (reader, split) -> currentSplitOwners.put(split.splitId(),
reader));
+ KafkaSourceEnumerator.SplitOwnerSelector strategyOwnerSelector =
splitAssignmentStrategy.createSplitOwnerSelector(kafkaClusterId);
+ KafkaSourceEnumerator.SplitOwnerSelector splitOwnerSelector =
+ strategyOwnerSelector == null
+ ? null
Review Comment:
A retained split handed straight back to its reader gets no
`currentSplitOwners` entry, so with `per_cluster` the hash owner takes it when
the holder restarts after a re-add. Unbounded, nobody gets it; bounded, it is
skipped. Recording the owner on those direct assignments and using the map in
every mode fixes both.
--
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]
