danny0405 commented on code in PR #19202:
URL: https://github.com/apache/hudi/pull/19202#discussion_r3592423843
##########
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:
**[P1] Preserve the `SplitReader.wakeUp()` contract now that `fetch()`
performs I/O.** This call now drains up to 2,048 I/O-backed records inside
`fetch()`, but `HoodieSourceSplitReader.wakeUp()` is still a no-op. The Flink
`SplitReader` contract requires a blocking `fetch()` to unblock when `wakeUp()`
is invoked, and `SplitFetcher.shutdown()` uses exactly that path before waiting
for the fetcher thread. A cancellation during a slow or stalled
`hasNext()`/`next()` can therefore wait for the whole minibatch—or
indefinitely—before `closeCurrentSplit()` is reached. Could we add a wake-up
flag that stops materialization between records, leaving the actual close on
the fetcher thread, so cancellation remains responsive without reintroducing
cross-thread teardown?
##########
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:
**[P2] Close the file-group reader if iterator initialization fails.**
`getClosableIterator()` calls `initRecordIterators()` and can throw after
opening and assigning the base iterator. With the reader now created only as
this temporary, that partially initialized reader becomes unreachable and
`AbstractSplitReaderFunction.close()` has nothing to close. Before this change,
`fileGroupReader` was assigned to the field before `getClosableIterator()`, so
fetcher teardown still closed it. Please retain the local
`HoodieFileGroupReader` and close it in the failure path, preserving the
initialization exception and suppressing any close failure.
--
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]
