[m1a2st]

RequestHeader.parse() derives the header version from the request API
version and parses the entire header using that schema. For an API
version unknown to the broker, it falls back to the latest supported
header version. If a future version adds a non-tagged field, parsing
fails with InvalidRequestException and closes the connection, so a
newer client sending ApiVersionsRequest never receives the v0
UNSUPPORTED_VERSION response needed to downgrade.

For unsupported API versions, RequestHeader.parse() now reads only
fields with a fixed layout across header versions (API key, API
version, correlation ID, and client ID), leaving the rest unparsed.
Supported versions continue to be parsed as before.

Reviewers: Jun Rao junrao@gmail.com, Andrew Schofield
aschofield@confluent.io, Chia-Ping Tsai chia7712@gmail.com

[junrao]

@m1a2st : Thanks for the PR. Left a few comments.

[m1a2st]

Thanks for @junrao comments, I have addressed all.

[junrao]

@m1a2st : Thanks for the updated PR. A few more comments.

[m1a2st]

Thanks @junrao for the review. I've addressed all the comments.

[junrao]

@m1a2st : Thanks for the updated PR. A few more minor comments.

[m1a2st]

Thanks @junrao for the review. I've addressed all the comments.

[junrao]

@m1a2st : Thanks for the updated PR. Just one more minor comment.

@AndrewJSchofield : Does this change look good to you too?

[AndrewJSchofield]

I'll take a look in the next few days.

[AndrewJSchofield]

Thanks for the PR. This looks like a useful hardening of the parsing logic. I do think it will be a very, very long time before releases before 4.5.0 are not supported though.

[AndrewJSchofield]

Isn't it theoretically possible that apiKey().isVersionSupported(apiVersion()) is false in which case the response header version is not really known? I suppose the response header version should be 1 in this case, and we must never add non-tagged fields to the response header.

[m1a2st]

In practice, only ApiVersionsRequest can reach toResponseHeader() with an unsupported version. For other APIs, Processor#parseRequestHeader rejects unsupported versions with UnsupportedVersionException, and RequestContext#parseRequest also throws in this case.

For ApiVersions, responseHeaderVersion() is hard-coded to return 0 for all versions (KIP-511), so the response header version is always well-defined.

[junrao]

@AndrewJSchofield : Good point. Claude suggested the following way to tighten this up.

RequestContext is the only production caller of toResponseHeader() (:159, :173). Compute both the body version and the header version from the same supported version:

public Send buildResponseSend(AbstractResponse body) {
    short version = apiVersion();
    return body.toSend(responseHeader(version), version);
}
public ByteBuffer buildResponseEnvelopePayload(AbstractResponse body) {
    short version = apiVersion();
    return body.serializeWithHeader(responseHeader(version), version);
}
/**
 * The response header follows the version the response is written at, which for an unsupported
 * ApiVersionsRequest is v0 rather than the version of the request.
 */
private ResponseHeader responseHeader(short responseVersion) {
    return new ResponseHeader(header.correlationId(), header.apiKey().responseHeaderVersion(responseVersion));
}

As a related issue, currently, KafkaApis uses 4 different ways to obtain the api version. They all agree except for unsupport ApiVersion.

Accessor | Returns | Why
-- | -- | --
request.header.apiVersion | N (6) | data.requestApiVersion(), the version in the request header
request.context.apiVersion() | 0 | isUnsupportedApiVersionsRequest() is true, so it returns 0, the version the response is written at (RequestContext.java:180-185)
request.context.requestVersion() | N (6) | header.apiVersion(), returned as an int (RequestContext.java:217-219)
request.body[ApiVersionsRequest].version | 0 | parseRequest builds new ApiVersionsRequest(new ApiVersionsRequestData(), (short) 0, header.apiVersion()) (RequestContext.java:131), so the body is a v0 request and its body isn't parsed

It would be useful to document what each api means and when to use which.

We probably can change request.context.apiVersion() to sth like request.context.effectiveApiVersion() and document it that this is the version parsed by the server. KafkaApis should typically use this or request.body[ApiVersionsRequest].version to get api version.

Document that request.context.requestVersion() is the way for getting the original Api Version sent by the client. This should only be used in the following in KafkaApis where the original version is needed.

