wombatu-kun commented on code in PR #19202:
URL: https://github.com/apache/hudi/pull/19202#discussion_r3592724014
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/HoodieSourceSplitReader.java:
##########
@@ -77,27 +84,34 @@ public HoodieSourceSplitReader(
@Override
public RecordsWithSplitIds<HoodieRecordWithPosition<T>> fetch() throws
IOException {
- // finish current split.
- if (currentSplit != null) {
- return finishSplit();
+ if (currentSplit == null) {
+ // Limit already satisfied: drain any remaining locally-queued splits as
immediately finished
+ // so that Flink's SourceReaderBase can reach end-of-input cleanly.
+ if (recordLimiter.map(RecordLimiter::isLimitReached).orElse(false)) {
+ return drainRemainingAsSplitsFinished();
+ }
+ HoodieSourceSplit nextSplit = splits.poll();
+ if (nextSplit == null) {
+ // return an empty result, which will lead to split fetch to be idle.
+ // SplitFetcherManager will then close idle fetcher.
+ return new RecordsBySplits<>(Collections.emptyMap(),
Collections.emptySet());
+ }
+ currentSplit = nextSplit;
+ readerFunction.open(currentSplit);
}
- // Limit already satisfied: drain any remaining locally-queued splits as
immediately finished
- // so that Flink's SourceReaderBase can reach end-of-input cleanly.
- if (recordLimiter.map(RecordLimiter::isLimitReached).orElse(false)) {
- return drainRemainingAsSplitsFinished();
+ // Read the next bounded minibatch of the open split, unless the global
limit is already reached.
+ if (!recordLimiter.map(RecordLimiter::isLimitReached).orElse(false)) {
+ BatchRecords<T> batch = readerFunction.readBatch(currentSplit,
DEFAULT_MINI_BATCH_SIZE);
Review Comment:
Done 77551b7. wakeUp() sets a volatile flag the minibatch drain polls
between records, so fetch() returns promptly on a wake-up while the split close
stays on the split-fetcher thread.
##########
hudi-flink-datasource/hudi-flink/src/main/java/org/apache/hudi/source/reader/function/HoodieSplitReaderFunction.java:
##########
@@ -72,26 +70,20 @@ public HoodieSplitReaderFunction(
}
@Override
- public RecordsWithSplitIds<HoodieRecordWithPosition<RowData>>
read(HoodieSourceSplit split) {
- final String splitId = split.splitId();
+ protected ClosableIterator<RowData> createRecordIterator(HoodieSourceSplit
split) {
HoodieTableMetaClient metaClient = StreamerUtil.metaClientForReader(conf,
getHadoopConf());
-
try {
- this.fileGroupReader = createFileGroupReader(split, metaClient);
- final ClosableIterator<RowData> recordIterator =
fileGroupReader.getClosableIterator();
- BatchRecords<RowData> records = BatchRecords.forRecords(splitId,
recordIterator, split.getFileOffset(), split.getConsumed());
- records.seek(split.getConsumed());
- return records;
+ // Closing the returned iterator cascade-closes the whole
HoodieFileGroupReader, so the base
+ // class only has to close the iterator in closeCurrentSplit().
+ return createFileGroupReader(split, metaClient).getClosableIterator();
Review Comment:
Done 77551b7
--
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]
