@Override
    public CoordinatorRecord deserialize(
        ByteBuffer keyBuffer,
        ByteBuffer valueBuffer
    ) throws RuntimeException {
        final short recordType = readVersion(keyBuffer, "key");
        final ApiMessage keyMessage = apiMessageKeyFor(recordType);
        readMessage(keyMessage, keyBuffer, recordType, "key");

        if (valueBuffer == null) {
            return new CoordinatorRecord(new ApiMessageAndVersion(keyMessage, recordType), null);
        }

        final ApiMessage valueMessage = apiMessageValueFor(recordType);
        final short valueVersion = readVersion(valueBuffer, "value");
        readMessage(valueMessage, valueBuffer, valueVersion, "value");

        return new CoordinatorRecord(
            new ApiMessageAndVersion(keyMessage, recordType),
            new ApiMessageAndVersion(valueMessage, valueVersion)
        );
    }
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