...

Today, all Kafka protocol requests include the client ID. This identifier can be set by the user setting 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 such as the availability zone 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. It's conceivable that a client could mutate the client ID between requests but it is expected to be static for the duration of a connection, so sending it repeatedly is wasteful. Second, the client ID is not sufficient to identify a particular client because it is unusual for users to assign unique identifiers to their clients.The client ID was meant for identifying an application, and not the instances of an application. Often, it is not sufficient to identify a particular client instance.

This KIP adds a UUID called the client instance ID into the request header of all Kafka protocol requests. Each client has a different client instance ID, so correlating requests from a particular client becomes much easier. 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 two changes to the Kafka protocol.

Client Instance ID

This KIP proposes adding This KIP adds a UUID called the client instance ID into the request header of all Kafka protocol requests. Each client has a different Those familiar with KIP-714 will be aware that it already introduced the client instance ID, so correlating requests from a particular client becomes much easier.

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.

Proposed Changes

This KIP proposes three changes to the Kafka protocol.

Client Instance ID

this KIP actually proposes extending its scope to a universal unique identifier for client instances in all RPCs.

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, but the change is compatible with the existing behavior of the broker. The client uses 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 characteristicsThis KIP proposes adding a UUID called the client instance ID 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 extending its scope to a universal unique identifier for client instances in all RPCs.

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, but the change is compatible with the existing behavior of the broker. The client uses 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.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 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 added as a tagged field in the RPC request header, so it is present on all RPCs that use the 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.

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 the ChannelMetadataRegistry is an option).  Once 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.

...

In KIP-714, the client instance ID was created by the broker which responded to a client's first GetTelemetrySubscriptions RPC. In KIP-848, the member ID was created by the group coordinator in response to a heartbeat, but subsequently KIP-1082 changed this so that the client created its own member ID. As a result, this KIP also changes the definition of the client instance ID so that the client creates it. After this KIP, when the client makes its first GetTelemetrySubscriptions request, it supplies the client instance ID which it has already created rather than specifying a zero client instance ID; the Apache Kafka Java client no longer supplies a zero client instance ID on GetTelemetrySubscriptions requests. The original behavior of KIP-714 is still supported for clients which do not yet send ClientInstanceId in the request header; they just specify a zero client instance ID in their first GetTelemetrySubscriptions request (v0) and then receive a client instance ID to use for client telemetry calculated by the broker in its response.

...

Consumer groups with the consumer group protocol (KIP-848 and KIP-1082), share groups (KIP-932) and streams groups (KIP-1071) all make use of a client-generated UUID as the member ID. This KIP proposes using the same client-generated UUID as the client instance ID and the member ID. This makes no change to the protocol at all. It simply creates a single UUID when a client is constructed, and when a client-generated member ID is required for the group protocol, it uses the client instance ID.

Public Interfaces

Client

...

This KIP also proposes a second change to the Kafka protocol. It proposes sending the client ID only on the initial request on each connection, and then sending 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. The initial client ID for each connection will be cached by the broker (this is an implementation detail, but caching it in the ChannelMetadataRegistry is an option). After this KIP, the broker will assume that the client ID from the initial request applies to all subsequent requests on a connection, and it will ignore the client ID specified on any subsequent requests.

Public Interfaces

Client API Changes

API Changes

The client instance ID is 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 is added to these interfaces:

public Uuid clientInstanceId()

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

public Uuid clientInstanceId(Duration timeout)

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

public ClientInstanceIds clientInstanceIds()

and then the following method

The client instance ID is 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 is added to these interfaces:

public Uuid clientInstanceId()

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

public Uuid clientInstanceId(Duration timeout)

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

public ClientInstanceIds clientInstanceIds()

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

public ClientInstanceIds 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.
  // Tagged field 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": "The uniqueclient instance ID, for thisidentifying clientan instance of an application." }
  ]
}

GetTelemetrySubscriptions API

This KIP introduces GetTelemetrySubscriptions v1.

Request

If the client uses v0, the original KIP-714 behavior is used. The client sends the value 0 for the ClientInstanceId in order to 

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 request header, it must match the value in the request body, and that value will not be zero, or else the request will be rejected with error code INVALID_REQUEST.

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.

PushTelemetry API

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 request header, it must match the value in the request body, and that value will not be zero, or else the request will be rejected with error code INVALID_REQUEST.

Compatibility, Deprecation, and Migration Plan

The addition of a tagged field in the request header should have no impact.

The broker will assume that the client ID from the initial request on a connection is used on all subsequent requests. There is no need to send a client ID on any requests except the initial request on a connection, and the client ID specified on requests after the initial request on a connection will be ignored. The client ID is expected to be static and not change from request to request, so this change should have no effect.

Test Plan

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

Rejected Alternatives

