...

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 client no longer supplies an empty a zero client instance ID expecting the broker to calculate the ID.

...

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.

Public Interfaces

Client API Changes

The client instance ID will be calculated during the constructor of the Producer , Consumer 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)

Kafka Protocol Changes

Request Header

...

{
  "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." }
  ]
}

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:

{
  "apiKey": 72,
  "type": "request",
  "listeners": ["broker"],
  "name": "PushTelemetryRequest",
  "validVersions": "0-1",
  "flexibleVersions": "0+",
  "fields": [
    {
      "name": "ClientInstanceId", "type": "uuid", "versions": "0", <<<- not present in v1
      "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."
    }
  ]
}

Compatibility, Deprecation, and Migration Plan

...