@Override
        public RecordsWithSplitIds<RecordAndPosition<T>> next() {
            if (!inputIterator.hasNext()) {
                throw new NoSuchElementException();
            }

            T[] batch = getCachedEntry();
            int recordCount = 0;
            while (inputIterator.hasNext() && recordCount < batchSize) {
                // The record produced by inputIterator can be reused like for the RowData case.
                // inputIterator.next() can't be called again until the copy is made
                // since the record is not consumed immediately.
                T nextRecord = inputIterator.next();
                recordFactory.clone(nextRecord, batch, recordCount);
                recordCount++;
                if (!inputIterator.currentFileHasNext()) {
                    // break early so that records in the ArrayResultIterator
                    // have the same fileOffset.
                    break;
                }
            }

            return ArrayBatchRecords.forRecords(
                    splitId,
                    pool.recycler(),
                    batch,
                    recordCount,
                    inputIterator.fileOffset(),
                    inputIterator.recordOffset() - recordCount);
        }
This implementation stores record batch in array from recyclable pool

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/ArrayPoolDataIteratorBatcher.java