MartijnVisser commented on code in PR #300:
URL:
https://github.com/apache/flink-connector-kafka/pull/300#discussion_r4054125815
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumeratorTest.java:
##########
@@ -2596,6 +2776,52 @@ public void close() throws Exception {
}
}
+ private static class RecordingSplitEnumeratorContext
+ extends MockSplitEnumeratorContext<DynamicKafkaSourceSplit> {
+ private final Map<Integer, List<Set<String>>> splitsAtCompletion = new
HashMap<>();
+
+ private RecordingSplitEnumeratorContext() {
+ super(NUM_SUBTASKS);
+ }
+
+ @Override
+ public void signalNoMoreSplits(int subtask) {
+ super.signalNoMoreSplits(subtask);
+ splitsAtCompletion
+ .computeIfAbsent(subtask, ignored -> new ArrayList<>())
+ .add(
+ getAssignedSplits(subtask).stream()
+ .map(DynamicKafkaSourceSplit::splitId)
+ .collect(Collectors.toSet()));
+ }
+
+ private List<DynamicKafkaSourceSplit> getAssignedSplits(int reader) {
+ return getSplitsAssignmentSequence().stream()
+ .flatMap(
+ assignment ->
+ assignment
+ .assignment()
+ .getOrDefault(reader,
Collections.emptyList())
+ .stream())
+ .collect(Collectors.toList());
+ }
+
+ private void assertReaderCompleted(int reader, int times) {
+ List<DynamicKafkaSourceSplit> assignedSplits =
getAssignedSplits(reader);
+
assertThat(assignedSplits).hasSize(DynamicKafkaSourceTestHelper.NUM_KAFKA_CLUSTERS);
+ Set<String> assignedSplitIds =
+ assignedSplits.stream()
+ .map(DynamicKafkaSourceSplit::splitId)
+ .collect(Collectors.toSet());
+ assertThat(splitsAtCompletion.get(reader))
+ .hasSize(times)
+ .allSatisfy(
Review Comment:
Asserting the split set at completion rather than the signal count is what
catches a reader being completed early, and it is why this PR carries the dedup
rather than #309.
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/enumerator/DynamicKafkaSourceEnumerator.java:
##########
@@ -785,11 +805,17 @@ private boolean tryCompletePendingReaderRegistration() {
return true;
}
- readerRecoveryGate.markInitialRegistrationComplete();
- if (readerRecoveryGate.hasReportedSplits()) {
- reassignReportedSplits();
- } else {
- flushPendingSplitAssignmentsForRegisteredReaders();
+ // Draining the gate clears its pending state before reassignment
finishes.
Review Comment:
Nit: third `splitAssignmentInProgress` window in this class. A small private
helper taking a `Runnable` would say it once.
--
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]
