private void readMessage(ApiMessage message, ByteBuffer buffer, short version, String name) throws RuntimeException {
        try {
            message.read(new ByteBufferAccessor(buffer), version);
        } catch (RuntimeException ex) {
            throw new RuntimeException(String.format("Could not read record with version %d from %s's buffer due to: %s.",
                version, name, ex.getMessage()), ex);
        }
    }
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