Status

Current state: Under discussionVoting

Discussion thread: here

Voting 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 the 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 was meant for identifying an application, and not the instances of an application. Often, it is not sufficient to identify a particular client because it is unusual for users to assign unique identifiers to their clientsinstance.

This KIP proposes introducing adds a UUID called the client instance ID into the request header of all Kafka protocol requests. Correlating Each client has a different client instance ID, so correlating requests from a particular client becomes much easier as a result. An immediate benefit is that it can be included in request logging for troubleshooting. It is being added to improve traceability and problem determination.

Proposed Changes

This KIP proposes extending the scope of the following changes to the Kafka protocol.

Client Instance ID

This KIP proposes adding a UUID called the client instance ID concept from into the request header of all Kafka protocol requests. Those familiar with KIP-714 will be aware that it already introduced the client instance ID, so this KIP actually proposes elevating that concept 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 now calculated by the client during the constructor of the client before it makes its initial connection to the cluster. This differs slightly from how KIP-714 initialized the client instance ID. The client instance ID generated is never the zero UUIDuses the same client instance ID for its connections to every broker throughout its lifetime, even when it rebootstraps and makes new connections. The consistency and uniqueness of the identifiers are both important characteristics.

In KIP-714, the client instance ID is was created by the broker which responds responded to a client's first GetTelemetrySubscription  GetTelemetrySubscriptions RPC. In KIP-848, initially the member ID was created by the brokergroup coordinator in response to a heartbeat, but subsequently KIP-1082 changed this so that the client creates created its own member ID. As a result, this KIP also changes the definition of calculates the client instance ID on the client.

The client instance ID is added as a tagged field in the RPC request header, so it is present on all RPCs that use 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 client no longer supplies a zero client instance ID expecting the broker to calculate the ID.

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.

v2 request header, which is almost all of the current versions of the RPCs (only SaslHandshake and OffsetDelete use the v1 request header for their latest versions). The client is required to be consistent in its use of client instance ID in request headers. If it specifies the value in its request headers, every request on a connection must specify the same value. If it does not specify the value in its request headers, every request on a connection must not specified a value.

Note that omitting a client instance ID from the request header and explicitly sending a zero UUID as the client instance ID in the request header are indistinguishable in the protocol and are considered semantically equivalent. This means that if a client explicitly sets a zero UUID, the broker will treat it as if the client had not set a client instance ID. When the following text says "does not specify a client instance ID", this includes specifying a zero UUID as the client instance ID.

If a connection specifies a client instance ID in the request header of its first request which uses the v2 request header, it must specify the same client instance ID in the request header for all subsequent requests which use the v2 request header. The initial client instance ID for each connection will be cached by the broker for checking (this is an implementation detail, but caching it in the ChannelMetadataRegistry is an option). Once a client has specified a client instance ID in the request header of its first request, any subsequent requests which are missing the client instance ID (with the exception of requests using the v1 request header) or which specify a different value for the client instance ID will be rejected with error code INVALID_REQUEST.

If a client does not specify a client instance ID in the request header of its first request which uses the v2 request header, it must not specify a client instance ID in the request header of any subsequent requests. If it does so, the request will be rejected with error code INVALID_REQUEST.

Alignment of identifiers

By adding client instance ID to the request headers, we now had a unique application instance identifier which we can use for other purposes such as the member ID in the group protocols and the client telemetry client instance ID. The client MUST use the same UUID in request headers as it uses for the client telemetry client instance ID. The alignment of other identifiers is by convention (and the Java client will follow the convention) rather than mandate. In the language of standards, the client SHOULD use the same UUID in request headers as it uses for the member ID in the group protocols. This alignment just makes traceability and problem determination more straightforward.

In the modern group protocol RPCs such as ConsumerGroupHeartbeat  and ShareGroupHeartbeat , the member ID is a string. In practice, it is a UUID which is encoded into a string, but the nature of this conversion is not specified in the protocol and it really is treated as a string in the broker. The Java client uses the org.apache.kafka.common.Uuid  class to generate the member ID and convert it into a string. The client instance ID really is a UUID in the protocol. When the Java code in the broker converts this into a string, it uses the same Java code as the client does for the member ID. As a result, the member ID and client instance ID can trivially be the same when represented as strings, even though they have different data types in the protocol. For traceability and problem determination, this "conventional" alignment works well.

However, at least one other Kafka client implementation uses a slightly different string encoding of the member ID which does not match that generated by org.apache.kafka.common.Uuid . This is valid with regards to KIP-848 and KIP-932, but does unfortunately mean that the member ID does not look the same as the client instance ID, even if they have the same original UUID value in the client.

