...

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

Public Interfaces

Add protocol ProtocolVersionRequest & Response message

ProtocolVersionRequest => (empty)
 
ProtocolVersionResponse => [ApiKey ApiName [ApiVersion]]
  ApiKey => int16                // API protocol key
  ApiName => string              // Human readable form of API, e.g. "ProduceRequest"
  ApiVersion => int16            // Supported versions in reverse order of preferredness, most preferred version last (thus typically ascending order)
  ApiDeprecatedVersion => int16  // Deprecated versions
 
Example in pseudo format:
 
ProtocolVersionResponse =>
[
 { ApiKey: 0, ApiName: "ProduceRequest", ApiVersion: [2,1,0], ApiDeprecatedVersion: [] },
 { ApiKey: 1, ApiName: "FetchRequest", ApiVersion: [1,0], ApiDeprecatedVersion: [] },
 { ApiKey: 2, ApiName: "OffsetRequest", ApiVersion: [2], ApiDeprecatedVersion: [0, 1] },
  ...
]

Proposed Changes

1. New protocol API: ProtocolVersionRequest and ProtocolVersionResponse

This KIP proposes a new protocol request and response to be added to the Kafka protocol, 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.

If authentication is enabled, client must have permissions enabling "describe" action on "cluster" resource.

The ProtocolVersionResponse is only valid for the current connection to the broker; the client will need to reissue the ProtocolVersionRequest upon reconnect or when connecting to other brokers.

This API is optional for a client to implement.

2. Improved handling of unsupported requests on broker

Additionally, to solve issue nr 2. - ungraceful handling of unknown requests - it is also suggested to make the following change in the broker request handling code:

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.

How can a broker inform a client that it does not support the protocol version, the client is trying to use.

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

Notes on managing the protocol specification

...

 How can a client know what all versions of a protocol does a broker supports.

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 single null topic and size of -1 can be used to fetch metadata response with only broker version information and no topic/broker info.
7. On receiving a metadata request with single null topic with size of -1, 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 are not affected by 1. (ProtocolVersionRequest), but will get the error propagation benefits of 2not closing the connection for unknown protocol versions. (graceful errors).
• Updated clients that add support for 1.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.Release with 0.9.0.0: Since 0.9.0.0 will incorporate a large number of new protocol request types that the various client implementations are likely to pick up and implement it makes sense to add this new functionality along with it.

Rejected Alternatives

• Use an aggregated global ProtocolVersion
• Include 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.