slachiewicz commented on code in PR #291:
URL:
https://github.com/apache/flink-connector-kafka/pull/291#discussion_r3961080386
##########
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:
Applied in `4eded31c`, with the capture placed above the
`isActivelyConsumingSplits = true` block so the flag still reads the pre-update
state.
Verified: removing the `firstMetadataUpdate` guard fails
`testActiveReaderWaitsForNewSplitsAfterMetadataChange` and leaves
`testIdleReaderFinishesWhenNoMoreSplitsArrivesBeforeMetadata` passing, so the
new test pins exactly the early finish you described. Wrote it on the branch
rather than taking the push, thanks for the offer.
*This comment was created with AI assistance.*
##########
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:
Applied in `4eded31c` — loop replaced with the pre-update availability
future plus a single `pollNext`.
*This comment was created with AI assistance.*
--
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]
