Status

Current state: Under Discussion

Discussion thread: here

JIRA:  Unable to render Jira issues macro, execution error.

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

This KIP aims to solve two problems:

1. There is currently no way for a Kafka client to know which API version the broker supports. This means a client might not be able to perform its desired functionality, nor report any meaningful errors back to the application. Which leads in to the next problem, problem #2.

2. Unsupported requests are not handled gracefully by the broker. A broker receiving a protocol request it does not support will simply close the TCP connection to the client. From the client's point of view it has no way of knowing that the broker did not support the request, it could just as well be a network or software problem causing the connection reset.

This makes it hard for clients and applications to support multiple versions of Kafka, which in turn limits the Kafka eco-system since applications and clients will need to be manually built or configured for a specific broker version.

Additional key points

• The broker version from MetadataResponse may be used in broker-broker communication to decide on a common protocol version between brokers with different versions, i.e., during upgrades.

Proposed Changes

In order for a client to successfully talk to a broker, it needs to know what version of protocol does the broker support. This in turn leads to following requirements.

Informing client of unknown protocol request or unsupported version

As mentioned before, broker on receiving a protocol request it does not support will simply close the TCP connection to the client. From the client's point of view it has no way of knowing that the broker did not support the request, it could just as well be a network or software problem causing the connection reset.

The KIP proposes that if a request is received for an unknown protocol request type, or for an unsupported version of a known type, the broker should respond with an empty response only including the common Length+CorrId header with Length=0, instead of closing the connection. The effect of this empty message received by the client is that the client will fail to parse the response (since it will expect a proper non-empty reply) and throw an error to the application with the correct context - that of a protocol parsing failure for request type XYZ. All existing clients should be able to handle this gracefully without any alterations to the code, while updated clients supporting the proposals in this KIP can handle the empty response accordingly by returning a "Request not supported by server" error to the application. The level of detail in the returned error message to the application naturally varies between client implementations but is still by far better than a closed connection which says nothing of the underlying error.

This is a nice to have part of this KIP and not a blocker.

Informing client of versions supported by a broker

The information can be provided in various ways. However after considering many alternatives, mentioned in rejected alternatives, the KIP suggests the following.

Maintain protocol versions per broker version

Protocol versions supported by a broker version will be documented, which will be versioned with broker version. Every time there is a version change in any of the protocols, documentation version needs to be updated and linked to main documentation page. Deprecation of a protocol version will be done via marking the version as deprecated on the protocol documentation.

Communicate broker version to clients

A client can deduce protocol versions supported by a broker by knowing its broker version. Note that broker version is only valid for the current connection to the broker, as the broker might upgrade/ downgrade to a different version during a re-connect. This leads to an interesting question, how can a broker convey broker version to its clients. There are a few options to achieve this.

1. Use (versionInt, versionString), e.g., (0x00100000, "0.10.0-IV0"). Client will have to parse versionString to get internal version, if they want to. Comparing versions will be tricky, due to internal versions. This requires adding and managing another entity in Kafka codebase that keeps track of broker version, apart from ApiVersion.

2. Use ApiVersion info which has one to one mapping with broker version, (versionId, versionString), e.g., (4, "0.10.0-IV0"). versionId is a monotonically increasing number that is bumped every time version for any protocol is bumped. This can re-use existing ApiVersion.
3. Use (versionInt, versionId, versionString), e.g., (0x00100000, 4, "0.10.0-IV0"). Combines option 1 and 2.

Based on above options, we are looking at 3 things to identify a broker version, i.e., versionInt, versionId and versionString. Whether we use versionInt or versionId, client needs to keep track of what each versionInt/versionId supports. The major difference between versionInt and versionId is that versionInt do not care about internal versions, while versionId does. Since internal version is maintained as of now, going with Option 2 or 3 makes sense. Among option 2 and 3, option 2 has the minimal information required, the KIP suggests to go with option 2. 

Summary of the changes proposed as part of this KIP

1. Every time a protocol version changes, for any request/response, ApiVersion will be bumped up.
2. Protocol documentation will be versioned with ApiVersion. Every time there is a ApiVersion change, protocol documentation version needs to be updated and linked to main documentation page.
3. Deprecation of protocol version will be done via marking the version as deprecated on the protocol documentation.
4. On getting unknown protocol version, broker will send an empty response, instead of simply closing client connection.
5. Metadata response will be enhanced to also contain broker version, i.e., (versionId, versionString), of the responding broker.
6. Metadata request with a null array can be used to fetch metadata response with only broker version information and no topic/broker info. Note that this will require changing MetadataRequest from ArrayOf[STRING] to NullableArrayOf[STRING].
7. On receiving a metadata request with null array, broker will respond without topics metadata, as this can lead to unnecessary penalty on large clusters.

Public Interfaces

Changes to MetadataRequest and MetadataResponse

MetadataRequest and response version will be bumped. On receiving a MetadataRequest with new version, broker will respond with new MetadataResponse containing broker version.

MetadataResponse => CorrelationID, BrokerVersion, BrokerEndPoints, TopicsMetaData // BrokerVersion is added
BrokerVersion => VersionId, VersionString // Version id representing Broker version, string representation of broker version

Compatibility, Deprecation, and Migration Plan

• Existing clients will get the error propagation benefits of not closing the connection for unknown protocol versions. (graceful errors).
• Updated clients that add support for new MetadataRequest will be able to adapt their functionality based on broker functionality and provide meaningful error propagation to applications when proper broker support is lacking.

Rejected Alternatives

• Include meta key-value tags in response ("broker.id=...", "message.max.bytes=...", ...)
• Include user-defined tags in response ("aws.zone=...", "rack=...", ..)

• Use a new protocol request and response, ProtocolVersionRequest and ProtocolVersionResponse. A client may use this new API to query the broker for its supported protocol request types and versions and use the information to select a suitable set of conforming API requests and versions for further communication with the broker.