danny0405 commented on code in PR #18325:
URL: https://github.com/apache/hudi/pull/18325#discussion_r2938493526
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/BatchRecords.java:
##########
@@ -117,8 +115,11 @@ public void seek(long startingRecordOffset) {
public static <T> BatchRecords<T> forRecords(
String splitId, ClosableIterator<T> recordIterator, int fileOffset, long
startingRecordOffset) {
-
- return new BatchRecords<>(
- splitId, recordIterator, fileOffset, startingRecordOffset, new
HashSet<>());
+ // Pre-populate finishedSplits with splitId so that FetchTask calls
splitFinishedCallback
+ // immediately after enqueueing the batch. This removes the split from
+ // SplitFetcher.assignedSplits, causing the fetcher to idle and invoke
+ // elementsQueue.notifyAvailable(), which is required to drive the
END_OF_INPUT signal
Review Comment:
cc @cshuo did you know the standard way to handle the bounded source end up
for source v2?
--
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]