if (!apiVersionManager.isApiEnabled(request.header.apiKey, request.header.apiVersion))

[m1a2st]

My understanding is that the original client version and the effective version serve different purposes. For an unsupported ApiVersionsRequest, the original version is N, while the effective version is 0. The current code still works because ApiVersions always uses response header v0, but deriving both formats from the effective version makes the response serialization more consistent.

Based on this, I’ve updated both response serialization paths to use the effective version. KafkaApis now uses the effective version, for handler logic, while retaining the original version for the initial API-enabled check.

Please let me know if I missed anything. Thanks!

[chia7712]

Should we hardcode the parsing of the four common fields instead of relying on the generated code? What we need is “read only the first four fields”, and header v1 only happens to match that today.

[m1a2st]

I'd prefer to keep the parsing in the generated code. Hand-written parsing would duplicate the decoding logic outside the schema.

I agree that v1 only happens to match the four common fields today, so I've made that explicit, RequestHeader.json now specifies that the first four fields must retain their order and encoding, and that v1 must not be removed because brokers use it to parse headers with unsupported API versions.

[chia7712]

I do think it will be a very, very long time before releases before 4.5.0 are not supported though.

I hope we are still here when that day comes :)

[junrao]

@m1a2st : Thanks for the updated PR. Just a minor comment.

[junrao]

@m1a2st : Thanks for updated PR. Left one reply to a previous comment.

[junrao]

@m1a2st : Thanks for the updated PR. A few more comments.

[junrao]

AbstractRequest is a generic class and not used just by request handlers. version() should be self explanatory, i.e., it's the version associated with this request object. We can just remove the comment here.

[junrao]

This is a generic method and is self explanatory. We can remove the comment.

[junrao]

How about the following?

Returns the API version parsed by the server. This is usually the API version in the original client request. The only exception is an unsupported {@link ApiVersionsRequest}, whose API version is set to 0 after parsing. The response should be sent and serialized using this version.

To retrieve the API version in the original client request, use #requestVersion() instead.

[junrao]

We can remove this line.

[junrao]

The API should be self explanatory. The server side details are explained in RequestContext already. We can remove this comment.

We can probably document RequestContext.header that this is the original header from the client request. For unsupported ApiVerionRequest, use effectiveApiVersion() to get the actual parsed API version.

[junrao]

Probably use offsetFetchRequest.version here?

[junrao]

It seems that all usage of requestVersion() in GroupCoordinatorService, GroupMetadataManager and OffsetMetadataManager should be using effectiveApiVersion.

[m1a2st]

The coordinator components receive AuthorizableRequestContext, which only exposes requestVersion() and defines it as the original version from the request header.

I think the possible approaches are:

• Add effectiveApiVersion() to AuthorizableRequestContext, which would extend the public authorizer API.
• Introduce an internal interface exposing the effective version, which would require updating the coordinator signatures.
• Use an instanceof check with a fallback to requestVersion(), although other implementations would still return the original version.

Would you prefer keeping these calls as-is for this PR and addressing the coordinator context separately?

[chia7712]

        CreateDelegationTokenResponse.prepareResponse(request.context.requestVersion, requestThrottleMs,
        CreateDelegationTokenResponse.prepareResponse(request.context.requestVersion, requestThrottleMs,
               request.context.requestVersion,

I think those parts related to preparing the response should use effectiveApiVersion instead.

[chia7712]

Should we consolidate the fallback logic into ApiKeys? For example:

public enum ApiKeys {
    ...
    /**
     * ApiVersions is used for version discovery, so the server accepts any version of it.
     * An unsupported version is handled as this version, and the client falls back to parsing
     * the response with it (KIP-511). This version must never be removed.
     */
    public static final short API_VERSIONS_FALLBACK_VERSION = 0;

    /**
     * Returns true if a request with the given version must be handled as
     * {@link #API_VERSIONS_FALLBACK_VERSION} instead of being parsed.
     */
    public boolean isFallbackVersion(short version) {
        return this == API_VERSIONS && !isVersionSupported(version);
    }

    public boolean isVersionEnabled(short apiVersion, boolean enableUnstableLastVersion) {
        // ApiVersions accepts any version, see isFallbackVersion
        if (this == API_VERSIONS) return true;
        ...
    }
}