DUE TO SPAM, SIGN-UP IS DISABLED. Goto Selfserve wiki signup and request an account.
...
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.
In addition, the client ID 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. Sending it repeatedly is wasteful.
Proposed Changes
This KIP proposes the following changes to the Kafka protocol.
...
If a connection uses the v3 request header in its first request, it must specify a non-zero client instance ID and it must specify the same client instance ID for all subsequent requests. Note that SaslHandshake is excluded from this checking because it cannot carry a client instance ID. 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 SaslHandshake as explained above) or which specify a different value for the client instance ID will be rejected with error code INVALID_REQUESTas explained above) or which specify a different value for the client instance ID will be rejected with error code INVALID_REQUEST.
Client ID
This KIP 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).
This new behavior can only be completely adopted when both client and broker support it. It works as follows:
- For the initial request regardless of request header version, the client sends the client ID
- For subsequent requests using the v3 request header (or higher), the client sends a null client ID
- For subsequent requests using an earlier version of the request header, the client sends the client ID
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.
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 client telemetry client instance ID and the member ID in the group protocols. The client instance ID from the request headers is used as the client telemetry client instance ID; new versions of the KIP-714 RPCs are introduced to ensure this. 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.
...
This KIP introduces a new property for the request schemas called "headerVersions" which contains a map from request version ranges to request header versions. The keys of the map are either can be single versions like "0" or , closed version ranges like "1-2", and must end with an open version range like "3+". The ranges of the keys must be non-overlapping, continuous and match cover the range of valid versions. For example, the DeleteGroups request schema becomes the following.
| Code Block |
|---|
{
"apiKey": 42,
"type": "request",
"listeners": ["broker"],
"name": "DeleteGroupsRequest",
"validVersions": "0-3",
"flexibleVersions": "2+",
"headerVersions": {
"0-1": "1",
"2": "2",
"3+": "3"
},
"fields": [
{ "name": "GroupsNames", "type": "[]string", "versions": "0+", "entityType": "groupId",
"about": "The group names to delete." }
]
} |
The modified schema describes 4 versions of the request. Request versions 0 to 1 use v1 of the request header. Request version 2 uses v2 of the request header. Request version 3 uses and higher use v3 of the request header, which includes the client instance ID.
...
| Code Block |
|---|
{
"apiKey": 42,
"type": "response",
"name": "DeleteGroupsResponse",
// Starting in version 1, on quota violation, brokers send out responses before throttling.
//
// Version 2 is the first flexible version.
"validVersions": "0-2",
"flexibleVersions": "2+",
"headerVersions": {
"0-1": "0",
"2+": "1"
},
"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": "Results", "type": "[]DeletableGroupResult", "versions": "0+",
"about": "The deletion results.", "fields": [
{ "name": "GroupId", "type": "string", "versions": "0+", "mapKey": true, "entityType": "groupId",
"about": "The group id." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The deletion error, or 0 if the deletion succeeded." }
]}
]
}
|
...
| Code Block |
|---|
{
"apiKey": 47,
"type": "request",
"listeners": ["broker"],
"name": "OffsetDeleteRequest",
"validVersions": "0-1",
"flexibleVersions": "1+",
"headerVersions": {
"0": "1",
"1+": "3"
},
"fields": [
{ "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
"about": "The unique group identifier." },
{ "name": "Topics", "type": "[]OffsetDeleteRequestTopic", "versions": "0+",
"about": "The topics to delete offsets for.", "fields": [
{ "name": "Name", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]OffsetDeleteRequestPartition", "versions": "0+",
"about": "Each partition to delete offsets for.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+",
"about": "The partition index." }
]
}
]
}
]
} |
...
| Code Block |
|---|
{
"apiKey": 47,
"type": "response",
"name": "OffsetDeleteResponse",
"validVersions": "0-1",
"flexibleVersions": "1+",
"headerVersions": {
"0": "0",
"1+": "1"
},
"fields": [
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The top-level error code, or 0 if there was no error." },
{ "name": "ThrottleTimeMs", "type": "int32", "versions": "0+", "ignorable": true,
"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": "Topics", "type": "[]OffsetDeleteResponseTopic", "versions": "0+",
"about": "The responses for each topic.", "fields": [
{ "name": "Name", "type": "string", "versions": "0+", "mapKey": true, "entityType": "topicName",
"about": "The topic name." },
{ "name": "Partitions", "type": "[]OffsetDeleteResponsePartition", "versions": "0+",
"about": "The responses for each partition in the topic.", "fields": [
{ "name": "PartitionIndex", "type": "int32", "versions": "0+", "mapKey": true,
"about": "The partition index." },
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
"about": "The error code, or 0 if there was no error." }
]
}
]
}
]
} |
...
| Code Block |
|---|
{
"apiKey": 71,
"type": "request",
"listeners": ["broker"],
"name": "GetTelemetrySubscriptionsRequest",
"validVersions": "0-1",
"flexibleVersions": "0+",
"headerVersions": {
"0": "2",
"1+": "3"
},
"fields": [
{
"name": "ClientInstanceId", "type": "uuid", "versions": "0",
"about": "Unique id for this client instance, must be set to 0 on the first request."
request." }
]
} |
Response
The v1 response removes the ClientInstanceId . With v1, the client is responsible for calculating the client instance ID.
| Code Block |
|---|
{
"apiKey": 71,
"type": "response",
"name": "GetTelemetrySubscriptionsResponse",
"validVersions": "0-1",
"flexibleVersions": "0+",
"headerVersions": {
"0-1+": "1"
},
"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",
"about": "Assigned client instance id if ClientInstanceId was 0 in the request, else 0."
},
{
"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
A new version of PushTelemetry is introduced which removes ClientInstanceId from the request.
...
| Code Block |
|---|
{
"apiKey": 72,
"type": "request",
"listeners": ["broker"],
"name": "PushTelemetryRequest",
"validVersions": "0-1",
"flexibleVersions": "0+",
"headerVersions": {
"0": "2",
"1+": "3"
},
"fields": [
{
"name": "ClientInstanceId", "type": "uuid", "versions": "0",
"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 is made to the schema of the response.
| Code Block |
|---|
{
"apiKey": 72,
"type": "response",
"name": "PushTelemetryResponse",
"validVersions": "0-1",
"flexibleVersions": "0+",
"headerVersions": {
"0-1+": "1"
},
"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
...