private ApiMessage apiMessageValueFor(short recordType) {
        switch (recordType) {
            case 0:
            case 1:
                return new OffsetCommitValue();
            case 2:
                return new GroupMetadataValue();
            case 3:
                return new ConsumerGroupMetadataValue();
            case 4:
                return new ConsumerGroupPartitionMetadataValue();
            case 5:
                return new ConsumerGroupMemberMetadataValue();
            case 6:
                return new ConsumerGroupTargetAssignmentMetadataValue();
            case 7:
                return new ConsumerGroupTargetAssignmentMemberValue();
            case 8:
                return new ConsumerGroupCurrentMemberAssignmentValue();
            default:
                throw new CoordinatorLoader.UnknownRecordTypeException(recordType);
        }
    }
Serializer/Deserializer for {@link CoordinatorRecord}. The format is defined below:
<pre>
    record_key   = [record_type key_message]
    record_value = [value_version value_message]

    record_type     : The record type is currently define as the version of the key
                      {@link ApiMessageAndVersion} object.
    key_message     : The serialized message of the key {@link ApiMessageAndVersion} object.
    value_version   : The value version is currently define as the version of the value
                      {@link ApiMessageAndVersion} object.
    value_message   : The serialized message of the value {@link ApiMessageAndVersion} object.
</pre>
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/CoordinatorRecordSerde.java