private static ControlRecord decodeControlRecord(Optional<ByteBuffer> key, Optional<ByteBuffer> value) {
        if (!key.isPresent()) {
            throw new IllegalArgumentException("Missing key in the record when a key was expected");
        } else if (key.get().remaining() == 0) {
            throw new IllegalArgumentException("Got an unexpected empty key in the record");
        }

        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");
        }

        ControlRecordType type = ControlRecordType.parse(key.get());

        final ApiMessage message;
        switch (type) {
            case LEADER_CHANGE:
                message = ControlRecordUtils.deserializeLeaderChangeMessage(value.get());
                break;
            case SNAPSHOT_HEADER:
                message = ControlRecordUtils.deserializeSnapshotHeaderRecord(value.get());
                break;
            case SNAPSHOT_FOOTER:
                message = ControlRecordUtils.deserializeSnapshotFooterRecord(value.get());
                break;
            case KRAFT_VERSION:
                message = ControlRecordUtils.deserializeKRaftVersionRecord(value.get());
                break;
            case KRAFT_VOTERS:
                message = ControlRecordUtils.deserializeVotersRecord(value.get());
                break;
            default:
                throw new IllegalArgumentException(String.format("Unknown control record type %s", type));
        }

        return new ControlRecord(type, message);
    }
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