@Override
    public ApiMessageAndVersion read(Readable input,
                                     int size) {
        short frameVersion = unsignedIntToShort(input, "frame version");

        if (frameVersion == 0) {
            throw new MetadataParseException("Could not deserialize metadata record with frame version 0. " +
                "Note that upgrades from the preview release of KRaft in 2.8 to newer versions are not supported.");
        } else if (frameVersion != DEFAULT_FRAME_VERSION) {
            throw new MetadataParseException("Could not deserialize metadata record due to unknown frame version "
                    + frameVersion + "(only frame version " + DEFAULT_FRAME_VERSION + " is supported)");
        }
        short apiKey = unsignedIntToShort(input, "type");
        short version = unsignedIntToShort(input, "version");

        ApiMessage record;
        try {
            record = apiMessageFor(apiKey);
        } catch (Exception e) {
            throw new MetadataParseException(e);
        }
        try {
            record.read(input, version);
        } catch (Exception e) {
            throw new MetadataParseException("Failed to deserialize record with type " + apiKey, e);
        }
        if (input.remaining() > 0) {
            throw new MetadataParseException("Found " + input.remaining() +
                    " byte(s) of garbage after " + apiKey);
        }
        return new ApiMessageAndVersion(record, 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