slachiewicz commented on code in PR #291:
URL:
https://github.com/apache/flink-connector-kafka/pull/291#discussion_r3967578285
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java:
##########
@@ -385,9 +389,13 @@ public void handleSourceEvents(SourceEvent sourceEvent) {
addSplits(validPendingSplits);
pendingSplits.clear();
- if (isNoMoreSplits) {
- notifyNoMoreSplits();
- }
+ }
+
+ // Replay only on the first metadata update. On a later metadata
change the reader must
+ // wait for the enumerator to signal again after the new assignments,
so replaying here
+ // would finish an active reader before the new topic's splits arrive.
Review Comment:
Added note regarding missing re-signal in FLINK-31006 in `5ae82e44`.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java:
##########
@@ -369,6 +369,61 @@ void testNotifyNoMoreSplits() throws Exception {
}
}
+ @Test
+ void testIdleReaderFinishesWhenNoMoreSplitsArrivesBeforeMetadata() throws
Exception {
+ TestingReaderContext context = new TestingReaderContext();
+ try (DynamicKafkaSourceReader<Integer> reader =
createReaderWithoutStart(context)) {
+ TrackingReaderOutput<Integer> readerOutput = new
TrackingReaderOutput<>();
+ reader.start();
+
+ // The enumerator signals no more splits before the reader
receives the metadata
+ // update event, e.g. when an idle reader registers after all
bounded
+ // sub-enumerators have already finished split discovery.
+ reader.notifyNoMoreSplits();
+
+ MetadataUpdateEvent metadata =
+ DynamicKafkaSourceTestHelper.getMetadataUpdateEvent(TOPIC);
+ CompletableFuture<Void> availableBeforeMetadata =
reader.isAvailable();
+ reader.handleSourceEvents(metadata);
+
+ assertThat(availableBeforeMetadata)
+ .as("the metadata update must wake up a task parked on the
earlier future")
+ .isDone();
+ assertThat(reader.pollNext(readerOutput))
+ .as(
+ "idle reader must reach END_OF_INPUT even when
no-more-splits precedes the metadata update event")
+ .isEqualTo(InputStatus.END_OF_INPUT);
+ }
+ }
+
+ @Test
+ void testActiveReaderWaitsForNewSplitsAfterMetadataChange() throws
Exception {
+ TestingReaderContext context = new TestingReaderContext();
+ try (DynamicKafkaSourceReader<Integer> reader =
createReaderWithoutStart(context)) {
+ TrackingReaderOutput<Integer> readerOutput = new
TrackingReaderOutput<>();
+ reader.start();
+
+ // First metadata update: only cluster 0 is known, so the reader
goes active with a
+ // single sub-reader.
+ KafkaStream clusterZeroOnly =
DynamicKafkaSourceTestHelper.getKafkaStream(TOPIC);
+ clusterZeroOnly.getClusterMetadataMap().remove(kafkaClusterId1);
+ reader.handleSourceEvents(
+ new
MetadataUpdateEvent(Collections.singleton(clusterZeroOnly)));
+
+ // The enumerator finished discovery for that metadata epoch.
+ reader.notifyNoMoreSplits();
+
+ // Cluster 1 appears, which creates a sub-reader the earlier
no-more-splits signal
+ // never covered. Its splits are assigned only after the
enumerator signals again.
Review Comment:
Applied suggestion in `5ae82e44`.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java:
##########
@@ -369,6 +369,61 @@ void testNotifyNoMoreSplits() throws Exception {
}
}
+ @Test
+ void testIdleReaderFinishesWhenNoMoreSplitsArrivesBeforeMetadata() throws
Exception {
+ TestingReaderContext context = new TestingReaderContext();
+ try (DynamicKafkaSourceReader<Integer> reader =
createReaderWithoutStart(context)) {
+ TrackingReaderOutput<Integer> readerOutput = new
TrackingReaderOutput<>();
+ reader.start();
+
+ // The enumerator signals no more splits before the reader
receives the metadata
+ // update event, e.g. when an idle reader registers after all
bounded
+ // sub-enumerators have already finished split discovery.
+ reader.notifyNoMoreSplits();
+
+ MetadataUpdateEvent metadata =
+ DynamicKafkaSourceTestHelper.getMetadataUpdateEvent(TOPIC);
+ CompletableFuture<Void> availableBeforeMetadata =
reader.isAvailable();
+ reader.handleSourceEvents(metadata);
+
+ assertThat(availableBeforeMetadata)
+ .as("the metadata update must wake up a task parked on the
earlier future")
+ .isDone();
+ assertThat(reader.pollNext(readerOutput))
+ .as(
+ "idle reader must reach END_OF_INPUT even when
no-more-splits precedes the metadata update event")
+ .isEqualTo(InputStatus.END_OF_INPUT);
+ }
+ }
+
+ @Test
+ void testActiveReaderWaitsForNewSplitsAfterMetadataChange() throws
Exception {
+ TestingReaderContext context = new TestingReaderContext();
+ try (DynamicKafkaSourceReader<Integer> reader =
createReaderWithoutStart(context)) {
+ TrackingReaderOutput<Integer> readerOutput = new
TrackingReaderOutput<>();
+ reader.start();
+
+ // First metadata update: only cluster 0 is known, so the reader
goes active with a
+ // single sub-reader.
+ KafkaStream clusterZeroOnly =
DynamicKafkaSourceTestHelper.getKafkaStream(TOPIC);
+ clusterZeroOnly.getClusterMetadataMap().remove(kafkaClusterId1);
+ reader.handleSourceEvents(
+ new
MetadataUpdateEvent(Collections.singleton(clusterZeroOnly)));
+
+ // The enumerator finished discovery for that metadata epoch.
+ reader.notifyNoMoreSplits();
+
+ // Cluster 1 appears, which creates a sub-reader the earlier
no-more-splits signal
+ // never covered. Its splits are assigned only after the
enumerator signals again.
+
reader.handleSourceEvents(DynamicKafkaSourceTestHelper.getMetadataUpdateEvent(TOPIC));
+
+ assertThat(reader.pollNext(readerOutput))
+ .as(
+ "reader must not finish on the stale
no-more-splits signal while a newly added cluster still awaits its splits")
+ .isNotEqualTo(InputStatus.END_OF_INPUT);
Review Comment:
Applied suggestion in `5ae82e44`.
--
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]
