bowenli86 commented on code in PR #323:
URL:
https://github.com/apache/flink-connector-kafka/pull/323#discussion_r4351184030
##########
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:
Fixed through ce209935. Direct retained assignments now record their
physical owner, and both modes honor that owner on restart. Ownership and local
report merging use cluster ID plus topic-partition, so distinct physical splits
cannot alias through their formatted IDs.
I added bounded/unbounded regressions for your restore/re-add sequence with
two successive holder restarts, plus collision cases with shared or separate
holders. They reproduce lost splits before the corresponding fixes; all 90
focused tests pass, preserving numeric offsets and assignment-before-completion
ordering. Initial recovery now batches returns per cluster as well.
The description explicitly explains reader-local recovery and bounded
completion timing, while retaining the initial all-reader restore barrier. CI
is pending on this head.
--
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]
