Efrat19 commented on code in PR #285:
URL:
https://github.com/apache/flink-connector-kafka/pull/285#discussion_r3614134479
##########
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:
It seems like this fix addresses a different issue than the one described in
the ticket?
were you able to reproduce this in actual env / ITCase?
Otherwise we should probably avoid this cross components notification on the
checkpoint interval 🤔
##########
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:
ditto - maybe consider a different ticket
--
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]
