@Override
    public RecordsWithSplitIds<RecordAndPosition<T>> fetch() throws IOException {
        metrics.incrementSplitReaderFetchCalls(1);
        if (currentReader == null) {
            IcebergSourceSplit nextSplit = splits.poll();
            if (nextSplit != null) {
                currentSplit = nextSplit;
                currentSplitId = nextSplit.splitId();
                currentReader = openSplitFunction.apply(currentSplit);
            } else {
                // 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());
            }
        }

        if (currentReader.hasNext()) {
            // Because Iterator#next() doesn't support checked exception,
            // we need to wrap and unwrap the checked IOException with UncheckedIOException
            try {
                ArrayBatchRecords<T> result = (ArrayBatchRecords<T>) currentReader.next();
                metrics.outputMetricsWithEstimate(result);
                return result;
            } catch (UncheckedIOException e) {
                throw e.getCause();
            }
        } else {
            return finishSplit();
        }
    }
Copy from iceberg-flink:iceberg-flink-1.15:1.3.1
inlong-sort/sort-flink/sort-flink-v1.15/sort-connectors/iceberg/src/main/java/org/apache/inlong/sort/iceberg/source/reader/IcebergSourceSplitReader.java