MartijnVisser commented on code in PR #291:
URL:
https://github.com/apache/flink-connector-kafka/pull/291#discussion_r3968115719
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java:
##########
@@ -369,6 +369,67 @@ 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. The reader holds no splits, so the metadata
change recreates every
+ // sub-reader and only the reader-level flag still remembers the
earlier signal. Splits and
+ // a fresh no-more-splits signal for the new metadata arrive after
this event.
Review Comment:
```suggestion
// Cluster 1 appears. The reader holds no splits, so the
metadata change recreates every
// sub-reader and only the reader-level flag still remembers the
earlier signal. Splits
// and a fresh no-more-splits signal for the new metadata arrive
after this event.
```
--
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]
