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 versions of various protocols does the broker support. The KIP suggests the following to achieve that.

1. Use existing ApiVersion as a unique identifier of protocols' versions supported by a broker. ApiVersion has a monotonically increasing id, which is increased by 1 every time a new version of a protocol is added.

   sealed trait ApiVersion extends Ordered[ApiVersion] {
     val version: String // Version string, also has information on internal version. E.g., "0.9.0.X", "0.10.0-IV0"
     val id: Int // Monotonically increasing id. E.g., "0.10.0-IV0" has id 4
   }

2. Maintain protocol documentation per ApiVersion. Each time a new version of any protocol is added, a new ApiVersion, with id as last ApiVersion's id +1, is created and a protocol documentation for the new ApiVersion is created with documentation on newly added version of protocol. The new protocol documentation for the latest ApiVersion will be then added to main protocol documentation page. This procedure on adding protocol documentation for each new ApiVersion will be documented.

3. Document requirements on deprecating a version of a protocol. Deprecation of a protocol will be done by marking a protocol version as deprecated in protocol documents corresponding to the ApiVersions in which the protocol version is deprecated. This procedure on deprecating a protocol version will be documented.
4. On receiving a request for an unknown protocol request type, or for an unsupported version of a known type, broker will 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.

5. New Metadata Request and Response version, v1, will be added to provide ApiVersion information to clients. ApiVersion will be composed of ProtocolVersion, an int, and BuildDescription, a String. BuildDescription should only be used where readability is a concern, like, logs, error-messages, tool outputs, etc.

   MetadataRequest => Topics
   	Topics => NullableArrayOf(STRING)
    
   MetadataResponse => ApiVersion, BrokerEndPoints, TopicsMetaData // ApiVersion is added
   	ApiVersion => ProtocolVersion, BuildDescription // E.g., 4, "0.10.0-IV0"
   		ProtocolVersion => INT16 // E.g., 4
   		BuildDescription => STRING // "0.10.0-IV0", this should only be used for error messages, etc.
   	BrokerEndPoints => Same as existing
   	TopicsMetaData => Same as existing

6. A client willing to and capable of handling multiple protocol versions should send a Metadata Request v1 or above with topics as null for each new connection to a broker. A broker supporting Metadata Request v1 and above will respond with MetadataResponse that will contain ApiVersion, i.e., ProtocolVersion and BuildDescription, and will NOT contain any TopicsMetadata. Note that broker only provides information on its ProtocolVersion and BuildDescription, while responding to a MetadataRequest.
7. Client should get broker's ApiVersion from the broker's MetadataResponse and use the info to determine protocol version it should use to communicate with the broker. Note that ApiVersion obtained for a broker from the broker's MetadataResponse is good only for the current connection. In the event of disconnection, the client should obtain the ApiVersion information from the broker again, as the broker might have upgraded/ downgraded in the mean time.

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 ApiVersion associated with the broker. This information can be used by client to determine which protocol versions are supported by the broker.

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=...", ..)
• Using a versionInt instead of ApiVersion's id as broker protocol version.

• 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.