MartijnVisser commented on code in PR #291:
URL:
https://github.com/apache/flink-connector-kafka/pull/291#discussion_r3956571563
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java:
##########
@@ -385,9 +385,10 @@ public void handleSourceEvents(SourceEvent sourceEvent) {
addSplits(validPendingSplits);
pendingSplits.clear();
- if (isNoMoreSplits) {
- notifyNoMoreSplits();
- }
+ }
+
+ if (isNoMoreSplits) {
Review Comment:
Replaying on every metadata update makes an active reader finish as soon as
its current splits are done, before the splits of a newly added topic arrive. I
added `testActiveReaderWaitsForNewSplitsAfterMetadataChange` to check this: it
passes on main and fails here with `END_OF_INPUT`.
Limiting the replay to the reader's first metadata update keeps your fix and
passes both that test and your
`testIdleReaderFinishesWhenNoMoreSplitsArrivesBeforeMetadata`:
```java
final boolean firstMetadataUpdate = !isActivelyConsumingSplits;
...
if (isNoMoreSplits && firstMetadataUpdate) {
notifyNoMoreSplits();
}
```
For later metadata changes the reader must wait for the enumerator to signal
again after the new assignments. That signal is currently missing for a
recreated sub-enumerator (FLINK-31006), so an unconditional replay would only
mask that as an early finish. Happy to push the test to your branch if you want.
##########
flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReaderTest.java:
##########
@@ -369,6 +369,34 @@ 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);
+ reader.handleSourceEvents(metadata);
+
+ long deadline = System.currentTimeMillis() + 10_000;
+ InputStatus status = reader.pollNext(readerOutput);
+ while (status != InputStatus.END_OF_INPUT &&
System.currentTimeMillis() < deadline) {
Review Comment:
The loop is not needed. After the metadata update the availability future
obtained before it is complete and a single `pollNext` returns `END_OF_INPUT`:
```java
CompletableFuture<Void> availableBeforeMetadata = reader.isAvailable();
reader.handleSourceEvents(metadata);
assertThat(availableBeforeMetadata).isDone();
assertThat(reader.pollNext(readerOutput)).isEqualTo(InputStatus.END_OF_INPUT);
```
This also asserts that the task gets woken up, which is what the ITCase
depends on.
--
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]
