Current state: Under discussion
Discussion thread: here
JIRA: here [Change the link from KAFKA-1 to your own ticket]
Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).
Today, all Kafka protocol requests include the client ID. This identifier can be set by the user by setting a client configuration property client.id . This is a useful capability, but it has limitations. First, the identifier is sent on each and every request. Users sometimes have use quite long client IDs, even encoding metadata into the string. The client ID is then sent on every request, in spite of the fact that the Kafka protocol is connection-oriented and it is really only necessary to send the string on the first request after connection initiation. Second, the identifier is not sufficient to identify a particular client because it is unusual for users to assign unique identifiers to their clients.
This KIP proposes introducing a UUID called the client instance ID into the request header of all Kafka protocol requests. Correlating requests from a particular client becomes much easier as a result.
This KIP proposes extending the scope of the client instance ID concept from KIP-714 into a universal unique identifier for client instances in all RPCs. The ID is introduced as a tagged field in the RPC request header, so it is present on all RPCs which use the v2 request header, which is almost all of the current versions of the RPCs. For those which do not, the client instance ID will be available in the broker's connection context.
The client instance ID is calculated by the client during the constructor of the client before it makes its initial connection to the cluster. The client instance ID generated is never the zero UUID.
In KIP-714, the client instance ID is created by the broker which responds to a client's first GetTelemetrySubscription RPC. In KIP-848, initially the member ID was created by the broker, but subsequently KIP-1082 changed this so that the client creates its own member ID. As a result, this KIP also changes the definition of the client instance ID so that the client creates its own UUID before it makes its initial connection to the cluster, and then it uses it for all future requests made to all brokers by that client instance. In addition, when the client makes its first GetTelemetrySubscription request, it will also supply the client instance ID which it created. There is no need to change the behaviour of the cluster in handling client telemetry requests, it will just be the case that the Apache Kafka Java client after this KIP no longer supplies a zero client instance ID expecting the broker to calculate the ID. However, for migration purposes, the original behavior of KIP-714 is still supported for clients which do not yet send ClientInstanceId in the request header.
Apart from its use in client telemetry, the addition of the client instance ID has no significance to the broker. It is being added to improve traceability and problem determination.
This KIP also proposes sending a null client ID for all requests on each client connection, with the exception of the initial request on each connection. This eliminates some redundant overhead of repeatedly sending the same string to the broker.
The client instance ID will be calculated during the constructor of the Producer , Consumer , ShareConsumer and Admin implementations so there is no need to have a timeout parameter on the accessor method. The following method will be added to these interfaces:
public Uuid clientInstanceId()
and then the following method will be deprecated for removal in Apache Kafka 5.0:
public Uuid clientInstanceId(Duration timeout)
In a similar vein, the following method in KafkaStreams will be added:
public ClientInstanceIds clientInstanceIds()
and then the following method will be deprecated for removal in Apache Kafka 5.0:
public ClientInstanceIds clientInstanceIds(Duration timeout)
This KIP introduces a tagged field ClientInstanceId into version 2 of the request header. This means it can be introduced without any other RPC changes.
It also proposes sending null as the ClientId for requests apart from the initial request on connections from the client to a broker. The client ID is already nullable so there's no schema change required.
{
"type": "header",
"name": "RequestHeader",
// Version 0 was removed in Apache Kafka 4.0, Version 1 is the new baseline.
//
// Version 0 of the RequestHeader is only used by v0 of ControlledShutdownRequest.
//
// Version 1 is the first version with ClientId.
//
// Version 2 is the first flexible version.
// Tag 0 introduces client instance ID (KIP-1313).
"validVersions": "1-2",
"flexibleVersions": "2+",
"fields": [
{ "name": "RequestApiKey", "type": "int16", "versions": "0+",
"about": "The API key of this request." },
{ "name": "RequestApiVersion", "type": "int16", "versions": "0+",
"about": "The API version of this request." },
{ "name": "CorrelationId", "type": "int32", "versions": "0+",
"about": "The correlation ID of this request." },
// The ClientId string must be serialized with the old-style two-byte length prefix.
// The reason is that older brokers must be able to read the request header for any
// ApiVersionsRequest, even if it is from a newer version.
// Since the client is sending the ApiVersionsRequest in order to discover what
// versions are supported, the client does not know the best version to use.
{ "name": "ClientId", "type": "string", "versions": "1+", "nullableVersions": "1+", "flexibleVersions": "none",
"about": "The client ID string." },
{ "name": "ClientInstanceId", "type": "uuid", "versions": "2+", "taggedVersions": "2+", "tag": 0, "ignorable": "true",
"about": "Unique id for this client instance." }
]
}
|
After this KIP, the Apache Kafka Java client will send the ClientInstanceId in the request header and also in the request body. If present in the header, it must match the value in the request body, and that value will not be zero.
It is still permitted to send the value zero for the ClientInstanceId in the request, which will cause the broker to create a ClientInstanceId and send it back in the response. This is the original KIP-714 behavior and it is still supported for clients which do not use the ClientInstanceId in the request header.
After this KIP, the Apache Kafka Java client will send the ClientInstanceId in the request header and also in the request body. If present in the header, it must match the value in the request body, and that value will not be zero.
The addition of a tagged field in the request headers should have no impact.
The removal of the client ID is more perhaps a more significant change. However, the client ID can easily be cached as part of the connection context in the broker. If further research reveals that the client ID removal is more problematic, the KIP can be reduced to just the addition of the client instance ID.
Unit tests will be added to ensure that the new behaviour works as expected. The existing integration and system tests should be entirely unaffected by the change, which would show that there was no behavioural impact.
It would be possible to bump the version of the request header but this is expensive. Each version of the Kafka protocol RPCs has an associated request header version, so it would be necessary to bump the versions of all the other RPCs.