supported for clients which do not use the ClientInstanceId  in the request header.

{
  "apiKey": 71,
  "type": "request",
  "listeners": ["broker"],
  "name": "GetTelemetrySubscriptionsRequest",
  // Version 0 is the initial version. (KIP-714)
  //
  // Version 1 removes ClientInstanceId from the request and response. (KIP-1313)
  "validVersions": "0-1",
  "flexibleVersions": "0+",
  "fields": [
    {
      "name": "ClientInstanceId", "type": "uuid", "versions": "0", << Only v0 supports this
      "about": "Unique id for this client instance, must be set to 0 on the first request (v0 only)."
    }
  ]
}

Response

The ClientInstanceId  is removed in the v1 schema.

{
  "apiKey": 71,
  "type": "response",
  "name": "GetTelemetrySubscriptionsResponse",
  // Version 0 is the initial version. (KIP-714)
  //
  // Version 1 removes ClientInstanceId from the request and response. (KIP-1313)
  "validVersions": "0-1",
  "flexibleVersions": "0+",
  "fields": [
    {
      "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota."
    },
    {
      "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error."
    },
    {
      "name": "ClientInstanceId", "type": "uuid", "versions": "0", << Only v0 supports this
      "about": "Assigned client instance id if ClientInstanceId was 0 in the request, else 0 (v0 only)."
    },
    {
      "name": "SubscriptionId", "type": "int32", "versions": "0+",
      "about": "Unique identifier for the current subscription set for this client instance."
    },
    {
      "name": "AcceptedCompressionTypes", "type": "[]int8", "versions": "0+",
      "about": "Compression types that broker accepts for the PushTelemetryRequest."
    },
    {
      "name": "PushIntervalMs", "type": "int32", "versions": "0+",
      "about": "Configured push interval, which is the lowest configured interval in the current subscription set."
    },
    {
      "name": "TelemetryMaxBytes", "type": "int32", "versions": "0+",
      "about": "The maximum bytes of binary data the broker accepts in PushTelemetryRequest."
    },
    {
      "name": "DeltaTemporality", "type": "bool", "versions": "0+",
      "about": "Flag to indicate monotonic/counter metrics are to be emitted as deltas or cumulative values."
    },
    {
      "name": "RequestedMetrics", "type": "[]string", "versions": "0+",
      "about": "Requested metrics prefix string match. Empty array: No metrics subscribed, Array[0] empty string: All metrics subscribed."
    }
  ]
}

PushTelemetry API

This KIP introduces PushTelemetry v1.

Request

v1 removes ClientInstanceId  which is now sent by the client in the request header. The ClientInstanceId  in the request header must not be zero, or else the request will be rejected with error code INVALID_REQUEST .

{
  "apiKey": 72,
  "type": "request",
  "listeners": ["broker"],
  "name": "PushTelemetryRequest",
  // Version 0 is the initial version. (KIP-714)
  //
  // Version 1 moves ClientInstanceId from the request to the request header. (KIP-1313)
  "validVersions": "0-1",
  "flexibleVersions": "0+",
  "fields": [
    {
      "name": "ClientInstanceId", "type": "uuid", "versions": "0", << Only v0 supports this.
      "about": "Unique id for this client instance."
    },
    {
      "name": "SubscriptionId", "type": "int32", "versions": "0+",
      "about": "Unique identifier for the current subscription."
    },
    {
      "name": "Terminating", "type": "bool", "versions": "0+",
      "about": "Client is terminating the connection."
    },
    {
      "name": "CompressionType", "type": "int8", "versions": "0+",
      "about": "Compression codec used to compress the metrics."
    },
    {
      "name": "Metrics", "type": "bytes", "versions": "0+", "zeroCopy": true,
      "about": "Metrics encoded in OpenTelemetry MetricsData v1 protobuf format."
    }
  ]
}

Response

No change to the schema with v1.

{
  "apiKey": 72,
  "type": "response",
  "name": "PushTelemetryResponse",
  // Version 0 is the initial version. (KIP-714)
  //
  // Version 1 moves ClientInstanceId from the request to the request header. (KIP-1313)
  "validVersions": "0-1",
  "flexibleVersions": "0+",
  "fields": [
    {
      "name": "ThrottleTimeMs", "type": "int32", "versions": "0+",
      "about": "The duration in milliseconds for which the request was throttled due to a quota violation, or zero if the request did not violate any quota."
    },
    {
      "name": "ErrorCode", "type": "int16", "versions": "0+",
      "about": "The error code, or 0 if there was no error."
    }
  ]
}

Compatibility, Deprecation, and Migration Plan

The addition of a tagged field in the request header should have no impact.

Test Plan

Unit tests will be added to ensure that the new behavior works as expected. The existing integration and system tests should be entirely unaffected by the change, which would show that there was no 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 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.