private Batch<T> readBatch(DefaultRecordBatch batch) {
        if (doCrcValidation) {
            // Perform a CRC validity check on this batch
            batch.ensureValid();
        }

        Integer numRecords = batch.countOrNull();
        if (numRecords == null) {
            throw new IllegalStateException("Expected a record count for the records batch");
        }

        InputStream input = batch.recordInputStream(bufferSupplier);
        final Batch<T> result;
        try {
            if (batch.isControlBatch()) {
                List<ControlRecord> records = new ArrayList<>(numRecords);
                for (int i = 0; i < numRecords; i++) {
                    ControlRecord record = readRecord(
                        input,
                        batch.sizeInBytes(),
                        RecordsIterator::decodeControlRecord
                    );
                    records.add(record);
                }
                result = Batch.control(
                    batch.baseOffset(),
                    batch.partitionLeaderEpoch(),
                    batch.maxTimestamp(),
                    batch.sizeInBytes(),
                    records
                );
            } else {
                List<T> records = new ArrayList<>(numRecords);
                for (int i = 0; i < numRecords; i++) {
                    T record = readRecord(input, batch.sizeInBytes(), this::decodeDataRecord);
                    records.add(record);
                }

                result = Batch.data(
                    batch.baseOffset(),
                    batch.partitionLeaderEpoch(),
                    batch.maxTimestamp(),
                    batch.sizeInBytes(),
                    records
                );
            }
        } finally {
            Utils.closeQuietly(input, "BytesStream for input containing records");
        }

        return result;
    }
This class provides an iterator over records retrieved via the raft client or from a snapshot
@param records the records
@param serde the serde to deserialize records
@param bufferSupplier the buffer supplier implementation to allocate buffers when reading records. This must return ByteBuffer allocated on the heap
@param batchSize the maximum batch size
raft/src/main/java/org/apache/kafka/raft/internals/RecordsIterator.java