@Override
    public AbstractIterator<MutableRecordBatch> batchIterator() {
        return new RecordBatchIterator<>(new ByteBufferLogInputStream(buffer.duplicate(), Integer.MAX_VALUE));
    }
The total number of bytes in this message set not including any partial, trailing messages. This
may be smaller than what is returned by {@link #sizeInBytes()}.
@return The number of valid bytes
clients/src/main/java/org/apache/kafka/common/record/MemoryRecords.java