You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sc...@apache.org on 2026/09/29 08:32:51 UTC
(kafka) branch trunk updated: KAFKA-20924: Introduce Request Header v3 and use it for OffsetDelete v1 (#23544)
This is an automated email from the ASF dual-hosted git repository.
AndrewJSchofield pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 0606659cc13 KAFKA-20924: Introduce Request Header v3 and use it for OffsetDelete v1 (#23544)
0606659cc13 is described below
commit 0606659cc13dc5573dd9213215001d3409ef68d9
Author: Alan Lau <72...@users.noreply.github.com>
AuthorDate: Tue Sep 29 04:32:42 2026 -0400
KAFKA-20924: Introduce Request Header v3 and use it for OffsetDelete v1 (#23544)
Jira: https://issues.apache.org/jira/browse/KAFKA-20924
KIP-1313:
https://cwiki.apache.org/confluence/spaces/KAFKA/pages/406624038/KIP-1313+Client+instance+ID+in+all+request+headers
This PR introduces the v3 header and wires it up for OffsetDeleteRequest
v1 as the first RPC to use it, and updates every client (producer,
consumer, share consumer, admin client) to generate a client instance ID
at construction and include it whenever the negotiated request maps to
the v3 header. Older requests, or ones negotiated down to earlier
versions, keep using the existing header versions with no observable
behavior change. Headers v3 for telemetry RPCs will come in a later PR.
Reviewers: Andrew Schofield <as...@confluent.io>
---
.../org/apache/kafka/clients/ClientRequest.java | 19 +++---
.../java/org/apache/kafka/clients/ClientUtils.java | 5 ++
.../org/apache/kafka/clients/NetworkClient.java | 14 +++-
.../kafka/clients/admin/KafkaAdminClient.java | 2 +
.../consumer/internals/AsyncKafkaConsumer.java | 2 +
.../consumer/internals/ClassicKafkaConsumer.java | 2 +
.../clients/consumer/internals/ConsumerUtils.java | 3 +
.../consumer/internals/NetworkClientDelegate.java | 3 +
.../consumer/internals/ShareConsumerImpl.java | 2 +
.../kafka/clients/producer/KafkaProducer.java | 6 +-
.../kafka/common/requests/RequestHeader.java | 24 ++++++-
.../common/message/OffsetDeleteRequest.json | 9 ++-
.../common/message/OffsetDeleteResponse.json | 8 ++-
.../resources/common/message/RequestHeader.json | 8 ++-
.../kafka/clients/ClientInstanceIdCapture.java | 75 ++++++++++++++++++++++
.../apache/kafka/clients/NetworkClientTest.java | 67 +++++++++++++++++--
.../kafka/clients/admin/KafkaAdminClientTest.java | 8 +++
.../kafka/clients/consumer/KafkaConsumerTest.java | 24 +++++++
.../clients/consumer/KafkaShareConsumerTest.java | 10 +++
.../NetworkClientDelegateInstanceIdCapture.java | 46 +++++++++++++
.../kafka/clients/producer/KafkaProducerTest.java | 13 +++-
.../kafka/common/message/ApiMessageTypeTest.java | 5 +-
.../kafka/common/requests/RequestHeaderTest.java | 44 +++++++++++++
.../java/org/apache/kafka/clients/MockClient.java | 4 +-
.../apache/kafka/message/HeaderVersionsTest.java | 22 ++-----
.../server/util/InterBrokerSendThreadTest.java | 7 +-
26 files changed, 381 insertions(+), 51 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/ClientRequest.java b/clients/src/main/java/org/apache/kafka/clients/ClientRequest.java
index abba79567eb..208cf92e888 100644
--- a/clients/src/main/java/org/apache/kafka/clients/ClientRequest.java
+++ b/clients/src/main/java/org/apache/kafka/clients/ClientRequest.java
@@ -16,7 +16,7 @@
*/
package org.apache.kafka.clients;
-import org.apache.kafka.common.message.RequestHeaderData;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.requests.AbstractRequest;
import org.apache.kafka.common.requests.RequestHeader;
@@ -30,6 +30,7 @@ public final class ClientRequest {
private final AbstractRequest.Builder<?> requestBuilder;
private final int correlationId;
private final String clientId;
+ private final Uuid clientInstanceId;
private final long createdTimeMs;
private final boolean expectResponse;
private final int requestTimeoutMs;
@@ -40,14 +41,19 @@ public final class ClientRequest {
* @param requestBuilder The builder for the request to make
* @param correlationId The correlation id for this client request
* @param clientId The client ID to use for the header
+ * @param clientInstanceId The client instance ID to use for the header
* @param createdTimeMs The unix timestamp in milliseconds for the time at which this request was created.
* @param expectResponse Should we expect a response message or is this request complete once it is sent?
+ * @param requestTimeoutMs Upper bound time in milliseconds to await a response before disconnecting the socket and
+ * cancelling the request. The request may get cancelled sooner if the socket disconnects
+ * for any reason including if another pending request to the same node timed out first.
* @param callback A callback to execute when the response has been received (or null if no callback is necessary)
*/
public ClientRequest(String destination,
AbstractRequest.Builder<?> requestBuilder,
int correlationId,
String clientId,
+ Uuid clientInstanceId,
long createdTimeMs,
boolean expectResponse,
int requestTimeoutMs,
@@ -56,6 +62,7 @@ public final class ClientRequest {
this.requestBuilder = requestBuilder;
this.correlationId = correlationId;
this.clientId = clientId;
+ this.clientInstanceId = clientInstanceId;
this.createdTimeMs = createdTimeMs;
this.expectResponse = expectResponse;
this.requestTimeoutMs = requestTimeoutMs;
@@ -69,6 +76,7 @@ public final class ClientRequest {
", destination=" + destination +
", correlationId=" + correlationId +
", clientId=" + clientId +
+ ", clientInstanceId=" + clientInstanceId +
", createdTimeMs=" + createdTimeMs +
", requestBuilder=" + requestBuilder +
")";
@@ -83,14 +91,7 @@ public final class ClientRequest {
}
public RequestHeader makeHeader(short version) {
- ApiKeys requestApiKey = apiKey();
- return new RequestHeader(
- new RequestHeaderData()
- .setRequestApiKey(requestApiKey.id)
- .setRequestApiVersion(version)
- .setClientId(clientId)
- .setCorrelationId(correlationId),
- requestApiKey.requestHeaderVersion(version));
+ return new RequestHeader(apiKey(), version, clientId, clientInstanceId, correlationId);
}
public AbstractRequest.Builder<?> requestBuilder() {
diff --git a/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java b/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java
index 5cd5886bce7..9dd72708118 100644
--- a/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java
+++ b/clients/src/main/java/org/apache/kafka/clients/ClientUtils.java
@@ -17,6 +17,7 @@
package org.apache.kafka.clients;
import org.apache.kafka.common.KafkaException;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.ConfigException;
import org.apache.kafka.common.config.SaslConfigs;
@@ -230,6 +231,7 @@ public final class ClientUtils {
public static NetworkClient createNetworkClient(AbstractConfig config,
List<String> bootstrapServers,
+ Uuid clientInstanceId,
Metrics metrics,
String metricsGroupPrefix,
LogContext logContext,
@@ -242,6 +244,7 @@ public final class ClientUtils {
return createNetworkClient(config,
bootstrapServers,
config.getString(CommonClientConfigs.CLIENT_ID_CONFIG),
+ clientInstanceId,
metrics,
metricsGroupPrefix,
logContext,
@@ -259,6 +262,7 @@ public final class ClientUtils {
public static NetworkClient createNetworkClient(AbstractConfig config,
List<String> bootstrapServers,
String clientId,
+ Uuid clientInstanceId,
Metrics metrics,
String metricsGroupPrefix,
LogContext logContext,
@@ -288,6 +292,7 @@ public final class ClientUtils {
metadata,
selector,
clientId,
+ clientInstanceId,
maxInFlightRequestsPerConnection,
config.getLong(CommonClientConfigs.RECONNECT_BACKOFF_MS_CONFIG),
config.getLong(CommonClientConfigs.RECONNECT_BACKOFF_MAX_MS_CONFIG),
diff --git a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
index 2fdc289346e..7f184b6c72c 100644
--- a/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/NetworkClient.java
@@ -21,6 +21,7 @@ import org.apache.kafka.common.ClusterResource;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.BootstrapResolutionException;
import org.apache.kafka.common.errors.DisconnectException;
@@ -68,6 +69,7 @@ import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import java.util.Optional;
import java.util.Random;
import java.util.concurrent.CompletableFuture;
@@ -116,6 +118,9 @@ public class NetworkClient implements KafkaClient {
/* the client id used to identify this client in requests to the server */
private final String clientId;
+ /* the client instance id sent in the v3 request header */
+ private final Uuid clientInstanceId;
+
/* the current correlation id to use when sending requests to servers */
private int correlation;
@@ -226,6 +231,7 @@ public class NetworkClient implements KafkaClient {
metadata,
selector,
clientId,
+ Uuid.randomUuid(),
maxInFlightRequestsPerConnection,
reconnectBackoffMs,
reconnectBackoffMax,
@@ -269,6 +275,7 @@ public class NetworkClient implements KafkaClient {
metadata,
selector,
clientId,
+ Uuid.randomUuid(),
maxInFlightRequestsPerConnection,
reconnectBackoffMs,
reconnectBackoffMax,
@@ -312,6 +319,7 @@ public class NetworkClient implements KafkaClient {
null,
selector,
clientId,
+ Uuid.randomUuid(),
maxInFlightRequestsPerConnection,
reconnectBackoffMs,
reconnectBackoffMax,
@@ -337,6 +345,7 @@ public class NetworkClient implements KafkaClient {
Metadata metadata,
Selectable selector,
String clientId,
+ Uuid clientInstanceId,
int maxInFlightRequestsPerConnection,
long reconnectBackoffMs,
long reconnectBackoffMax,
@@ -369,6 +378,7 @@ public class NetworkClient implements KafkaClient {
}
this.selector = selector;
this.clientId = clientId;
+ this.clientInstanceId = Objects.requireNonNull(clientInstanceId, "clientInstanceId must not be null");
this.inFlightRequests = new InFlightRequests(maxInFlightRequestsPerConnection);
this.connectionStates = new ClusterConnectionStates(
reconnectBackoffMs, reconnectBackoffMax,
@@ -1769,8 +1779,8 @@ public class NetworkClient implements KafkaClient {
boolean expectResponse,
int requestTimeoutMs,
RequestCompletionHandler callback) {
- return new ClientRequest(nodeId, requestBuilder, nextCorrelationId(), clientId, createdTimeMs, expectResponse,
- requestTimeoutMs, callback);
+ return new ClientRequest(nodeId, requestBuilder, nextCorrelationId(), clientId, clientInstanceId, createdTimeMs,
+ expectResponse, requestTimeoutMs, callback);
}
public boolean discoverBrokerVersions() {
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
index de76a01dc8e..9253de92d78 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
@@ -614,9 +614,11 @@ public class KafkaAdminClient extends AdminClient {
config.originalsWithPrefix(CommonClientConfigs.METRICS_CONTEXT_PREFIX));
metrics = new Metrics(metricConfig, reporters, time, metricsContext);
+ Uuid clientInstanceId = Uuid.randomUuid();
networkClient = ClientUtils.createNetworkClient(config,
bootstrapAddressesToUse,
clientId,
+ clientInstanceId,
metrics,
"admin-client",
logContext,
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
index 811e058b56b..27993538ca4 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/AsyncKafkaConsumer.java
@@ -518,11 +518,13 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
// This FetchBuffer is shared between the application and network threads.
this.fetchBuffer = new FetchBuffer(logContext);
this.positionsValidator = new PositionsValidator(logContext, time, subscriptions, metadata);
+ Uuid clientInstanceId = Uuid.randomUuid();
final Supplier<NetworkClientDelegate> networkClientDelegateSupplier = NetworkClientDelegate.supplier(time,
logContext,
metadata,
config,
apiVersions,
+ clientInstanceId,
metrics,
fetchMetricsManager.throttleTimeSensor(),
clientTelemetryReporter.map(ClientTelemetryReporter::telemetrySender).orElse(null),
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java
index 0b61ca37ba6..0339a7e5886 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ClassicKafkaConsumer.java
@@ -198,7 +198,9 @@ public class ClassicKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
this.isolationLevel = fetchConfig.isolationLevel;
ApiVersions apiVersions = new ApiVersions();
+ Uuid clientInstanceId = Uuid.randomUuid();
this.client = createConsumerNetworkClient(config,
+ clientInstanceId,
metrics,
logContext,
apiVersions,
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerUtils.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerUtils.java
index 88a30625df0..aca83aedd2a 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerUtils.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerUtils.java
@@ -28,6 +28,7 @@ import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.InterruptException;
import org.apache.kafka.common.errors.TimeoutException;
import org.apache.kafka.common.metrics.KafkaMetricsContext;
@@ -80,6 +81,7 @@ public final class ConsumerUtils {
private static final Logger log = LoggerFactory.getLogger(ConsumerUtils.class);
public static ConsumerNetworkClient createConsumerNetworkClient(ConsumerConfig config,
+ Uuid clientInstanceId,
Metrics metrics,
LogContext logContext,
ApiVersions apiVersions,
@@ -90,6 +92,7 @@ public final class ConsumerUtils {
ClientTelemetrySender clientTelemetrySender) {
NetworkClient netClient = ClientUtils.createNetworkClient(config,
config.getList(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG),
+ clientInstanceId,
metrics,
CONSUMER_METRIC_GROUP_PREFIX,
logContext,
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java
index 8189011c091..730d186fe7e 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegate.java
@@ -29,6 +29,7 @@ import org.apache.kafka.clients.consumer.internals.events.BackgroundEventHandler
import org.apache.kafka.clients.consumer.internals.events.ErrorEvent;
import org.apache.kafka.clients.consumer.internals.metrics.AsyncConsumerMetrics;
import org.apache.kafka.common.Node;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.BootstrapResolutionException;
import org.apache.kafka.common.errors.DisconnectException;
@@ -479,6 +480,7 @@ public class NetworkClientDelegate implements AutoCloseable {
final Metadata metadata,
final ConsumerConfig config,
final ApiVersions apiVersions,
+ final Uuid clientInstanceId,
final Metrics metrics,
final Sensor throttleTimeSensor,
final ClientTelemetrySender clientTelemetrySender,
@@ -490,6 +492,7 @@ public class NetworkClientDelegate implements AutoCloseable {
protected NetworkClientDelegate create() {
KafkaClient client = ClientUtils.createNetworkClient(config,
config.getList(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG),
+ clientInstanceId,
metrics,
CONSUMER_METRIC_GROUP_PREFIX,
logContext,
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java
index dcb94cd321c..e4bb6339d15 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/internals/ShareConsumerImpl.java
@@ -284,12 +284,14 @@ public class ShareConsumerImpl<K, V> implements ShareConsumerDelegate<K, V> {
// This FetchBuffer is shared between the application and network threads.
this.fetchBuffer = new ShareFetchBuffer(logContext);
+ Uuid clientInstanceId = Uuid.randomUuid();
final Supplier<NetworkClientDelegate> networkClientDelegateSupplier = NetworkClientDelegate.supplier(
time,
logContext,
metadata,
config,
apiVersions,
+ clientInstanceId,
metrics,
shareFetchMetricsManager.throttleTimeSensor(),
clientTelemetryReporter.map(ClientTelemetryReporter::telemetrySender).orElse(null),
diff --git a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
index 109306b6402..895fd8731c1 100644
--- a/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/producer/KafkaProducer.java
@@ -512,7 +512,8 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
}
this.errors = this.metrics.sensor("errors");
- this.sender = newSender(logContext, kafkaClient, this.metadata);
+ Uuid clientInstanceId = Uuid.randomUuid();
+ this.sender = newSender(logContext, kafkaClient, this.metadata, clientInstanceId);
String ioThreadName = NETWORK_THREAD_PREFIX + " | " + clientId;
this.ioThread = new Sender.SenderThread(ioThreadName, this.sender, true);
this.ioThread.start();
@@ -568,13 +569,14 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
}
// visible for testing
- Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadata metadata) {
+ Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadata metadata, Uuid clientInstanceId) {
int maxInflightRequests = producerConfig.getInt(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION);
int requestTimeoutMs = producerConfig.getInt(ProducerConfig.REQUEST_TIMEOUT_MS_CONFIG);
ProducerMetrics metricsRegistry = new ProducerMetrics(this.metrics);
Sensor throttleTimeSensor = Sender.throttleTimeSensor(metricsRegistry.senderMetrics);
KafkaClient client = kafkaClient != null ? kafkaClient : ClientUtils.createNetworkClient(producerConfig,
producerConfig.getList(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG),
+ clientInstanceId,
this.metrics,
"producer",
logContext,
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java b/clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java
index a628811cde8..4ca813d0e47 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/RequestHeader.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.common.requests;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.message.RequestHeaderData;
@@ -36,12 +37,21 @@ public class RequestHeader implements AbstractRequestResponse {
private int size = SIZE_NOT_INITIALIZED;
public RequestHeader(ApiKeys requestApiKey, short requestVersion, String clientId, int correlationId) {
- this(new RequestHeaderData().
+ this(requestApiKey, requestVersion, clientId, null, correlationId);
+ }
+
+ public RequestHeader(ApiKeys requestApiKey, short requestVersion, String clientId, Uuid clientInstanceId,
+ int correlationId) {
+ this.headerVersion = requestApiKey.requestHeaderVersion(requestVersion);
+ this.data = new RequestHeaderData().
setRequestApiKey(requestApiKey.id).
setRequestApiVersion(requestVersion).
setClientId(clientId).
- setCorrelationId(correlationId),
- requestApiKey.requestHeaderVersion(requestVersion));
+ setCorrelationId(correlationId);
+ // The client instance ID is a tagged field from header version 3, so writing it at an older version fails.
+ // We only set it from header version 3. Otherwise, the header leaves the ID unset as its default, `Uuid.ZERO_UUID`.
+ if (clientInstanceId != null && headerVersion >= 3)
+ this.data.setClientInstanceId(clientInstanceId);
}
public RequestHeader(RequestHeaderData data, short headerVersion) {
@@ -69,6 +79,13 @@ public class RequestHeader implements AbstractRequestResponse {
return data.correlationId();
}
+ /**
+ * Returns {@link Uuid#ZERO_UUID} if the header predates v3 or the client sent no ID.
+ */
+ public Uuid clientInstanceId() {
+ return data.clientInstanceId();
+ }
+
public RequestHeaderData data() {
return data;
}
@@ -165,6 +182,7 @@ public class RequestHeader implements AbstractRequestResponse {
return "RequestHeader(apiKey=" + apiKey() +
", apiVersion=" + apiVersion() +
", clientId=" + clientId() +
+ ", clientInstanceId=" + clientInstanceId() +
", correlationId=" + correlationId() +
", headerVersion=" + headerVersion +
")";
diff --git a/clients/src/main/resources/common/message/OffsetDeleteRequest.json b/clients/src/main/resources/common/message/OffsetDeleteRequest.json
index d515ee58434..a411316ac11 100644
--- a/clients/src/main/resources/common/message/OffsetDeleteRequest.json
+++ b/clients/src/main/resources/common/message/OffsetDeleteRequest.json
@@ -18,10 +18,13 @@
"type": "request",
"listeners": ["broker"],
"name": "OffsetDeleteRequest",
- "validVersions": "0",
- "flexibleVersions": "none",
+ // Version 1 enables flexible versions and uses the v3 request header (KIP-1313).
+ "latestVersionUnstable": true,
+ "validVersions": "0-1",
+ "flexibleVersions": "1+",
"headerVersions": {
- "0+": "1"
+ "0": "1",
+ "1+": "3"
},
"fields": [
{ "name": "GroupId", "type": "string", "versions": "0+", "entityType": "groupId",
diff --git a/clients/src/main/resources/common/message/OffsetDeleteResponse.json b/clients/src/main/resources/common/message/OffsetDeleteResponse.json
index fbaa6f2cc38..f83dafaa9ed 100644
--- a/clients/src/main/resources/common/message/OffsetDeleteResponse.json
+++ b/clients/src/main/resources/common/message/OffsetDeleteResponse.json
@@ -17,10 +17,12 @@
"apiKey": 47,
"type": "response",
"name": "OffsetDeleteResponse",
- "validVersions": "0",
- "flexibleVersions": "none",
+ // Version 1 enables flexible versions.
+ "validVersions": "0-1",
+ "flexibleVersions": "1+",
"headerVersions": {
- "0+": "0"
+ "0": "0",
+ "1+": "1"
},
"fields": [
{ "name": "ErrorCode", "type": "int16", "versions": "0+",
diff --git a/clients/src/main/resources/common/message/RequestHeader.json b/clients/src/main/resources/common/message/RequestHeader.json
index 50589768fe1..1f63cf849b1 100644
--- a/clients/src/main/resources/common/message/RequestHeader.json
+++ b/clients/src/main/resources/common/message/RequestHeader.json
@@ -23,7 +23,9 @@
// Version 1 is the first version with ClientId.
//
// Version 2 is the first flexible version.
- "validVersions": "1-2",
+ //
+ // Version 3 introduces ClientInstanceId. (KIP-1313)
+ "validVersions": "1-3",
"flexibleVersions": "2+",
"fields": [
{ "name": "RequestApiKey", "type": "int16", "versions": "0+",
@@ -39,6 +41,8 @@
// 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." }
+ "about": "The client ID, for identifying an application." },
+ { "name": "ClientInstanceId", "type": "uuid", "versions": "3+", "taggedVersions": "3+", "tag": 0,
+ "about": "The client instance ID, for identifying an instance of an application." }
]
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/ClientInstanceIdCapture.java b/clients/src/test/java/org/apache/kafka/clients/ClientInstanceIdCapture.java
new file mode 100644
index 00000000000..6c7fcba7817
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/clients/ClientInstanceIdCapture.java
@@ -0,0 +1,75 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients;
+
+import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.utils.Utils;
+
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedStatic;
+import org.mockito.internal.stubbing.answers.CallsRealMethods;
+
+import java.util.function.Function;
+import java.util.function.Supplier;
+
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
+import static org.mockito.Mockito.mockStatic;
+
+/**
+ * Asserts that a client generates a client instance ID in its constructor and passes it to {@link ClientUtils}
+ * (KIP-1313).
+ */
+public class ClientInstanceIdCapture {
+
+ private ClientInstanceIdCapture() {
+ }
+
+ /**
+ * Asserts that the client passes a usable client instance ID to {@link ClientUtils}. Verifying the
+ * longer createNetworkClient overload also covers the shorter overload.
+ */
+ public static void assertGenerated(Supplier<? extends AutoCloseable> clientFactory) {
+ assertGenerated(ClientUtils.class,
+ captor -> () -> ClientUtils.createNetworkClient(any(), any(), any(), captor.capture(), any(), any(),
+ any(), any(), any(), anyInt(), anyInt(), any(), any(), any(), any(), any()),
+ clientFactory);
+ }
+
+ /**
+ * Asserts that the client passes a usable client instance ID to {@code factoryClass}.
+ */
+ public static <T> void assertGenerated(Class<T> factoryClass,
+ Function<ArgumentCaptor<Uuid>, MockedStatic.Verification> verification,
+ Supplier<? extends AutoCloseable> clientFactory) {
+ ArgumentCaptor<Uuid> captor = ArgumentCaptor.forClass(Uuid.class);
+ try (MockedStatic<T> mocked = mockStatic(factoryClass, new CallsRealMethods())) {
+ AutoCloseable client = clientFactory.get();
+ try {
+ mocked.verify(verification.apply(captor));
+ } finally {
+ Utils.closeQuietly(client, "client");
+ }
+ }
+ Uuid clientInstanceId = captor.getValue();
+ assertNotNull(clientInstanceId, "no client instance ID was passed to the network client");
+ // KIP-1313 does not permit a reserved UUID; Uuid.randomUuid never returns one.
+ assertFalse(Uuid.RESERVED.contains(clientInstanceId));
+ }
+}
diff --git a/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java b/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java
index fa1b4c4176e..c7f82343053 100644
--- a/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/NetworkClientTest.java
@@ -19,6 +19,7 @@ package org.apache.kafka.clients;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Node;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.RebootstrapRequiredException;
import org.apache.kafka.common.internals.ClusterResourceListeners;
@@ -29,6 +30,7 @@ import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersion;
import org.apache.kafka.common.message.ApiVersionsResponseData.ApiVersionCollection;
import org.apache.kafka.common.message.GetTelemetrySubscriptionsRequestData;
import org.apache.kafka.common.message.GetTelemetrySubscriptionsResponseData;
+import org.apache.kafka.common.message.OffsetDeleteRequestData;
import org.apache.kafka.common.message.ProduceRequestData;
import org.apache.kafka.common.message.ProduceResponseData;
import org.apache.kafka.common.message.PushTelemetryRequestData;
@@ -42,6 +44,7 @@ import org.apache.kafka.common.requests.GetTelemetrySubscriptionsRequest;
import org.apache.kafka.common.requests.GetTelemetrySubscriptionsResponse;
import org.apache.kafka.common.requests.MetadataRequest;
import org.apache.kafka.common.requests.MetadataResponse;
+import org.apache.kafka.common.requests.OffsetDeleteRequest;
import org.apache.kafka.common.requests.ProduceRequest;
import org.apache.kafka.common.requests.ProduceResponse;
import org.apache.kafka.common.requests.PushTelemetryRequest;
@@ -198,6 +201,60 @@ public class NetworkClientTest {
CommonClientConfigs.DEFAULT_RETRY_BACKOFF_MS);
}
+ @Test
+ public void testClientInstanceIdIsSentInTheV3RequestHeader() {
+ Uuid clientInstanceId = Uuid.randomUuid();
+ NetworkClient client = new NetworkClient(metadataUpdater, null, selector, "mock", clientInstanceId, Integer.MAX_VALUE,
+ reconnectBackoffMsTest, reconnectBackoffMaxMsTest, 64 * 1024, 64 * 1024,
+ defaultRequestTimeoutMs, connectionSetupTimeoutMsTest, connectionSetupTimeoutMaxMsTest,
+ time, true, new ApiVersions(), null, new LogContext(), new DefaultHostResolver(), null,
+ Long.MAX_VALUE, MetadataRecoveryStrategy.NONE, BootstrapConfiguration.DISABLED, false);
+ // OffsetDelete v1 uses the v3 request header, while v0 does not.
+ ClientRequest request = client.newClientRequest(node.idString(),
+ new OffsetDeleteRequest.Builder(new OffsetDeleteRequestData()), time.milliseconds(), true);
+ assertEquals(clientInstanceId, request.makeHeader((short) 1).clientInstanceId());
+ assertEquals(Uuid.ZERO_UUID, request.makeHeader((short) 0).clientInstanceId());
+ }
+
+ @Test
+ public void testClientInstanceIdIsRequired() {
+ // Pass a null client instance ID, which the constructor should reject
+ assertThrows(NullPointerException.class, () -> new NetworkClient(metadataUpdater, null, selector, "mock", null,
+ Integer.MAX_VALUE, reconnectBackoffMsTest, reconnectBackoffMaxMsTest, 64 * 1024, 64 * 1024,
+ defaultRequestTimeoutMs, connectionSetupTimeoutMsTest, connectionSetupTimeoutMaxMsTest,
+ time, true, new ApiVersions(), null, new LogContext(), new DefaultHostResolver(), null,
+ Long.MAX_VALUE, MetadataRecoveryStrategy.NONE, BootstrapConfiguration.DISABLED, false));
+ }
+
+ @Test
+ public void testShortConstructorsGenerateClientInstanceId() {
+ Metadata metadata = new Metadata(50, 50, 5000, new LogContext(), new ClusterResourceListeners());
+ assertGeneratedClientInstanceId(new NetworkClient(selector, metadata, "mock", Integer.MAX_VALUE,
+ reconnectBackoffMsTest, 0, 64 * 1024, 64 * 1024,
+ defaultRequestTimeoutMs, connectionSetupTimeoutMsTest, connectionSetupTimeoutMaxMsTest, time, false, new ApiVersions(), new LogContext(),
+ MetadataRecoveryStrategy.NONE, BootstrapConfiguration.DISABLED, false));
+ assertGeneratedClientInstanceId(new NetworkClient(selector, metadata, "mock", Integer.MAX_VALUE,
+ reconnectBackoffMsTest, 0, 64 * 1024, 64 * 1024,
+ defaultRequestTimeoutMs, connectionSetupTimeoutMsTest, connectionSetupTimeoutMaxMsTest, time, false, new ApiVersions(), new LogContext(),
+ Long.MAX_VALUE, MetadataRecoveryStrategy.NONE, BootstrapConfiguration.DISABLED, false));
+ assertGeneratedClientInstanceId(new NetworkClient(selector, metadata, "mock", Integer.MAX_VALUE,
+ reconnectBackoffMsTest, 0, 64 * 1024, 64 * 1024,
+ defaultRequestTimeoutMs, connectionSetupTimeoutMsTest, connectionSetupTimeoutMaxMsTest, time, false, new ApiVersions(), null, new LogContext(),
+ MetadataRecoveryStrategy.NONE, false));
+ assertGeneratedClientInstanceId(new NetworkClient(selector, metadataUpdater, "mock", Integer.MAX_VALUE,
+ reconnectBackoffMsTest, 0, 64 * 1024, 64 * 1024,
+ defaultRequestTimeoutMs, connectionSetupTimeoutMsTest, connectionSetupTimeoutMaxMsTest, time, false, new ApiVersions(), new LogContext(),
+ MetadataRecoveryStrategy.NONE, BootstrapConfiguration.DISABLED, false));
+ }
+
+ private static void assertGeneratedClientInstanceId(NetworkClient client) {
+ // OffsetDelete v1 uses the v3 request header.
+ Uuid clientInstanceId = client.newClientRequest("0", new OffsetDeleteRequest.Builder(new OffsetDeleteRequestData()), 0, true)
+ .makeHeader((short) 1).clientInstanceId();
+ // KIP-1313 does not permit a reserved UUID; Uuid.randomUuid never returns one.
+ assertFalse(Uuid.RESERVED.contains(clientInstanceId));
+ }
+
@Test
public void testSendToUnreadyNode() {
MetadataRequest.Builder builder = new MetadataRequest.Builder(Collections.singletonList("test"), true);
@@ -1214,7 +1271,7 @@ public class NetworkClientTest {
ClientTelemetrySender mockClientTelemetrySender = mock(ClientTelemetrySender.class);
when(mockClientTelemetrySender.timeToNextUpdate(anyLong())).thenReturn(0L);
- NetworkClient client = new NetworkClient(metadataUpdater, null, selector, "mock", Integer.MAX_VALUE,
+ NetworkClient client = new NetworkClient(metadataUpdater, null, selector, "mock", Uuid.randomUuid(), Integer.MAX_VALUE,
reconnectBackoffMsTest, reconnectBackoffMaxMsTest, 64 * 1024, 64 * 1024,
defaultRequestTimeoutMs, connectionSetupTimeoutMsTest, connectionSetupTimeoutMaxMsTest,
time, false, new ApiVersions(), null, new LogContext(), mockHostResolver, mockClientTelemetrySender,
@@ -1275,7 +1332,7 @@ public class NetworkClientTest {
ClientTelemetrySender mockClientTelemetrySender = mock(ClientTelemetrySender.class);
when(mockClientTelemetrySender.timeToNextUpdate(anyLong())).thenReturn(0L);
- NetworkClient client = new NetworkClient(metadataUpdater, null, selector, "mock", Integer.MAX_VALUE,
+ NetworkClient client = new NetworkClient(metadataUpdater, null, selector, "mock", Uuid.randomUuid(), Integer.MAX_VALUE,
reconnectBackoffMsTest, reconnectBackoffMaxMsTest, 64 * 1024, 64 * 1024,
defaultRequestTimeoutMs, connectionSetupTimeoutMsTest, connectionSetupTimeoutMaxMsTest,
time, false, new ApiVersions(), null, new LogContext(), mockHostResolver, mockClientTelemetrySender,
@@ -1328,7 +1385,7 @@ public class NetworkClientTest {
ClientTelemetrySender mockClientTelemetrySender = mock(ClientTelemetrySender.class);
when(mockClientTelemetrySender.timeToNextUpdate(anyLong())).thenReturn(0L);
- NetworkClient client = new NetworkClient(metadataUpdater, null, selector, "mock", Integer.MAX_VALUE,
+ NetworkClient client = new NetworkClient(metadataUpdater, null, selector, "mock", Uuid.randomUuid(), Integer.MAX_VALUE,
reconnectBackoffMsTest, reconnectBackoffMaxMsTest, 64 * 1024, 64 * 1024,
defaultRequestTimeoutMs, connectionSetupTimeoutMsTest, connectionSetupTimeoutMaxMsTest,
time, false, new ApiVersions(), null, new LogContext(), mockHostResolver, mockClientTelemetrySender,
@@ -1437,7 +1494,7 @@ public class NetworkClientTest {
ClientTelemetrySender mockClientTelemetrySender = mock(ClientTelemetrySender.class);
when(mockClientTelemetrySender.timeToNextUpdate(anyLong())).thenReturn(0L);
- NetworkClient client = new NetworkClient(metadataUpdater, null, selector, "mock", Integer.MAX_VALUE,
+ NetworkClient client = new NetworkClient(metadataUpdater, null, selector, "mock", Uuid.randomUuid(), Integer.MAX_VALUE,
reconnectBackoffMsTest, reconnectBackoffMaxMsTest, 64 * 1024, 64 * 1024,
defaultRequestTimeoutMs, connectionSetupTimeoutMsTest, connectionSetupTimeoutMaxMsTest,
time, true, new ApiVersions(), null, new LogContext(), new DefaultHostResolver(), mockClientTelemetrySender,
@@ -1556,7 +1613,7 @@ public class NetworkClientTest {
ManualMetadataUpdater updater = new ManualMetadataUpdater(Collections.singletonList(staleNode));
NetworkClient testClient = new NetworkClient(
- updater, null, capturingSelector, "test-client",
+ updater, null, capturingSelector, "test-client", Uuid.randomUuid(),
Integer.MAX_VALUE,
0L, 0L, // reconnectBackoffMs = 0 for instant reconnect
64 * 1024, 64 * 1024,
diff --git a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
index 21b170eb301..7cccdac0651 100644
--- a/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.clients.admin;
+import org.apache.kafka.clients.ClientInstanceIdCapture;
import org.apache.kafka.clients.ClientRequest;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.MetadataRecoveryStrategy;
@@ -1856,4 +1857,11 @@ public class KafkaAdminClientTest extends KafkaAdminClientTestBase {
KafkaException e = assertThrows(KafkaException.class, () -> Admin.create(configs));
assertInstanceOf(ConfigException.class, e.getCause());
}
+
+ @Test
+ public void testClientInstanceIdIsPassedToTheNetworkClient() {
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ ClientInstanceIdCapture.assertGenerated(() -> Admin.create(configs));
+ }
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
index ab3c0de50fa..f14ccec1731 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaConsumerTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.clients.consumer;
+import org.apache.kafka.clients.ClientInstanceIdCapture;
import org.apache.kafka.clients.ClientRequest;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.KafkaClient;
@@ -30,6 +31,7 @@ import org.apache.kafka.clients.consumer.internals.ConsumerProtocol;
import org.apache.kafka.clients.consumer.internals.Fetcher;
import org.apache.kafka.clients.consumer.internals.GroupCoordinatorNode;
import org.apache.kafka.clients.consumer.internals.MockRebalanceListener;
+import org.apache.kafka.clients.consumer.internals.NetworkClientDelegateInstanceIdCapture;
import org.apache.kafka.clients.consumer.internals.SubscriptionState;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.IsolationLevel;
@@ -4488,4 +4490,26 @@ public void testPollIdleRatio(GroupProtocol groupProtocol) {
CLOSE_COUNT.set(0);
}
}
+
+ private Map<String, Object> clientInstanceIdConfigs(GroupProtocol groupProtocol) {
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ configs.put(ConsumerConfig.GROUP_ID_CONFIG, "group");
+ configs.put(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name().toLowerCase(Locale.ROOT));
+ return configs;
+ }
+
+ @Test
+ public void testClassicConsumerPassesTheClientInstanceIdToTheNetworkClient() {
+ Map<String, Object> configs = clientInstanceIdConfigs(GroupProtocol.CLASSIC);
+ ClientInstanceIdCapture.assertGenerated(
+ () -> new KafkaConsumer<>(configs, new StringDeserializer(), new StringDeserializer()));
+ }
+
+ @Test
+ public void testAsyncConsumerPassesTheClientInstanceIdToTheNetworkClient() {
+ Map<String, Object> configs = clientInstanceIdConfigs(GroupProtocol.CONSUMER);
+ NetworkClientDelegateInstanceIdCapture.assertGenerated(
+ () -> new KafkaConsumer<>(configs, new StringDeserializer(), new StringDeserializer()));
+ }
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerTest.java
index 05353a5464f..fe57348c2a1 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerTest.java
@@ -21,6 +21,7 @@ import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.clients.MockClient;
import org.apache.kafka.clients.consumer.internals.AutoOffsetResetStrategy;
import org.apache.kafka.clients.consumer.internals.GroupCoordinatorNode;
+import org.apache.kafka.clients.consumer.internals.NetworkClientDelegateInstanceIdCapture;
import org.apache.kafka.clients.consumer.internals.ShareConsumerMetadata;
import org.apache.kafka.clients.consumer.internals.SubscriptionState;
import org.apache.kafka.common.KafkaException;
@@ -477,4 +478,13 @@ public class KafkaShareConsumerTest {
KafkaException e = assertThrows(KafkaException.class, () -> new KafkaShareConsumer<>(configs));
assertInstanceOf(ConfigException.class, e.getCause());
}
+
+ @Test
+ public void testClientInstanceIdIsPassedToTheNetworkClient() {
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ configs.put(ConsumerConfig.GROUP_ID_CONFIG, "group");
+ NetworkClientDelegateInstanceIdCapture.assertGenerated(
+ () -> new KafkaShareConsumer<>(configs, new StringDeserializer(), new StringDeserializer()));
+ }
}
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegateInstanceIdCapture.java b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegateInstanceIdCapture.java
new file mode 100644
index 00000000000..e8d13a48ab7
--- /dev/null
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/internals/NetworkClientDelegateInstanceIdCapture.java
@@ -0,0 +1,46 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.kafka.clients.consumer.internals;
+
+import org.apache.kafka.clients.ClientInstanceIdCapture;
+
+import java.util.function.Supplier;
+
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyBoolean;
+
+/**
+ * Asserts that a consumer generates a client instance ID in its constructor and passes it to
+ * {@link NetworkClientDelegate#supplier} (KIP-1313).
+ */
+public class NetworkClientDelegateInstanceIdCapture {
+
+ private NetworkClientDelegateInstanceIdCapture() {
+ }
+
+ /**
+ * Asserts that the consumer passes a usable client instance ID to {@link NetworkClientDelegate#supplier}.
+ * The consumer does not call {@link org.apache.kafka.clients.ClientUtils} on the constructing thread, and a
+ * static mock only covers the thread that created it, so this verifies the call it does make there instead.
+ */
+ public static void assertGenerated(Supplier<? extends AutoCloseable> consumerFactory) {
+ ClientInstanceIdCapture.assertGenerated(NetworkClientDelegate.class,
+ captor -> () -> NetworkClientDelegate.supplier(any(), any(), any(), any(), any(), captor.capture(), any(),
+ any(), any(), any(), anyBoolean(), any()),
+ consumerFactory);
+ }
+}
diff --git a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
index 6180ffee369..68904ee798d 100644
--- a/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/producer/KafkaProducerTest.java
@@ -18,6 +18,7 @@ package org.apache.kafka.clients.producer;
import org.apache.kafka.clients.ApiVersions;
import org.apache.kafka.clients.ClientDnsLookup;
+import org.apache.kafka.clients.ClientInstanceIdCapture;
import org.apache.kafka.clients.ClientUtils;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.KafkaClient;
@@ -789,9 +790,9 @@ public class KafkaProducerTest {
new ProducerConfig(ProducerConfig.appendSerializerToConfig(configs, new StringSerializer(), new StringSerializer())),
new StringSerializer(), new StringSerializer(), metadata, mockClient, null, new ApiVersions(), time) {
@Override
- Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadata metadata) {
+ Sender newSender(LogContext logContext, KafkaClient kafkaClient, ProducerMetadata metadata, Uuid clientInstanceId) {
// give Sender its own Metadata instance so that we can isolate Metadata calls from KafkaProducer
- return super.newSender(logContext, kafkaClient, newMetadata(0, 0, 100_000));
+ return super.newSender(logContext, kafkaClient, newMetadata(0, 0, 100_000), clientInstanceId);
}
};
}
@@ -3474,4 +3475,12 @@ public class KafkaProducerTest {
KafkaException e = assertThrows(KafkaException.class, () -> new KafkaProducer<>(configs));
assertInstanceOf(ConfigException.class, e.getCause());
}
+
+ @Test
+ public void testClientInstanceIdIsPassedToTheNetworkClient() {
+ Map<String, Object> configs = new HashMap<>();
+ configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
+ ClientInstanceIdCapture.assertGenerated(
+ () -> new KafkaProducer<>(configs, new StringSerializer(), new StringSerializer()));
+ }
}
diff --git a/clients/src/test/java/org/apache/kafka/common/message/ApiMessageTypeTest.java b/clients/src/test/java/org/apache/kafka/common/message/ApiMessageTypeTest.java
index 8ca2d5ee908..c78c5a60dca 100644
--- a/clients/src/test/java/org/apache/kafka/common/message/ApiMessageTypeTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/message/ApiMessageTypeTest.java
@@ -89,14 +89,17 @@ public class ApiMessageTypeTest {
assertEquals((short) 2, ApiMessageType.CREATE_TOPICS.requestHeaderVersion((short) 5));
assertEquals((short) 1, ApiMessageType.CREATE_TOPICS.responseHeaderVersion((short) 5));
- // SaslHandshake and OffsetDelete are non-flexible: header v1 request / v0 response at every version.
+ // SaslHandshake is non-flexible: header v1 request / v0 response at every version.
assertEquals((short) 1, ApiMessageType.SASL_HANDSHAKE.requestHeaderVersion((short) 0));
assertEquals((short) 0, ApiMessageType.SASL_HANDSHAKE.responseHeaderVersion((short) 0));
assertEquals((short) 1, ApiMessageType.SASL_HANDSHAKE.requestHeaderVersion((short) 1));
assertEquals((short) 0, ApiMessageType.SASL_HANDSHAKE.responseHeaderVersion((short) 1));
+ // OffsetDelete v0 is non-flexible; v1 is flexible: header v3 request / v1 response (KIP-1313).
assertEquals((short) 1, ApiMessageType.OFFSET_DELETE.requestHeaderVersion((short) 0));
assertEquals((short) 0, ApiMessageType.OFFSET_DELETE.responseHeaderVersion((short) 0));
+ assertEquals((short) 3, ApiMessageType.OFFSET_DELETE.requestHeaderVersion((short) 1));
+ assertEquals((short) 1, ApiMessageType.OFFSET_DELETE.responseHeaderVersion((short) 1));
// ApiVersions request follows the flexible rule, but the response always uses a v0 header (KIP-511).
assertEquals((short) 1, ApiMessageType.API_VERSIONS.requestHeaderVersion((short) 0));
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestHeaderTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestHeaderTest.java
index 5c441c82d2a..7432a4a3f89 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/RequestHeaderTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestHeaderTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.common.requests;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.message.RequestHeaderData;
import org.apache.kafka.common.protocol.ApiKeys;
@@ -60,6 +61,49 @@ public class RequestHeaderTest {
assertEquals(header, deserialized);
}
+ @Test
+ public void testRequestHeaderV3WithoutClientInstanceId() {
+ // OffsetDelete v1 uses the v3 request header.
+ short apiVersion = 1;
+ RequestHeader header = new RequestHeader(ApiKeys.OFFSET_DELETE, apiVersion, "", 10);
+ assertEquals(3, header.headerVersion());
+
+ // The client instance ID is tagged, so a v3 header which leaves it unset is the size of a v2 header.
+ ByteBuffer buffer = RequestTestUtils.serializeRequestHeader(header);
+ assertEquals(11, buffer.remaining());
+ RequestHeader deserialized = RequestHeader.parse(buffer);
+ assertEquals(header, deserialized);
+ assertEquals(Uuid.ZERO_UUID, deserialized.data().clientInstanceId());
+ }
+
+ @Test
+ public void testRequestHeaderV3WithClientInstanceId() {
+ Uuid clientInstanceId = Uuid.randomUuid();
+ RequestHeader header = new RequestHeader(ApiKeys.OFFSET_DELETE, (short) 1, "", clientInstanceId, 10);
+ assertEquals(3, header.headerVersion());
+ assertEquals(clientInstanceId, header.clientInstanceId());
+
+ // The 10 bytes of header fields, plus the tagged field's count, tag, size and 16-byte UUID.
+ ByteBuffer buffer = RequestTestUtils.serializeRequestHeader(header);
+ assertEquals(29, buffer.remaining());
+ RequestHeader deserialized = RequestHeader.parse(buffer);
+ assertEquals(header, deserialized);
+ assertEquals(clientInstanceId, deserialized.clientInstanceId());
+ }
+
+ @Test
+ public void testClientInstanceIdIsUnsetBelowTheV3Header() {
+ // OffsetDelete v0 uses the v1 request header, which has no ClientInstanceId field, so the header
+ // leaves the ID unset as its default, ZERO_UUID.
+ RequestHeader header = new RequestHeader(ApiKeys.OFFSET_DELETE, (short) 0, "", Uuid.randomUuid(), 10);
+ assertEquals(1, header.headerVersion());
+ assertEquals(Uuid.ZERO_UUID, header.clientInstanceId());
+
+ ByteBuffer buffer = RequestTestUtils.serializeRequestHeader(header);
+ assertEquals(10, buffer.remaining());
+ assertEquals(header, RequestHeader.parse(buffer));
+ }
+
@Test
public void parseHeaderFromBufferWithNonZeroPosition() {
ByteBuffer buffer = ByteBuffer.allocate(64);
diff --git a/clients/src/testFixtures/java/org/apache/kafka/clients/MockClient.java b/clients/src/testFixtures/java/org/apache/kafka/clients/MockClient.java
index e24c9f4dc95..f1cb624458d 100644
--- a/clients/src/testFixtures/java/org/apache/kafka/clients/MockClient.java
+++ b/clients/src/testFixtures/java/org/apache/kafka/clients/MockClient.java
@@ -17,6 +17,7 @@
package org.apache.kafka.clients;
import org.apache.kafka.common.Node;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.InterruptException;
import org.apache.kafka.common.errors.UnsupportedVersionException;
@@ -69,6 +70,7 @@ public class MockClient implements KafkaClient {
}
+ private final Uuid clientInstanceId = Uuid.randomUuid();
private int correlation;
private Runnable wakeupHook;
private boolean advanceTimeDuringPoll;
@@ -573,7 +575,7 @@ public class MockClient implements KafkaClient {
boolean expectResponse,
int requestTimeoutMs,
RequestCompletionHandler callback) {
- return new ClientRequest(nodeId, requestBuilder, correlation++, "mockClientId", createdTimeMs,
+ return new ClientRequest(nodeId, requestBuilder, correlation++, "mockClientId", clientInstanceId, createdTimeMs,
expectResponse, requestTimeoutMs, callback);
}
diff --git a/generator/src/test/java/org/apache/kafka/message/HeaderVersionsTest.java b/generator/src/test/java/org/apache/kafka/message/HeaderVersionsTest.java
index ee583abf4a2..36e7a31ef58 100644
--- a/generator/src/test/java/org/apache/kafka/message/HeaderVersionsTest.java
+++ b/generator/src/test/java/org/apache/kafka/message/HeaderVersionsTest.java
@@ -42,10 +42,10 @@ public class HeaderVersionsTest {
"', 'flexibleVersions': '" + flexibleVersions + "'}");
}
- // The real RequestHeader / ResponseHeader shapes: request header 1-2 (flexible from 2),
+ // The real RequestHeader / ResponseHeader shapes: request header 1-3 (flexible from 2),
// response header 0-1 (flexible from 1).
private static MessageSpec requestHeader() throws Exception {
- return headerSpec("RequestHeader", "1-2", "2+");
+ return headerSpec("RequestHeader", "1-3", "2+");
}
private static MessageSpec responseHeader() throws Exception {
@@ -337,11 +337,11 @@ public class HeaderVersionsTest {
@Test
public void testRequestHeaderVersionMustExist() {
- // The synthetic request header schema stops at v2, so a schema may not declare header v3.
+ // The request header schema stops at v3, so a schema may not declare header v4.
assertMessageContains("does not exist",
- () -> checkHeaderVersions(parse(requestSpec("0-5", "0+", "{'0+': '3'}"))));
+ () -> checkHeaderVersions(parse(requestSpec("0-5", "0+", "{'0+': '4'}"))));
assertMessageContains("does not exist",
- () -> checkHeaderVersions(parse(requestSpec("0-6", "3+", "{'0-2': '1', '3-5': '2', '6+': '3'}"))));
+ () -> checkHeaderVersions(parse(requestSpec("0-6", "3+", "{'0-2': '1', '3-5': '2', '6+': '4'}"))));
}
@Test
@@ -360,9 +360,9 @@ public class HeaderVersionsTest {
@Test
public void testHighestExistingHeaderVersionsAccepted() throws Exception {
- MessageSpec spec = parse(requestSpec("0-5", "0+", "{'0+': '2'}"));
+ MessageSpec spec = parse(requestSpec("0-5", "0+", "{'0+': '3'}"));
checkHeaderVersions(spec);
- assertEquals((short) 2, spec.headerVersions().orElseThrow().entries().get(0).headerVersion());
+ assertEquals((short) 3, spec.headerVersions().orElseThrow().entries().get(0).headerVersion());
spec = parse(responseSpec(0, "FooResponse", "0-5", "0+", "{'0+': '1'}"));
checkHeaderVersions(spec);
assertEquals((short) 1, spec.headerVersions().orElseThrow().entries().get(0).headerVersion());
@@ -405,14 +405,6 @@ public class HeaderVersionsTest {
"'flexibleVersions': '3+', 'headerVersions': {'0+': '1'}}")));
}
- @Test
- public void testHeaderVersionThreeAcceptedWhenHeaderSchemaAllowsIt() throws Exception {
- // Once the request header schema is bumped to 1-3, a map may declare header v3.
- MessageSpec spec = parse(requestSpec("0-5", "2+", "{'0-1': '1', '2': '2', '3+': '3'}"));
- spec.checkHeaderVersions(headerSpec("RequestHeader", "1-3", "2+"), responseHeader());
- assertEquals((short) 3, spec.headerVersions().orElseThrow().entries().get(2).headerVersion());
- }
-
@Test
public void testFirstFlexibleHeaderFollowsHeaderSchema() {
// A header schema that only becomes flexible at v3 makes a flexible message version mapped to header v2 fail.
diff --git a/server-common/src/test/java/org/apache/kafka/server/util/InterBrokerSendThreadTest.java b/server-common/src/test/java/org/apache/kafka/server/util/InterBrokerSendThreadTest.java
index e2417e40cfe..3efdc47bb92 100644
--- a/server-common/src/test/java/org/apache/kafka/server/util/InterBrokerSendThreadTest.java
+++ b/server-common/src/test/java/org/apache/kafka/server/util/InterBrokerSendThreadTest.java
@@ -21,6 +21,7 @@ import org.apache.kafka.clients.ClientResponse;
import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.clients.RequestCompletionHandler;
import org.apache.kafka.common.Node;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.DisconnectException;
import org.apache.kafka.common.internals.FatalExitError;
@@ -166,7 +167,7 @@ public class InterBrokerSendThreadTest {
final TestInterBrokerSendThread sendThread = new TestInterBrokerSendThread();
final ClientRequest clientRequest =
- new ClientRequest("dest", request, 0, "1", 0, true, requestTimeoutMs, handler.handler);
+ new ClientRequest("dest", request, 0, "1", Uuid.randomUuid(), 0, true, requestTimeoutMs, handler.handler);
when(networkClient.newClientRequest(
ArgumentMatchers.eq("1"),
@@ -209,7 +210,7 @@ public class InterBrokerSendThreadTest {
final TestInterBrokerSendThread sendThread = new TestInterBrokerSendThread();
final ClientRequest clientRequest =
- new ClientRequest("dest", request, 0, "1", 0, true, requestTimeoutMs, handler.handler);
+ new ClientRequest("dest", request, 0, "1", Uuid.randomUuid(), 0, true, requestTimeoutMs, handler.handler);
when(networkClient.newClientRequest(
ArgumentMatchers.eq("1"),
@@ -261,7 +262,7 @@ public class InterBrokerSendThreadTest {
final ClientRequest clientRequest =
new ClientRequest(
- "dest", request, 0, "1", time.milliseconds(), true, requestTimeoutMs, handler.handler);
+ "dest", request, 0, "1", Uuid.randomUuid(), time.milliseconds(), true, requestTimeoutMs, handler.handler);
time.sleep(1500L);
when(networkClient.newClientRequest(