Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/common/record/internal/ControlRecordType.java (Java)

Code before:
    }

    public short type() {
        return type;
    }

    public ByteBuffer recordKey() {
        if (this == UNKNOWN)
            throw new IllegalArgumentException("Cannot serialize UNKNOWN control record type");
        return buffer.duplicate();
    }

    public int controlRecordKeySize() {
        return buffer.remaining();
    }

    public static short parseTypeId(ByteBuffer key) {
        // We should duplicate the original buffer since it will be read again in some cases, for example,
        // read by KafkaRaftClient and RaftClient.Listener
        ByteBuffer buffer = key.duplicate();
        short version = buffer.getShort();
        if (version < ControlRecordTypeSchema.LOWEST_SUPPORTED_VERSION)
            throw new InvalidRecordException("Invalid version found for control record: " + version +
                    ". May indicate data corruption");

        if (version > ControlRecordTypeSchema.HIGHEST_SUPPORTED_VERSION) {
            log.debug("Received unknown control record key version {}. Parsing as version {}", version,
                    ControlRecordTypeSchema.HIGHEST_SUPPORTED_VERSION);
            version = ControlRecordTypeSchema.HIGHEST_SUPPORTED_VERSION;
        }
        ControlRecordTypeSchema schema = new ControlRecordTypeSchema(new ByteBufferAccessor(buffer), version);
        return schema.type();
    }

    public static ControlRecordType fromTypeId(short typeId) {
        switch (typeId) {
            case 0:
                return ABORT;
            case 1:
                return COMMIT;
            case 2:
                return LEADER_CHANGE;
            case 3:
                return SNAPSHOT_HEADER;
            case 4:
                return SNAPSHOT_FOOTER;
            case 5:
                return KRAFT_VERSION;
            case 6:
                return KRAFT_VOTERS;


Review comment (bug):
Could you consider wrapping the error from `BufferUnderflowException` to `InvalidRecordException` for compatibility?

Code after:
        buffer = MessageUtil.toVersionPrefixedByteBuffer(ControlRecordTypeSchema.HIGHEST_SUPPORTED_VERSION, schema);
    }

    public short type() {
        return type;
    }

    public ByteBuffer recordKey() {
        if (this == UNKNOWN)
            throw new IllegalArgumentException("Cannot serialize UNKNOWN control record type");
        return buffer.duplicate();
    }

    public int controlRecordKeySize() {
        return buffer.remaining();
    }

    public static short parseTypeId(ByteBuffer key) {
        // We should duplicate the original buffer since it will be read again in some cases, for example,
        // read by KafkaRaftClient and RaftClient.Listener
        ByteBuffer buffer = key.duplicate();
        if (buffer.remaining() < CONTROL_RECORD_KEY_SIZE)
            throw new InvalidRecordException("Invalid value size found for control record key. " +
                    "Must have at least " + CONTROL_RECORD_KEY_SIZE + " bytes, but found only " + buffer.remaining());

        short version = buffer.getShort();
        if (version < ControlRecordTypeSchema.LOWEST_SUPPORTED_VERSION)
            throw new InvalidRecordException("Invalid version found for control record: " + version +
                    ". May indicate data corruption");

        if (version > ControlRecordTypeSchema.HIGHEST_SUPPORTED_VERSION) {
            log.debug("Received unknown control record key version {}. Parsing as version {}", version,
                    ControlRecordTypeSchema.HIGHEST_SUPPORTED_VERSION);
            version = ControlRecordTypeSchema.HIGHEST_SUPPORTED_VERSION;
        }
        ControlRecordTypeSchema schema = new ControlRecordTypeSchema(new ByteBufferAccessor(buffer), version);
        return schema.type();
    }

    public static ControlRecordType fromTypeId(short typeId) {
        switch (typeId) {
            case 0:
                return ABORT;
            case 1:
                return COMMIT;
            case 2:
                return LEADER_CHANGE;
            case 3:
                return SNAPSHOT_HEADER;
            case 4:
                return SNAPSHOT_FOOTER;