It would be possible to go around the existing RPCs such as ConsumerGroupHeartbeat  and GetTelemetrySubscriptions , and remove the fields containing the existing identifiers which are intended to be aligned. Doing so would be a bad idea though, because we would then have RPC versions which essentially depend upon the presence of a tagged field in the request header. This is a protocol-compatibility nightmare because tagged fields are optional and having a hard dependency on an optional field seems unwise.

This KIP makes one change to the GetTelemetrySubscriptions  behavior. The client may only request a new client instance ID on its initial GetTelemetrySubscriptions  request (request.clientInstanceId  is 0), if it does not also send a client instance ID in the request header. After this KIP, the broker is only expecting to generate telemetry client instance ID for older clients which do not use the request header. This will automatically align the UUID in the request headers and client telemetry.

Key:

• UUID-H - client instance ID sent by client in header, generated by client
• UUID-R - client instance ID sent by client in request, which is not equal to UUID-H
• UUID-B - client instance ID sent by broker in response, generated by broker

Pre-KIP-1313 - broker does not expect ClientInstanceId  in request header and ignores it

Post-KIP-1313 - broker is aware of optional ClientInstanceId  in request header

In summary, for GetTelemetrySubscriptions v0, here are the combinations:

Public Interfaces

Client API Changes

The client instance ID will be is calculated during the constructor of the Producer , Consumer and , ShareConsumer and Admin  implementations so there is no need to have a timeout parameter on the accessor method. The following method will be is added to these interfaces:

...

and then the following method will be is deprecated for removal in Apache Kafka 5.0:

...

In a similar vein, the following method in KafkaStreams will be is added:

public ClientInstanceIds clientInstanceIds()

and then the following method will be is deprecated for removal in Apache Kafka 5.0:

...

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.
  // Tagged Tagfield 0 introduces client instance IDClientInstanceId. (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, for identifying an application." },
    { "name": "ClientInstanceId", "type": "uuid", "versions": "2+", "taggedVersions": "2+", "tag":  0, "ignorable":  "true",
      "about": "Unique idThe client instance ID, for thisidentifying clientan instance of an application." }
  ]
}

GetTelemetrySubscriptions API

A new version 1 of this RPC is introduced which removes ClientInstanceId from the request and the response. Here's the updated request.

{
  "apiKey": 71,
  "type": "request",
  "listeners": ["broker"],
  "name": "GetTelemetrySubscriptionsRequest",
  "validVersions": "0-1",
  "flexibleVersions": "0+",
  "fields": [
    {
      "name": "ClientInstanceId", "type": "uuid", "versions": "0", <<<- not present in v1
      "about": "Unique id for this client instance, must be set to 0 on the first request."
    }
  ]
}

PushTelemetry API

A new version 1 of the RPC is introduced with removes ClientInstanceId  from the request. Here's the updated request:

...

very small behavioral change is made in the broker handling of GetTelemetrySubscriptions v0.

A post-KIP-1313 client will not send a zero ClientInstanceId  in the request body to ask the broker to assign an ID.

If the client sends a zero ClientInstanceId  in the request body, it must not send a ClientInstanceId  in the request header.

Similarly, if the client sends a ClientInstanceId  in the request header and also sends a non-zero ClientInstanceId  in the request body, the values must be the same. It is not permitted to use one client instance ID for request headers and a different client instance ID for telemetry.

As a result, the description for the ClientInstanceId  field in the request becomes "Unique id for this client instance. If client sends ClientInstanceId in the header, must equal that value. If not, must be set to 0 on the first request." .

Compatibility, Deprecation, and Migration Plan

The addition of a tagged field in the request headers header 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.

Test Plan

Unit tests will be added to ensure that the new behaviour works behavior works as expected. The existing integration and system tests should be entirely unaffected by the change, which would show that there was no behavioural behavioral impact.

Rejected Alternatives

It would be possible to add an untagged field to the request header and 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.

It was also planned to send the client ID only on the initial request on each connection, and then send a null client ID for all subsequent requests. This eliminates the unnecessary overhead of repeatedly sending the same client ID string to the broker on every request. However, doing such a change without bumping the request versions was considered to be too risky. For example, if a new client was communicating with an older broker, the fact that client ID was only present on the initial request could break client ID-based quotas. As a result, the client ID change was removed from this KIP, and could potentially be introduced the future alongside version bumps of the RPCs.