...

Proposed Changes

This KIP proposes two the following changes to the Kafka protocol.

...

This 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 that concept 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

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 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 v2 request header, which is almost all of the current versions of the RPCs (only SaslHandshake 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 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 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.In KIP-714, the

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 alignment of these 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 and the client telemetry client instance ID. This alignment just makes traceability and problem determination more straightforward.

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.

This KIP makes one change to the GetTelemetrySubscriptions  behavior. If the client requests a new client instance ID on its initial GetTelemetrySubscriptions  request and it sends a client instance ID in the request header, the broker will send back that client instance ID rather than generating a new UUID. 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

Public Interfaces

Client 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 is deprecated for removal in Apache Kafka 5.0:

public ClientInstanceIds clientInstanceIds(Duration timeout)

Kafka Protocol Changes

Request Header

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.

{
  "type": "header"

Member ID in group protocols

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 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 is deprecated for removal in Apache Kafka 5.0:

public ClientInstanceIds clientInstanceIds(Duration timeout)

Kafka Protocol Changes

Request Header

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.

{
  "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 ClientInstanceId. (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, for identifying an application." },
    { "name": "ClientInstanceId", "type": "uuid", "versions": "2+", "taggedVersions": "2+", "tag": 0, "ignorable": "true",
      "about": "The client instance ID, for identifying an instance of an application." }
  ]
}

GetTelemetrySubscriptions API

This KIP introduces GetTelemetrySubscriptions v1.

Request

If the client uses v0, the original KIP-714 behavior is used. It is still permitted to send the value zero for the ClientInstanceId  in the v0 request, which will cause the broker to assign 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 generate its own client instance ID and send it in the ClientInstanceId  in the request header. The value must not be zero, or else the request will be rejected with error code INVALID_REQUEST.

{
  "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": "PushTelemetryRequestRequestHeader",
  // Version 0 was removed in Apache Kafka 4.0, Version 1 is the initialnew version. (KIP-714)baseline.
  //
  // Version 10 moves ClientInstanceIdof 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."
    },
    {
      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 ClientInstanceId. (KIP-1313)
  "validVersions": "1-2",
  "flexibleVersions": "2+",
  "fields": [
    { "name": "SubscriptionIdRequestApiKey", "type": "int32int16", "versions": "0+",
      "about": "UniqueThe identifierAPI forkey theof currentthis subscriptionrequest."
    },
    {
      "name": "TerminatingRequestApiVersion", "type": "boolint16", "versions": "0+",
      "about": "ClientThe API isversion terminatingof thethis connectionrequest."
    },
    {
      "name": "CompressionTypeCorrelationId", "type": "int8int32", "versions": "0+",
      "about": "CompressionThe codeccorrelation usedID toof compressthis the metricsrequest."
    },

    {
// The ClientId string must  "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+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 duration in millisecondsclient ID, for whichidentifying the request was throttled due to a quota violation, or zero if the request did not violate any quota."
   an application." },
    {
 "name": "ClientInstanceId",   "type": "uuid", "nameversions": "ErrorCode2+", "typetaggedVersions": "2+", "int16tag": 0, "versionsignorable": "0+true",
      "about": "The client instance ID, for identifying an instance of an application." }
 "about": "The error code, or 0 if there was no error."
    }
  ]
} ]
}

GetTelemetrySubscriptions API

A very small behavioral change is made in the broker handling of GetTelemetrySubscriptions v0. If the client sends a zero ClientInstanceId  in the request body and it also sends a ClientInstanceId  in the request header, rather than generating a new UUID, the broker will respond with the ClientInstanceId  from the request header. This naturally aligns the request header and client telemetry values for ClientInstanceId .

Compatibility, Deprecation, and Migration Plan

...