private T decodeDataRecord(Optional<ByteBuffer> key, Optional<ByteBuffer> value) {
        if (key.isPresent()) {
            throw new IllegalArgumentException("Got key in the record when no key was expected");
        }

        if (!value.isPresent()) {
            throw new IllegalArgumentException("Missing value in the record when a value was expected");
        } else if (value.get().remaining() == 0) {
            throw new IllegalArgumentException("Got an unexpected empty value in the record");
        }

        ByteBuffer valueBuffer = value.get();

        return serde.read(new ByteBufferAccessor(valueBuffer), valueBuffer.remaining());
    }
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