@Override
    public CloseableIterator<RecordsWithSplitIds<RecordAndPosition<T>>> batch(
            String splitId, DataIterator<T> inputIterator) {
        Preconditions.checkArgument(inputIterator != null, "Input data iterator can't be null");
        // lazily create pool as it is not serializable
        if (pool == null) {
            this.pool = createPoolOfBatches(handoverQueueSize);
        }
        return new ArrayPoolBatchIterator(splitId, inputIterator, pool);
    }
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