Status

Current state: Under DiscussionAccepted

Discussion thread: here

JIRA: 

Release: Broker protocol - 0.10.0, Java clients - 0.10.2

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

...

1. A new ApiVersionRequest and Response type (version 0) will be added to allow clients to query the broker for supported API request types and versions.

   Code Block
   ApiVersionRequest => ApiKeys ApiKeys = [INT16]// empty   ApiVersionResponse => ApiVersions ErrorCode = INT16 ApiVersions = [ApiVersion] ApiVersion = ApiKey MinVersion MaxVersion ApiKey = INT16 MinVersion = INT16 MaxVersion = INT16

2. ApiVersionRequest .ApiKeys semantics:

3. Null array (size -1): return ApiVersions for all APIs supported by broker.

Array of some ApiKeys: return ApiVersions for the requested API keys only

requests all ApiKeys & versions supported by broker.

4. ApiVersionResponse..MinVersion-MaxVersion semantics:
   1. MinVersion and MaxVersion dictates the lowest and highest supported API versions for the API key (inclusive).
   2. All versions in the MinVersion-MaxVersion must be supported by the broker. Specific version may be deprecated through protocol documentation but must still be supported (broker must not disconnect the client and return a valid protocol response, although it is valid to return an error code if the specific API supports it).

5. Clients are recommended to use latest common supported API version.

6. ApiVersionResponse.ErrorCode is guaranteed to be the first int16 of the response for all future versions of ApiVersionRequest and is to be used to indicate that the client's ApiVersionRequest with version X (greater than 0) is not supported by the broker and the client should revert to version 0.

7. Clients are recommended to use latest version supported by the broker and itself.

8. Deprecation of a protocol version will be done by marking a protocol version Deprecation of a protocol version will be done by marking a protocol version as deprecated in protocol documentation. Documentation shall also be used to indicate a protocol version that must not be used, or for any such information. For instance, say 0.9.0 had protocol versions [0] for api key 1. On trunk, version 1 of the api key was added. Users running off trunk started using version 1 of the api and found out a major bug. To rectify that version 2 of the api is added to trunk. For some reason, it is now deemed important to have version 2 of the api in 0.the next minor release on 0.9.x, i.e., 0.9.1, as well. To do so, version 1 and version 2 both of the api will be backported to the 0.9.1 branch. 0.9.1 broker will return 0 as min supported version for the api and 2 for the max supported version for the api. However, the version 1 should be clearly marked as deprecated on its documentation. It will be client's responsibility to make sure they are not using any such deprecated version to the best knowledge of the client at the time of development (or alternatively by configuration). 
   Note that backporting of protocols has not been done in Kafka so far and in practice it can not be done until the Java client is changed so that it doesn't blindly use the latest protocol version. Otherwise, if new request versions were added to 0.9.0.2, the client would not be able to talk to a 0.9.0.1 broker.

9. Supported protocol versions obtained from a broker, is good only for current connection on which that information is obtained. In the event of disconnection, the client should obtain the information from the broker again, as the broker might have upgraded/ downgraded in the mean time.

10. Versions of ApiVersion api should never be deprecated, as clients will rely on the api to inquire supported versions.
11. Java clients will use this API to assert that current API versions are all supported by the brokers it is talking to, and raise an exception if they are not. Java clients check will test the new API.
12. The new API will be tested with librdkafka's implementation as well.

Public Interfaces

Addition of ApiVersionRequest and Response version 0.

Compatibility, Deprecation, and Migration Plan

• This is an optional interface and does not affect existing clients.
• Updated clients that add support for the new ApiVersionRequest will be able to adapt their functionality based on supported broker functionality.

Scenarios

A core Kafka developer wants to add a new version of an existing API or a new API

1. Dev adds the new version of API or new API the way they do it today. Protocol documentation, which is auto generated, will automatically be updated with this new version or API in docs. Nothing else required to be done.

1. The broker returns its full list of supported ApiKeys and versions regardless of current authentication state (e.g., before SASL authentication on an SASL listener, do note that no Kafka protocol requests may take place on a SSL listener before the SSL handshake is finished). If this is considered to leak information about the broker version a workaround is to use SSL with client authentication which is performed at an earlier stage of the connection where the ApiVersionRequest is not available.

Public Interfaces

Addition of ApiVersionRequest and Response version 0.

Compatibility, Deprecation, and Migration Plan

• This is an optional interface and does not affect existing clients.
• Updated clients that add support for the new ApiVersionRequest will be able to adapt their functionality based on supported broker functionality.

Scenarios

A core Kafka developer wants to

...

add a new version of an existing API or a new API

1. Dev adds the new version of API or new API the way they do it today. Protocol documentation, which is auto generated, will automatically be updated with this new version or API in docs. Nothing else required to be done.

A core Kafka developer wants to deprecate a version of an existing API

1. Dev modifies the api description in code to add deprecation modifies the api description in code to add deprecation information. Protocol documentation, which is auto generated, will automatically be updated with the deprecation information. Note that it will be the client developers responsibility to check deprecation information before using or supporting an api version.

...

1. Client sends metadata request and gets to know what brokers it needs to talk to for producing, consuming, etc. and also gets information on the broker's global api version. The client maintains this info.
2. Client opens a connection to other brokers that it needs to talk to and sends Metadata request with topics set as null, as it does not want topic metadata again that it has received in previous step. 
3. Client collects global api versions from all brokers it is interested in and builds a state that will look something like below.

   api versions received:

   B1 -> 4
   B2 -> 5
4. Client is expected to have a map of feature to required minimum global api versions. Something like this.
   Feature1 -> 6 // Min. global api version
   Feature2 -> 4
   Note, that this KIP is not proposing to provide this information to clients. It is clients responsibility to maintain this information.
5. Using information from 3 and 4, client can see if it can use a feature or not. For the client in example, the result will be as below.
   Feature1 can not be used as all brokers, client is interested in, do not have >= min. req. global api version.
   Feature2 can be used. 

Reason for rejection

1. = min. req. global api version.
   Feature2 can be used. 

Reason for rejection

This approach makes selective backporting of protocol versions to previous branch difficult, hard to manage. Requires a lot of manual coordination and so has lots of room for mistakes.

Sub-proposal: improved handling of unsupported requests

This proposal was not intended to solve the main feature detection functionality of KIP-35 but rather to fix the suboptimal handling of unsupported requests in the broker to provide meaningful error messages to the end user.

This is an optional nice-to-have feataure and was removed from KIP-35 to minimize scope.

Original text:

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 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 proposal is not a good fit for feature discovery since there is no way to reliably and without side-effects check if a certain protocol version is supported. As an example let's say that a future KafkaConsumer needs to commit offsets with OffsetCommit v99 (which features something spiffy), with this proposal the client would need to Join the group, pass through rebalancing, Consume and process message(s) and after that perform an OffsetCommit. If at this point it turns out OffsetCommit v99 wasn't supported the application has processed messages that it now can't commit. This late erroring handling is hard to get right.

 This approach makes selective backporting of protocol versions to previous branch difficult, hard to manage. Requires a lot of manual coordination and so has lots of room for mistakes.