private ArrayBatchRecords<T> finishSplit() throws IOException {
        if (currentReader != null) {
            currentReader.close();
            currentReader = null;
        }

        ArrayBatchRecords<T> finishRecords = ArrayBatchRecords.finishedSplit(currentSplitId);
        LOG.info("Split reader {} finished split: {}", indexOfSubtask, currentSplitId);
        metrics.incrementFinishedSplits(1);
        metrics.incrementFinishedBytes(calculateBytes(currentSplit));
        currentSplitId = null;
        return finishRecords;
    }
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