slachiewicz commented on code in PR #285:
URL:
https://github.com/apache/flink-connector-kafka/pull/285#discussion_r3751325908
##########
flink-connector-kafka/src/main/java/org/apache/flink/connector/kafka/dynamic/source/reader/DynamicKafkaSourceReader.java:
##########
@@ -385,9 +385,14 @@ public void handleSourceEvents(SourceEvent sourceEvent) {
addSplits(validPendingSplits);
pendingSplits.clear();
- if (isNoMoreSplits) {
- notifyNoMoreSplits();
- }
+ }
+
+ // Re-deliver a no-more-splits signal that arrived before this
metadata update, since
+ // sub-readers created above have not seen it. This must also happen
when there are no
+ // pending splits at all: an idle reader that registers after all
bounded sub-enumerators
+ // finished discovery receives no-more-splits first and would
otherwise never finish.
+ if (isNoMoreSplits) {
Review Comment:
You were right — moved to #291 under FLINK-40362, where the reproduction and
the notification-frequency question are answered.
##########
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 {
Review Comment:
You were right — moved to #291 under FLINK-40362, where the reproduction and
the notification-frequency question are answered.
--
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]
