@Override
    public void write(ApiMessageAndVersion data,
                      ObjectSerializationCache serializationCache,
                      Writable out) {
        out.writeUnsignedVarint(DEFAULT_FRAME_VERSION);
        out.writeUnsignedVarint(data.message().apiKey());
        out.writeUnsignedVarint(data.version());
        data.message().write(out, serializationCache, data.version());
    }
This is an implementation of {@code RecordSerde} with {@link ApiMessageAndVersion} but implementors need to implement
{@link #apiMessageFor(short)} to return a {@code ApiMessage} instance for the given {@code apiKey}.
<br>
This can be used as the underlying serialization mechanism for records defined with {@link ApiMessage}s.
<br><br>
Serialization format for the given {@code ApiMessageAndVersion} is below:
<pre>
    [data_frame_version header message]
    header =&gt; [api_key version]

    data_frame_version   : This is the header version, current value is 0. Header includes both api_key and version.
    api_key              : apiKey of {@code ApiMessageAndVersion} object.
    version              : version of {@code ApiMessageAndVersion} object.
    message              : serialized message of {@code ApiMessageAndVersion} object.
</pre>
server-common/src/main/java/org/apache/kafka/server/common/serialization/AbstractApiMessageSerde.java