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/10/14 09:52:08 UTC
(kafka) branch trunk updated: KAFKA-20924: Use Request Header v3 for GetTelemetrySubscriptions and PushTelemetry v1 (#23651)
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 515a8c50c88 KAFKA-20924: Use Request Header v3 for GetTelemetrySubscriptions and PushTelemetry v1 (#23651)
515a8c50c88 is described below
commit 515a8c50c88e03a4a84f3b1175140802997bd830
Author: Alan Lau <72...@users.noreply.github.com>
AuthorDate: Wed Oct 14 05:52:00 2026 -0400
KAFKA-20924: Use Request Header v3 for GetTelemetrySubscriptions and PushTelemetry v1 (#23651)
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
Follows #23544. This PR introduces v1 of GetTelemetrySubscriptions and
PushTelemetry, which move the client instance ID from the request body
into the v3 request header, and makes client telemetry use the client
instance ID that each client generates at construction.
Reviewers: Andrew Schofield <as...@confluent.io>
---
.../apache/kafka/clients/CommonClientConfigs.java | 5 +-
.../java/org/apache/kafka/clients/admin/Admin.java | 6 +-
.../kafka/clients/admin/KafkaAdminClient.java | 7 +-
.../kafka/clients/consumer/KafkaConsumer.java | 6 +-
.../kafka/clients/consumer/KafkaShareConsumer.java | 6 +-
.../consumer/internals/AsyncKafkaConsumer.java | 4 +-
.../consumer/internals/ClassicKafkaConsumer.java | 4 +-
.../consumer/internals/ShareConsumerImpl.java | 4 +-
.../kafka/clients/producer/KafkaProducer.java | 16 +-
.../requests/GetTelemetrySubscriptionsRequest.java | 6 +
.../common/requests/PushTelemetryRequest.java | 6 +
.../internals/ClientTelemetryReporter.java | 39 ++--
.../telemetry/internals/ClientTelemetrySender.java | 2 +-
.../telemetry/internals/ClientTelemetryUtils.java | 2 +-
.../message/GetTelemetrySubscriptionsRequest.json | 15 +-
.../message/GetTelemetrySubscriptionsResponse.json | 59 +++---
.../common/message/PushTelemetryRequest.json | 39 ++--
.../common/message/PushTelemetryResponse.json | 17 +-
.../kafka/clients/admin/KafkaAdminClientTest.java | 4 +-
.../kafka/clients/consumer/KafkaConsumerTest.java | 6 +-
.../consumer/KafkaShareConsumerMetricsTest.java | 4 +-
.../kafka/clients/producer/KafkaProducerTest.java | 24 +--
.../kafka/common/message/ApiMessageTypeTest.java | 8 +
.../GetTelemetrySubscriptionsRequestTest.java | 26 +++
.../common/requests/PushTelemetryRequestTest.java | 30 +++
.../kafka/common/requests/RequestResponseTest.java | 9 +-
.../internals/ClientTelemetryReporterTest.java | 56 +++---
.../internals/ClientTelemetryUtilsTest.java | 1 +
.../java/org/apache/kafka/clients/MockClient.java | 4 +
.../clients/admin/AdminClientUnitTestEnv.java | 2 +-
.../apache/kafka/server/ClientMetricsManager.java | 59 ++++--
.../metrics/ClientTelemetryExporterPlugin.java | 9 +-
.../metrics/DefaultClientTelemetryPayload.java | 5 +-
.../kafka/server/ClientMetricsManagerTest.java | 204 ++++++++++++++-------
.../metrics/ClientMetricsTelemetryPluginTest.java | 19 +-
.../kafka/server/metrics/ClientTelemetryTest.java | 12 +-
.../server/metrics/ClientMetricsTestUtils.java | 18 ++
37 files changed, 466 insertions(+), 277 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java
index 349a5aeb95a..57078885d1a 100644
--- a/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java
+++ b/clients/src/main/java/org/apache/kafka/clients/CommonClientConfigs.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.clients;
+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;
@@ -341,12 +342,12 @@ public class CommonClientConfigs {
MetricsReporter.class, clientIdOverride);
}
- public static Optional<ClientTelemetryReporter> telemetryReporter(String clientId, AbstractConfig config) {
+ public static Optional<ClientTelemetryReporter> telemetryReporter(String clientId, Uuid clientInstanceId, AbstractConfig config) {
if (!config.getBoolean(CommonClientConfigs.ENABLE_METRICS_PUSH_CONFIG)) {
return Optional.empty();
}
- ClientTelemetryReporter telemetryReporter = new ClientTelemetryReporter(Time.SYSTEM);
+ ClientTelemetryReporter telemetryReporter = new ClientTelemetryReporter(Time.SYSTEM, clientInstanceId);
telemetryReporter.configure(config.originals(Collections.singletonMap(CommonClientConfigs.CLIENT_ID_CONFIG, clientId)));
return Optional.of(telemetryReporter);
}
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java b/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
index b2bff248900..728f47d332f 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
@@ -1896,8 +1896,8 @@ public interface Admin extends AutoCloseable {
* The ID is useful for correlating client operations with telemetry sent to the broker and
* to its eventual monitoring destinations.
* <p>
- * If telemetry is enabled, this will first require a connection to the cluster to generate
- * the unique client instance ID. This method waits up to {@code timeout} for the admin
+ * If telemetry is enabled, this will first require a connection to the cluster to receive
+ * a telemetry subscription. This method waits up to {@code timeout} for the admin
* client to complete the request.
* <p>
* Client telemetry is controlled by the {@link AdminClientConfig#ENABLE_METRICS_PUSH_CONFIG}
@@ -1913,7 +1913,7 @@ public interface Admin extends AutoCloseable {
* @throws IllegalArgumentException If the {@code timeout} is negative.
* @throws IllegalStateException If telemetry is not enabled ie, config `{@code enable.metrics.push}`
* is set to `{@code false}`.
- * @return The client's assigned instance id used for metrics collection.
+ * @return The client's instance id used for metrics collection.
*/
Uuid clientInstanceId(Duration timeout);
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 9253de92d78..6a2e3032188 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
@@ -603,7 +603,8 @@ public class KafkaAdminClient extends AdminClient {
// Otherwise, let NetworkClient.ensureBootstrapped() handle it during the first poll
// after DNS resolution succeeds.
List<MetricsReporter> reporters = CommonClientConfigs.metricsReporters(clientId, config);
- clientTelemetryReporter = CommonClientConfigs.telemetryReporter(clientId, config);
+ Uuid clientInstanceId = Uuid.randomUuid();
+ clientTelemetryReporter = CommonClientConfigs.telemetryReporter(clientId, clientInstanceId, config);
clientTelemetryReporter.ifPresent(reporters::add);
Map<String, String> metricTags = Collections.singletonMap("client-id", clientId);
MetricConfig metricConfig = new MetricConfig().samples(config.getInt(AdminClientConfig.METRICS_NUM_SAMPLES_CONFIG))
@@ -614,7 +615,6 @@ 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,
@@ -644,11 +644,12 @@ public class KafkaAdminClient extends AdminClient {
static KafkaAdminClient createInternal(AdminClientConfig config,
AdminMetadataManager metadataManager,
KafkaClient client,
+ Uuid clientInstanceId,
Time time) {
Metrics metrics = null;
String clientId = generateClientId(config);
List<MetricsReporter> reporters = CommonClientConfigs.metricsReporters(clientId, config);
- Optional<ClientTelemetryReporter> clientTelemetryReporter = CommonClientConfigs.telemetryReporter(clientId, config);
+ Optional<ClientTelemetryReporter> clientTelemetryReporter = CommonClientConfigs.telemetryReporter(clientId, clientInstanceId, config);
clientTelemetryReporter.ifPresent(reporters::add);
try {
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
index 299936665e3..fb0c7ccd662 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java
@@ -1476,8 +1476,8 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
* The ID is useful for correlating client operations with telemetry sent to the broker and
* to its eventual monitoring destinations.
* <p>
- * If telemetry is enabled, this will first require a connection to the cluster to generate
- * the unique client instance ID. This method waits up to {@code timeout} for the consumer
+ * If telemetry is enabled, this will first require a connection to the cluster to receive
+ * a telemetry subscription. This method waits up to {@code timeout} for the consumer
* client to complete the request.
* <p>
* Client telemetry is controlled by the {@link ConsumerConfig#ENABLE_METRICS_PUSH_CONFIG}
@@ -1493,7 +1493,7 @@ public class KafkaConsumer<K, V> implements Consumer<K, V> {
* @throws IllegalArgumentException If the {@code timeout} is negative.
* @throws IllegalStateException If telemetry is not enabled ie, config `{@code enable.metrics.push}`
* is set to `{@code false}`.
- * @return The client's assigned instance id used for metrics collection.
+ * @return The client's instance id used for metrics collection.
*/
@Override
public Uuid clientInstanceId(Duration timeout) {
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaShareConsumer.java b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaShareConsumer.java
index 42717e9f731..c390c4be4a3 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaShareConsumer.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaShareConsumer.java
@@ -702,8 +702,8 @@ public class KafkaShareConsumer<K, V> implements ShareConsumer<K, V> {
* The ID is useful for correlating client operations with telemetry sent to the broker and
* to its eventual monitoring destinations.
* <p>
- * If telemetry is enabled, this will first require a connection to the cluster to generate
- * the unique client instance ID. This method waits up to {@code timeout} for the consumer
+ * If telemetry is enabled, this will first require a connection to the cluster to receive
+ * a telemetry subscription. This method waits up to {@code timeout} for the consumer
* client to complete the request.
* <p>
* Client telemetry is controlled by the {@link ConsumerConfig#ENABLE_METRICS_PUSH_CONFIG}
@@ -713,7 +713,7 @@ public class KafkaShareConsumer<K, V> implements ShareConsumer<K, V> {
* The value must be non-negative. Specifying a timeout of zero means do not
* wait for the initial request to complete if it hasn't already.
*
- * @return The client's assigned instance id used for metrics collection.
+ * @return The client's instance id used for metrics collection.
*
* @throws IllegalArgumentException if the {@code timeout} is negative
* @throws IllegalStateException if telemetry is not enabled
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 27993538ca4..42efe867599 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
@@ -484,7 +484,8 @@ public class AsyncKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
this.defaultApiTimeoutMs = Duration.ofMillis(config.getInt(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG));
this.time = time;
List<MetricsReporter> reporters = CommonClientConfigs.metricsReporters(clientId, config);
- this.clientTelemetryReporter = CommonClientConfigs.telemetryReporter(clientId, config);
+ Uuid clientInstanceId = Uuid.randomUuid();
+ this.clientTelemetryReporter = CommonClientConfigs.telemetryReporter(clientId, clientInstanceId, config);
this.clientTelemetryReporter.ifPresent(reporters::add);
this.metrics = createMetrics(config, time, reporters);
this.asyncConsumerMetrics = new AsyncConsumerMetrics(metrics, CONSUMER_METRIC_GROUP);
@@ -518,7 +519,6 @@ 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,
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 0339a7e5886..cf9556b7724 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
@@ -175,7 +175,8 @@ public class ClassicKafkaConsumer<K, V> implements ConsumerDelegate<K, V> {
this.defaultApiTimeoutMs = config.getInt(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG);
this.time = Time.SYSTEM;
List<MetricsReporter> reporters = CommonClientConfigs.metricsReporters(clientId, config);
- this.clientTelemetryReporter = CommonClientConfigs.telemetryReporter(clientId, config);
+ Uuid clientInstanceId = Uuid.randomUuid();
+ this.clientTelemetryReporter = CommonClientConfigs.telemetryReporter(clientId, clientInstanceId, config);
this.clientTelemetryReporter.ifPresent(reporters::add);
this.metrics = createMetrics(config, time, reporters);
this.retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
@@ -198,7 +199,6 @@ 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,
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 e4bb6339d15..acc222f7c64 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
@@ -259,7 +259,8 @@ public class ShareConsumerImpl<K, V> implements ShareConsumerDelegate<K, V> {
this.defaultApiTimeoutMs = config.getInt(ConsumerConfig.DEFAULT_API_TIMEOUT_MS_CONFIG);
this.time = time;
List<MetricsReporter> reporters = CommonClientConfigs.metricsReporters(clientId, config);
- this.clientTelemetryReporter = CommonClientConfigs.telemetryReporter(clientId, config);
+ Uuid clientInstanceId = Uuid.randomUuid();
+ this.clientTelemetryReporter = CommonClientConfigs.telemetryReporter(clientId, clientInstanceId, config);
this.clientTelemetryReporter.ifPresent(reporters::add);
this.metrics = createMetrics(config, time, reporters);
this.asyncConsumerMetrics = new AsyncConsumerMetrics(metrics, CONSUMER_SHARE_METRIC_GROUP);
@@ -284,7 +285,6 @@ 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,
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 895fd8731c1..3cd1a808bbf 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
@@ -321,7 +321,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
*/
public KafkaProducer(Map<String, Object> configs, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
this(new ProducerConfig(ProducerConfig.appendSerializerToConfig(configs, keySerializer, valueSerializer)),
- keySerializer, valueSerializer, null, null, null, new ApiVersions(), Time.SYSTEM);
+ keySerializer, valueSerializer);
}
/**
@@ -350,6 +350,10 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
this(Utils.propsToMap(properties), keySerializer, valueSerializer);
}
+ KafkaProducer(ProducerConfig config, Serializer<K> keySerializer, Serializer<V> valueSerializer) {
+ this(config, keySerializer, valueSerializer, null, null, Uuid.randomUuid(), null, new ApiVersions(), Time.SYSTEM);
+ }
+
// visible for testing
@SuppressWarnings({"unchecked", "this-escape"})
KafkaProducer(ProducerConfig config,
@@ -357,6 +361,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
Serializer<V> valueSerializer,
ProducerMetadata metadata,
KafkaClient kafkaClient,
+ Uuid clientInstanceId,
ProducerInterceptors<K, V> interceptors,
ApiVersions apiVersions,
Time time) {
@@ -382,7 +387,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
.recordLevel(Sensor.RecordingLevel.forName(config.getString(ProducerConfig.METRICS_RECORDING_LEVEL_CONFIG)))
.tags(metricTags);
List<MetricsReporter> reporters = CommonClientConfigs.metricsReporters(clientId, config);
- this.clientTelemetryReporter = CommonClientConfigs.telemetryReporter(clientId, config);
+ this.clientTelemetryReporter = CommonClientConfigs.telemetryReporter(clientId, clientInstanceId, config);
this.clientTelemetryReporter.ifPresent(reporters::add);
MetricsContext metricsContext = new KafkaMetricsContext(JMX_PREFIX,
config.originalsWithPrefix(CommonClientConfigs.METRICS_CONTEXT_PREFIX));
@@ -512,7 +517,6 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
}
this.errors = this.metrics.sensor("errors");
- 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);
@@ -1525,8 +1529,8 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
* The ID is useful for correlating client operations with telemetry sent to the broker and
* to its eventual monitoring destinations.
* <p>
- * If telemetry is enabled, this will first require a connection to the cluster to generate
- * the unique client instance ID. This method waits up to {@code timeout} for the producer
+ * If telemetry is enabled, this will first require a connection to the cluster to receive
+ * a telemetry subscription. This method waits up to {@code timeout} for the producer
* client to complete the request.
* <p>
* Client telemetry is controlled by the {@link ProducerConfig#ENABLE_METRICS_PUSH_CONFIG}
@@ -1542,7 +1546,7 @@ public class KafkaProducer<K, V> implements Producer<K, V> {
* @throws IllegalArgumentException If the {@code timeout} is negative.
* @throws IllegalStateException If telemetry is not enabled ie, config `{@code enable.metrics.push}`
* is set to `{@code false}`.
- * @return The client's assigned instance id used for metrics collection.
+ * @return The client's instance id used for metrics collection.
*/
@Override
public Uuid clientInstanceId(Duration timeout) {
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/GetTelemetrySubscriptionsRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/GetTelemetrySubscriptionsRequest.java
index c507d930e24..699bbf89aaf 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/GetTelemetrySubscriptionsRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/GetTelemetrySubscriptionsRequest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.common.requests;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.message.GetTelemetrySubscriptionsRequestData;
import org.apache.kafka.common.message.GetTelemetrySubscriptionsResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
@@ -40,6 +41,11 @@ public class GetTelemetrySubscriptionsRequest extends AbstractRequest {
@Override
public GetTelemetrySubscriptionsRequest build(short version) {
+ // From v1 the client instance ID is carried by the request header (KIP-1313) and the body
+ // field only exists in v0, so a non-zero value must not reach the writer.
+ if (version >= 1 && !Uuid.ZERO_UUID.equals(data.clientInstanceId())) {
+ return new GetTelemetrySubscriptionsRequest(data.duplicate().setClientInstanceId(Uuid.ZERO_UUID), version);
+ }
return new GetTelemetrySubscriptionsRequest(data, version);
}
diff --git a/clients/src/main/java/org/apache/kafka/common/requests/PushTelemetryRequest.java b/clients/src/main/java/org/apache/kafka/common/requests/PushTelemetryRequest.java
index b3502677dc5..32b8e09a2bf 100644
--- a/clients/src/main/java/org/apache/kafka/common/requests/PushTelemetryRequest.java
+++ b/clients/src/main/java/org/apache/kafka/common/requests/PushTelemetryRequest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.common.requests;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.message.PushTelemetryRequestData;
import org.apache.kafka.common.message.PushTelemetryResponseData;
import org.apache.kafka.common.protocol.ApiKeys;
@@ -45,6 +46,11 @@ public class PushTelemetryRequest extends AbstractRequest {
@Override
public PushTelemetryRequest build(short version) {
+ // From v1 the client instance ID is carried by the request header (KIP-1313) and the body
+ // field only exists in v0, so a non-zero value must not reach the writer.
+ if (version >= 1 && !Uuid.ZERO_UUID.equals(data.clientInstanceId())) {
+ return new PushTelemetryRequest(data.duplicate().setClientInstanceId(Uuid.ZERO_UUID), version);
+ }
return new PushTelemetryRequest(data, version);
}
diff --git a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java
index 62f041b7500..0b1c390bc31 100644
--- a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java
+++ b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java
@@ -118,12 +118,15 @@ public class ClientTelemetryReporter implements MetricsReporter {
private final ClientTelemetryProvider telemetryProvider;
private final ClientTelemetrySender clientTelemetrySender;
private final Time time;
+ // Generated by the owning client. Sent in the v0 request body or the v1 request header (KIP-1313).
+ private final Uuid clientInstanceId;
private Map<String, Object> rawOriginalConfig;
private KafkaMetricsCollector kafkaMetricsCollector;
- public ClientTelemetryReporter(Time time) {
+ public ClientTelemetryReporter(Time time, Uuid clientInstanceId) {
this.time = time;
+ this.clientInstanceId = ClientTelemetryUtils.validateClientInstanceId(clientInstanceId);
telemetryProvider = new ClientTelemetryProvider();
clientTelemetrySender = new DefaultClientTelemetrySender();
}
@@ -384,7 +387,7 @@ public class ClientTelemetryReporter implements MetricsReporter {
}
if (localState == ClientTelemetryState.SUBSCRIPTION_NEEDED) {
- return createSubscriptionRequest(localSubscription);
+ return createSubscriptionRequest();
} else if (localState == ClientTelemetryState.PUSH_NEEDED || localState == ClientTelemetryState.TERMINATING_PUSH_NEEDED) {
return createPushRequest(localSubscription);
}
@@ -426,7 +429,6 @@ public class ClientTelemetryReporter implements MetricsReporter {
return;
}
- Uuid clientInstanceId = ClientTelemetryUtils.validateClientInstanceId(data.clientInstanceId());
int intervalMs = ClientTelemetryUtils.validateIntervalMs(data.pushIntervalMs());
Predicate<? super MetricKeyable> selector = ClientTelemetryUtils.getSelectorFromRequestedMetrics(
data.requestedMetrics());
@@ -446,7 +448,6 @@ public class ClientTelemetryReporter implements MetricsReporter {
}
ClientTelemetrySubscription clientTelemetrySubscription = new ClientTelemetrySubscription(
- clientInstanceId,
data.subscriptionId(),
intervalMs,
acceptedCompressionTypes,
@@ -480,7 +481,7 @@ public class ClientTelemetryReporter implements MetricsReporter {
}
updateSubscriptionResult(clientTelemetrySubscription, nowMs);
- log.info("Client telemetry registered with client instance id: {}", subscription.clientInstanceId());
+ log.info("Client telemetry registered with client instance id: {}", clientInstanceId);
} finally {
lock.writeLock().unlock();
}
@@ -564,12 +565,6 @@ public class ClientTelemetryReporter implements MetricsReporter {
return Optional.empty();
}
- Uuid clientInstanceId = subscription.clientInstanceId();
- if (clientInstanceId == null) {
- log.info("Client instance ID was null in telemetry subscription while in state {}", state);
- return Optional.empty();
- }
-
return Optional.of(clientInstanceId);
} finally {
lock.writeLock().unlock();
@@ -635,13 +630,12 @@ public class ClientTelemetryReporter implements MetricsReporter {
(maybeFatalException.getCause() != null && maybeFatalException.getCause() instanceof RetriableException);
}
- private Optional<Builder<?>> createSubscriptionRequest(ClientTelemetrySubscription localSubscription) {
+ private Optional<Builder<?>> createSubscriptionRequest() {
/*
- If we've previously retrieved a subscription, it will contain the client instance ID
- that the broker assigned. Otherwise, per KIP-714, we send a special "null" UUID to
- signal to the broker that we need to have a client instance ID assigned.
+ The client generates its own client instance ID (KIP-1313). In v0 it is sent in the request
+ body, where the broker adopts it instead of assigning one; in v1 it travels in the request
+ header and the request Builder clears it from the body.
*/
- Uuid clientInstanceId = (localSubscription != null) ? localSubscription.clientInstanceId() : Uuid.ZERO_UUID;
log.debug("Creating telemetry subscription request with client instance id {}", clientInstanceId);
lock.writeLock().lock();
@@ -671,7 +665,7 @@ public class ClientTelemetryReporter implements MetricsReporter {
return Optional.empty();
}
- log.debug("Creating telemetry push request with client instance id {}", localSubscription.clientInstanceId());
+ log.debug("Creating telemetry push request with client instance id {}", clientInstanceId);
/*
Don't send a push request if we don't have the collector initialized. Re-attempt
the push on the next interval.
@@ -758,7 +752,7 @@ public class ClientTelemetryReporter implements MetricsReporter {
AbstractRequest.Builder<?> requestBuilder = new PushTelemetryRequest.Builder(
new PushTelemetryRequestData()
- .setClientInstanceId(localSubscription.clientInstanceId())
+ .setClientInstanceId(clientInstanceId)
.setSubscriptionId(localSubscription.subscriptionId())
.setTerminating(terminating)
.setCompressionType(compressionType.id)
@@ -972,17 +966,15 @@ public class ClientTelemetryReporter implements MetricsReporter {
*/
static class ClientTelemetrySubscription {
- private final Uuid clientInstanceId;
private final int subscriptionId;
private final int pushIntervalMs;
private final List<CompressionType> acceptedCompressionTypes;
private final boolean deltaTemporality;
private final Predicate<? super MetricKeyable> selector;
- ClientTelemetrySubscription(Uuid clientInstanceId, int subscriptionId, int pushIntervalMs,
+ ClientTelemetrySubscription(int subscriptionId, int pushIntervalMs,
List<CompressionType> acceptedCompressionTypes, boolean deltaTemporality,
Predicate<? super MetricKeyable> selector) {
- this.clientInstanceId = clientInstanceId;
this.subscriptionId = subscriptionId;
this.pushIntervalMs = pushIntervalMs;
this.acceptedCompressionTypes = List.copyOf(acceptedCompressionTypes);
@@ -990,10 +982,6 @@ public class ClientTelemetryReporter implements MetricsReporter {
this.selector = selector;
}
- public Uuid clientInstanceId() {
- return clientInstanceId;
- }
-
public int subscriptionId() {
return subscriptionId;
}
@@ -1017,7 +1005,6 @@ public class ClientTelemetryReporter implements MetricsReporter {
@Override
public String toString() {
return new StringJoiner(", ", "ClientTelemetrySubscription{", "}")
- .add("clientInstanceId=" + clientInstanceId)
.add("subscriptionId=" + subscriptionId)
.add("pushIntervalMs=" + pushIntervalMs)
.add("acceptedCompressionTypes=" + acceptedCompressionTypes)
diff --git a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetrySender.java b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetrySender.java
index d7fd81364fb..0d9c4d6b27a 100644
--- a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetrySender.java
+++ b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetrySender.java
@@ -99,7 +99,7 @@ public interface ClientTelemetrySender extends AutoCloseable {
* enclosing client instance is otherwise unusable.
* @throws IllegalArgumentException If the <code>timeout</code> is negative.
*
- * @return If present, optional of the client's assigned instance id used for metrics collection.
+ * @return If present, optional of the client's instance id used for metrics collection.
*/
Optional<Uuid> clientInstanceId(Duration timeout);
diff --git a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtils.java b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtils.java
index 9d723bd3ffd..00a2cac2ba1 100644
--- a/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtils.java
+++ b/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtils.java
@@ -142,7 +142,7 @@ public class ClientTelemetryUtils {
}
public static Uuid validateClientInstanceId(Uuid clientInstanceId) {
- if (clientInstanceId == null || clientInstanceId.equals(Uuid.ZERO_UUID)) {
+ if (clientInstanceId == null || Uuid.RESERVED.contains(clientInstanceId)) {
throw new IllegalArgumentException("clientInstanceId is not valid");
}
diff --git a/clients/src/main/resources/common/message/GetTelemetrySubscriptionsRequest.json b/clients/src/main/resources/common/message/GetTelemetrySubscriptionsRequest.json
index a56d5099191..bfce6baf9d2 100644
--- a/clients/src/main/resources/common/message/GetTelemetrySubscriptionsRequest.json
+++ b/clients/src/main/resources/common/message/GetTelemetrySubscriptionsRequest.json
@@ -18,15 +18,16 @@
"type": "request",
"listeners": ["broker"],
"name": "GetTelemetrySubscriptionsRequest",
- "validVersions": "0",
+ // Version 1 removes ClientInstanceId and uses the v3 request header (KIP-1313).
+ "latestVersionUnstable": true,
+ "validVersions": "0-1",
"flexibleVersions": "0+",
"headerVersions": {
- "0+": "2"
+ "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."
- }
+ { "name": "ClientInstanceId", "type": "uuid", "versions": "0",
+ "about": "Unique id for this client instance, must be set to 0 on the first request." }
]
-}
\ No newline at end of file
+}
diff --git a/clients/src/main/resources/common/message/GetTelemetrySubscriptionsResponse.json b/clients/src/main/resources/common/message/GetTelemetrySubscriptionsResponse.json
index 32b9ba07908..4512330a765 100644
--- a/clients/src/main/resources/common/message/GetTelemetrySubscriptionsResponse.json
+++ b/clients/src/main/resources/common/message/GetTelemetrySubscriptionsResponse.json
@@ -17,47 +17,30 @@
"apiKey": 71,
"type": "response",
"name": "GetTelemetrySubscriptionsResponse",
- "validVersions": "0",
+ // Version 1 removes ClientInstanceId (KIP-1313).
+ "validVersions": "0-1",
"flexibleVersions": "0+",
"headerVersions": {
"0+": "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."
- }
+ { "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." }
]
-}
\ No newline at end of file
+}
diff --git a/clients/src/main/resources/common/message/PushTelemetryRequest.json b/clients/src/main/resources/common/message/PushTelemetryRequest.json
index a9c4c9c64fb..63d33e5adfe 100644
--- a/clients/src/main/resources/common/message/PushTelemetryRequest.json
+++ b/clients/src/main/resources/common/message/PushTelemetryRequest.json
@@ -18,31 +18,24 @@
"type": "request",
"listeners": ["broker"],
"name": "PushTelemetryRequest",
- "validVersions": "0",
+ // Version 1 removes ClientInstanceId and uses the v3 request header (KIP-1313).
+ "latestVersionUnstable": true,
+ "validVersions": "0-1",
"flexibleVersions": "0+",
"headerVersions": {
- "0+": "2"
+ "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."
- }
+ { "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." }
]
-}
\ No newline at end of file
+}
diff --git a/clients/src/main/resources/common/message/PushTelemetryResponse.json b/clients/src/main/resources/common/message/PushTelemetryResponse.json
index ee30bf15faf..3680b1c3580 100644
--- a/clients/src/main/resources/common/message/PushTelemetryResponse.json
+++ b/clients/src/main/resources/common/message/PushTelemetryResponse.json
@@ -17,19 +17,16 @@
"apiKey": 72,
"type": "response",
"name": "PushTelemetryResponse",
- "validVersions": "0",
+ // Version 1 is the same as version 0.
+ "validVersions": "0-1",
"flexibleVersions": "0+",
"headerVersions": {
"0+": "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": "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." }
]
-}
\ No newline at end of file
+}
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 7cccdac0651..695bf838699 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
@@ -1127,7 +1127,7 @@ public class KafkaAdminClientTest extends KafkaAdminClientTestBase {
try (MockedStatic<CommonClientConfigs> mockedCommonClientConfigs = mockStatic(CommonClientConfigs.class, new CallsRealMethods())) {
ClientTelemetryReporter clientTelemetryReporter = mock(ClientTelemetryReporter.class);
clientTelemetryReporter.configure(any());
- mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any())).thenReturn(Optional.of(clientTelemetryReporter));
+ mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any(), any())).thenReturn(Optional.of(clientTelemetryReporter));
try (AdminClientUnitTestEnv env = mockClientEnv(AdminClientConfig.ENABLE_METRICS_PUSH_CONFIG, "true")) {
ClientTelemetrySender clientTelemetrySender = mock(ClientTelemetrySender.class);
@@ -1334,7 +1334,7 @@ public class KafkaAdminClientTest extends KafkaAdminClientTestBase {
mockClient.setNodeApiVersions(NodeApiVersions.create());
try (KafkaAdminClient admin = KafkaAdminClient.createInternal(
- new AdminClientConfig(Collections.emptyMap()), metadataManager, mockClient, Time.SYSTEM)) {
+ new AdminClientConfig(Collections.emptyMap()), metadataManager, mockClient, mockClient.clientInstanceId(), Time.SYSTEM)) {
DescribeClusterResult result = admin.describeCluster(new DescribeClusterOptions());
// make sure maybeDrainPendingCalls doesn't remove duplicate pending calls
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 f14ccec1731..a9f1366e261 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
@@ -338,7 +338,7 @@ public class KafkaConsumerTest {
try (MockedStatic<CommonClientConfigs> mockedCommonClientConfigs = mockStatic(CommonClientConfigs.class, new CallsRealMethods())) {
ClientTelemetryReporter clientTelemetryReporter = mock(ClientTelemetryReporter.class);
clientTelemetryReporter.configure(any());
- mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any())).thenReturn(Optional.of(clientTelemetryReporter));
+ mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any(), any())).thenReturn(Optional.of(clientTelemetryReporter));
Properties props = new Properties();
props.setProperty(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name());
@@ -358,7 +358,7 @@ public class KafkaConsumerTest {
try (MockedStatic<CommonClientConfigs> mockedCommonClientConfigs = mockStatic(CommonClientConfigs.class, new CallsRealMethods())) {
ClientTelemetryReporter clientTelemetryReporter = mock(ClientTelemetryReporter.class);
clientTelemetryReporter.configure(any());
- mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any())).thenReturn(Optional.of(clientTelemetryReporter));
+ mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any(), any())).thenReturn(Optional.of(clientTelemetryReporter));
Properties props = new Properties();
props.setProperty(ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name());
@@ -4039,7 +4039,7 @@ public void testPollIdleRatio(GroupProtocol groupProtocol) {
clientTelemetryReporter.configure(any());
try (MockedStatic<CommonClientConfigs> mockedCommonClientConfigs = mockStatic(CommonClientConfigs.class, new CallsRealMethods())) {
- mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any())).thenReturn(Optional.of(clientTelemetryReporter));
+ mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any(), any())).thenReturn(Optional.of(clientTelemetryReporter));
ClientTelemetrySender clientTelemetrySender = mock(ClientTelemetrySender.class);
Uuid expectedUuid = Uuid.randomUuid();
diff --git a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerMetricsTest.java b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerMetricsTest.java
index 84044d5fafc..366fca6d432 100644
--- a/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerMetricsTest.java
+++ b/clients/src/test/java/org/apache/kafka/clients/consumer/KafkaShareConsumerMetricsTest.java
@@ -261,7 +261,7 @@ public class KafkaShareConsumerMetricsTest {
try (MockedStatic<CommonClientConfigs> mockedCommonClientConfigs = mockStatic(CommonClientConfigs.class, new CallsRealMethods())) {
ClientTelemetryReporter clientTelemetryReporter = mock(ClientTelemetryReporter.class);
clientTelemetryReporter.configure(any());
- mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any())).thenReturn(Optional.of(clientTelemetryReporter));
+ mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any(), any())).thenReturn(Optional.of(clientTelemetryReporter));
Time time = new MockTime(1L);
ShareConsumerMetadata metadata = createMetadata(subscription);
@@ -282,7 +282,7 @@ public class KafkaShareConsumerMetricsTest {
try (MockedStatic<CommonClientConfigs> mockedCommonClientConfigs = mockStatic(CommonClientConfigs.class, new CallsRealMethods())) {
ClientTelemetryReporter clientTelemetryReporter = mock(ClientTelemetryReporter.class);
clientTelemetryReporter.configure(any());
- mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any())).thenReturn(Optional.of(clientTelemetryReporter));
+ mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any(), any())).thenReturn(Optional.of(clientTelemetryReporter));
Time time = new MockTime(1L);
ShareConsumerMetadata metadata = createMetadata(subscription);
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 68904ee798d..283464dbb91 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
@@ -216,11 +216,12 @@ public class KafkaProducerTest {
Serializer<K> keySerializer,
Serializer<V> valueSerializer,
ProducerMetadata metadata,
- KafkaClient kafkaClient,
+ MockClient kafkaClient,
ProducerInterceptors<K, V> interceptors,
Time time) {
+ Uuid clientInstanceId = kafkaClient != null ? kafkaClient.clientInstanceId() : Uuid.randomUuid();
return new KafkaProducer<>(new ProducerConfig(ProducerConfig.appendSerializerToConfig(configs, keySerializer, valueSerializer)),
- keySerializer, valueSerializer, metadata, kafkaClient, interceptors, new ApiVersions(), time);
+ keySerializer, valueSerializer, metadata, kafkaClient, clientInstanceId, interceptors, new ApiVersions(), time);
}
@BeforeEach
@@ -788,7 +789,7 @@ public class KafkaProducerTest {
return new KafkaProducer<>(
new ProducerConfig(ProducerConfig.appendSerializerToConfig(configs, new StringSerializer(), new StringSerializer())),
- new StringSerializer(), new StringSerializer(), metadata, mockClient, null, new ApiVersions(), time) {
+ new StringSerializer(), new StringSerializer(), metadata, mockClient, mockClient.clientInstanceId(), null, new ApiVersions(), time) {
@Override
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
@@ -1784,7 +1785,7 @@ public class KafkaProducerTest {
client.prepareResponse(endTxnResponse(Errors.NONE));
try (KafkaProducer<String, String> producer = new KafkaProducer<>(
- config, new StringSerializer(), new StringSerializer(), metadata, client, interceptor, apiVersions, time)
+ config, new StringSerializer(), new StringSerializer(), metadata, client, client.clientInstanceId(), interceptor, apiVersions, time)
) {
producer.initTransactions();
producer.beginTransaction();
@@ -2049,7 +2050,7 @@ public class KafkaProducerTest {
client.prepareResponse(endTxnResponse(Errors.NONE));
try (KafkaProducer<String, String> producer = new KafkaProducer<>(
- new ProducerConfig(properties), new StringSerializer(), new StringSerializer(), metadata, client,
+ new ProducerConfig(properties), new StringSerializer(), new StringSerializer(), metadata, client, client.clientInstanceId(),
new ProducerInterceptors<>(Collections.emptyList(), null), apiVersions, time)) {
producer.initTransactions();
producer.beginTransaction();
@@ -2122,6 +2123,7 @@ public class KafkaProducerTest {
new StringSerializer(),
metadata,
client,
+ client.clientInstanceId(),
new ProducerInterceptors<>(List.of(), null),
apiVersions,
time
@@ -2200,6 +2202,7 @@ public class KafkaProducerTest {
new StringSerializer(),
metadata,
client,
+ client.clientInstanceId(),
new ProducerInterceptors<>(List.of(), null),
apiVersions,
time
@@ -2268,7 +2271,7 @@ public class KafkaProducerTest {
client.prepareResponse(endTxnResponse(Errors.NONE));
try (KafkaProducer<String, String> producer = new KafkaProducer<>(
- config, new StringSerializer(), new StringSerializer(), metadata, client, interceptor, apiVersions, time)
+ config, new StringSerializer(), new StringSerializer(), metadata, client, client.clientInstanceId(), interceptor, apiVersions, time)
) {
producer.initTransactions();
producer.beginTransaction();
@@ -2414,7 +2417,7 @@ public class KafkaProducerTest {
clientTelemetryReporter.configure(any());
try (MockedStatic<CommonClientConfigs> mockedCommonClientConfigs = mockStatic(CommonClientConfigs.class, new CallsRealMethods())) {
- mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any())).thenReturn(Optional.of(clientTelemetryReporter));
+ mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any(), any())).thenReturn(Optional.of(clientTelemetryReporter));
ClientTelemetrySender clientTelemetrySender = mock(ClientTelemetrySender.class);
Uuid expectedUuid = Uuid.randomUuid();
@@ -2767,8 +2770,7 @@ public class KafkaProducerTest {
assertTrue(config.unused().contains(SslConfigs.SSL_PROTOCOL_CONFIG));
- try (KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(config, null, null,
- null, null, null, null, Time.SYSTEM)) {
+ try (KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(config, null, null)) {
assertTrue(config.unused().contains(SslConfigs.SSL_PROTOCOL_CONFIG));
}
}
@@ -3225,7 +3227,7 @@ public class KafkaProducerTest {
try (MockedStatic<CommonClientConfigs> mockedCommonClientConfigs = mockStatic(CommonClientConfigs.class, new CallsRealMethods())) {
ClientTelemetryReporter clientTelemetryReporter = mock(ClientTelemetryReporter.class);
clientTelemetryReporter.configure(any());
- mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any())).thenReturn(Optional.of(clientTelemetryReporter));
+ mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any(), any())).thenReturn(Optional.of(clientTelemetryReporter));
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
@@ -3244,7 +3246,7 @@ public class KafkaProducerTest {
try (MockedStatic<CommonClientConfigs> mockedCommonClientConfigs = mockStatic(CommonClientConfigs.class, new CallsRealMethods())) {
ClientTelemetryReporter clientTelemetryReporter = mock(ClientTelemetryReporter.class);
clientTelemetryReporter.configure(any());
- mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any())).thenReturn(Optional.of(clientTelemetryReporter));
+ mockedCommonClientConfigs.when(() -> CommonClientConfigs.telemetryReporter(anyString(), any(), any())).thenReturn(Optional.of(clientTelemetryReporter));
Map<String, Object> props = new HashMap<>();
props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9999");
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 c78c5a60dca..03afd77151b 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
@@ -101,6 +101,14 @@ public class ApiMessageTypeTest {
assertEquals((short) 3, ApiMessageType.OFFSET_DELETE.requestHeaderVersion((short) 1));
assertEquals((short) 1, ApiMessageType.OFFSET_DELETE.responseHeaderVersion((short) 1));
+ // The telemetry RPCs move the client instance ID from the body into the v3 request header in v1 (KIP-1313).
+ assertEquals((short) 2, ApiMessageType.GET_TELEMETRY_SUBSCRIPTIONS.requestHeaderVersion((short) 0));
+ assertEquals((short) 3, ApiMessageType.GET_TELEMETRY_SUBSCRIPTIONS.requestHeaderVersion((short) 1));
+ assertEquals((short) 1, ApiMessageType.GET_TELEMETRY_SUBSCRIPTIONS.responseHeaderVersion((short) 1));
+ assertEquals((short) 2, ApiMessageType.PUSH_TELEMETRY.requestHeaderVersion((short) 0));
+ assertEquals((short) 3, ApiMessageType.PUSH_TELEMETRY.requestHeaderVersion((short) 1));
+ assertEquals((short) 1, ApiMessageType.PUSH_TELEMETRY.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));
assertEquals((short) 1, ApiMessageType.API_VERSIONS.requestHeaderVersion((short) 2));
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/GetTelemetrySubscriptionsRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/GetTelemetrySubscriptionsRequestTest.java
index d19c6d851e0..fd9c58241af 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/GetTelemetrySubscriptionsRequestTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/GetTelemetrySubscriptionsRequestTest.java
@@ -17,11 +17,15 @@
package org.apache.kafka.common.requests;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.message.GetTelemetrySubscriptionsRequestData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
import org.apache.kafka.common.protocol.Errors;
import org.junit.jupiter.api.Test;
+import java.nio.ByteBuffer;
import java.util.Collections;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -34,4 +38,26 @@ public class GetTelemetrySubscriptionsRequestTest {
GetTelemetrySubscriptionsResponse response = req.getErrorResponse(0, Errors.CLUSTER_AUTHORIZATION_FAILED.exception());
assertEquals(Collections.singletonMap(Errors.CLUSTER_AUTHORIZATION_FAILED, 1), response.errorCounts());
}
+
+ @Test
+ public void testBuildV1SendsClientInstanceIdInHeaderOnly() {
+ Uuid clientInstanceId = Uuid.randomUuid();
+ GetTelemetrySubscriptionsRequestData data = new GetTelemetrySubscriptionsRequestData().setClientInstanceId(clientInstanceId);
+ GetTelemetrySubscriptionsRequest.Builder builder = new GetTelemetrySubscriptionsRequest.Builder(data, true);
+
+ assertEquals(clientInstanceId, builder.build((short) 0).data().clientInstanceId());
+
+ // The data carries the v0 body ID because the version is only chosen when the request is built.
+ GetTelemetrySubscriptionsRequest v1 = builder.build((short) 1);
+ ByteBuffer buffer = v1.serializeWithHeader(
+ new RequestHeader(ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS, (short) 1, "client", clientInstanceId, 0));
+
+ RequestHeader header = RequestHeader.parse(buffer);
+ assertEquals(clientInstanceId, header.clientInstanceId());
+ GetTelemetrySubscriptionsRequest parsed = (GetTelemetrySubscriptionsRequest) AbstractRequest.parseRequest(
+ ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS, (short) 1, new ByteBufferAccessor(buffer)).request;
+ assertEquals(data.duplicate().setClientInstanceId(Uuid.ZERO_UUID), parsed.data());
+ // Building v1 must not mutate the data the builder was given.
+ assertEquals(clientInstanceId, builder.build((short) 0).data().clientInstanceId());
+ }
}
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/PushTelemetryRequestTest.java b/clients/src/test/java/org/apache/kafka/common/requests/PushTelemetryRequestTest.java
index 3816e8ba714..2aa6d62f010 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/PushTelemetryRequestTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/PushTelemetryRequestTest.java
@@ -17,7 +17,10 @@
package org.apache.kafka.common.requests;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.message.PushTelemetryRequestData;
+import org.apache.kafka.common.protocol.ApiKeys;
+import org.apache.kafka.common.protocol.ByteBufferAccessor;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.internal.CompressionType;
import org.apache.kafka.common.telemetry.internals.ClientTelemetryUtils;
@@ -31,6 +34,7 @@ import org.junit.jupiter.params.provider.EnumSource;
import java.io.IOException;
import java.nio.ByteBuffer;
+import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.util.ArrayList;
import java.util.Collections;
@@ -50,6 +54,32 @@ public class PushTelemetryRequestTest {
assertEquals(Collections.singletonMap(Errors.CLUSTER_AUTHORIZATION_FAILED, 1), response.errorCounts());
}
+ @Test
+ public void testBuildV1SendsClientInstanceIdInHeaderOnly() {
+ Uuid clientInstanceId = Uuid.randomUuid();
+ PushTelemetryRequestData data = new PushTelemetryRequestData()
+ .setClientInstanceId(clientInstanceId)
+ .setSubscriptionId(1)
+ .setCompressionType(CompressionType.NONE.id)
+ .setMetrics(ByteBuffer.wrap("test-metrics".getBytes(StandardCharsets.UTF_8)));
+ PushTelemetryRequest.Builder builder = new PushTelemetryRequest.Builder(data, true);
+
+ assertEquals(clientInstanceId, builder.build((short) 0).data().clientInstanceId());
+
+ // The data carries the v0 body ID because the version is only chosen when the request is built.
+ PushTelemetryRequest v1 = builder.build((short) 1);
+ ByteBuffer buffer = v1.serializeWithHeader(
+ new RequestHeader(ApiKeys.PUSH_TELEMETRY, (short) 1, "client", clientInstanceId, 0));
+
+ RequestHeader header = RequestHeader.parse(buffer);
+ assertEquals(clientInstanceId, header.clientInstanceId());
+ PushTelemetryRequest parsed = (PushTelemetryRequest) AbstractRequest.parseRequest(
+ ApiKeys.PUSH_TELEMETRY, (short) 1, new ByteBufferAccessor(buffer)).request;
+ assertEquals(data.duplicate().setClientInstanceId(Uuid.ZERO_UUID), parsed.data());
+ // Building v1 must not mutate the data the builder was given.
+ assertEquals(clientInstanceId, builder.build((short) 0).data().clientInstanceId());
+ }
+
@ParameterizedTest
@EnumSource(CompressionType.class)
public void testMetricsDataCompression(CompressionType compressionType) throws IOException {
diff --git a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
index cfb27d70517..9d76d6ee74a 100644
--- a/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java
@@ -1218,7 +1218,7 @@ public class RequestResponseTest {
case CONSUMER_GROUP_HEARTBEAT: return createConsumerGroupHeartbeatResponse();
case CONSUMER_GROUP_DESCRIBE: return createConsumerGroupDescribeResponse();
case CONTROLLER_REGISTRATION: return createControllerRegistrationResponse();
- case GET_TELEMETRY_SUBSCRIPTIONS: return createGetTelemetrySubscriptionsResponse();
+ case GET_TELEMETRY_SUBSCRIPTIONS: return createGetTelemetrySubscriptionsResponse(version);
case PUSH_TELEMETRY: return createPushTelemetryResponse();
case ASSIGN_REPLICAS_TO_DIRS: return createAssignReplicasToDirsResponse();
case LIST_CONFIG_RESOURCES: return createListConfigResourcesResponse();
@@ -3709,9 +3709,12 @@ public class RequestResponseTest {
).build(version);
}
- private GetTelemetrySubscriptionsResponse createGetTelemetrySubscriptionsResponse() {
+ private GetTelemetrySubscriptionsResponse createGetTelemetrySubscriptionsResponse(short version) {
GetTelemetrySubscriptionsResponseData response = new GetTelemetrySubscriptionsResponseData();
- response.setClientInstanceId(Uuid.randomUuid());
+ // Only the v0 response has the client instance ID field.
+ if (version == 0) {
+ response.setClientInstanceId(Uuid.randomUuid());
+ }
response.setSubscriptionId(1);
response.setAcceptedCompressionTypes(Collections.singletonList(CompressionType.GZIP.id));
response.setPushIntervalMs(60 * 1000);
diff --git a/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporterTest.java b/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporterTest.java
index b75a4915a4e..31d14f9615d 100644
--- a/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporterTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporterTest.java
@@ -86,11 +86,11 @@ public class ClientTelemetryReporterTest {
@BeforeEach
public void setUp() {
time = new MockTime();
- clientTelemetryReporter = new ClientTelemetryReporter(time);
+ uuid = Uuid.randomUuid();
+ clientTelemetryReporter = new ClientTelemetryReporter(time, uuid);
configs = new HashMap<>();
metricsContext = new KafkaMetricsContext("test");
- uuid = Uuid.randomUuid();
- subscription = new ClientTelemetryReporter.ClientTelemetrySubscription(uuid, 1234, 20000,
+ subscription = new ClientTelemetryReporter.ClientTelemetrySubscription(1234, 20000,
Collections.emptyList(), true, null);
}
@@ -227,10 +227,11 @@ public class ClientTelemetryReporterTest {
assertNotNull(requestOptional);
assertTrue(requestOptional.isPresent());
assertInstanceOf(GetTelemetrySubscriptionsRequest.class, requestOptional.get().build());
- GetTelemetrySubscriptionsRequest request = (GetTelemetrySubscriptionsRequest) requestOptional.get().build();
+ // Built at v0, where the client instance ID is part of the body rather than the request header.
+ GetTelemetrySubscriptionsRequest request = (GetTelemetrySubscriptionsRequest) requestOptional.get().build((short) 0);
GetTelemetrySubscriptionsRequest expectedResult = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData().setClientInstanceId(Uuid.ZERO_UUID), true).build();
+ new GetTelemetrySubscriptionsRequestData().setClientInstanceId(uuid), true).build((short) 0);
assertEquals(expectedResult.data(), request.data());
assertEquals(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS, telemetrySender.state());
@@ -246,10 +247,10 @@ public class ClientTelemetryReporterTest {
assertNotNull(requestOptional);
assertTrue(requestOptional.isPresent());
assertInstanceOf(GetTelemetrySubscriptionsRequest.class, requestOptional.get().build());
- GetTelemetrySubscriptionsRequest request = (GetTelemetrySubscriptionsRequest) requestOptional.get().build();
+ GetTelemetrySubscriptionsRequest request = (GetTelemetrySubscriptionsRequest) requestOptional.get().build((short) 0);
GetTelemetrySubscriptionsRequest expectedResult = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData().setClientInstanceId(subscription.clientInstanceId()), true).build();
+ new GetTelemetrySubscriptionsRequestData().setClientInstanceId(uuid), true).build((short) 0);
assertEquals(expectedResult.data(), request.data());
assertEquals(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS, telemetrySender.state());
@@ -270,11 +271,11 @@ public class ClientTelemetryReporterTest {
assertNotNull(requestOptional);
assertTrue(requestOptional.isPresent());
assertInstanceOf(PushTelemetryRequest.class, requestOptional.get().build());
- PushTelemetryRequest request = (PushTelemetryRequest) requestOptional.get().build();
+ PushTelemetryRequest request = (PushTelemetryRequest) requestOptional.get().build((short) 0);
PushTelemetryRequest expectedResult = new PushTelemetryRequest.Builder(
- new PushTelemetryRequestData().setClientInstanceId(subscription.clientInstanceId())
- .setSubscriptionId(subscription.subscriptionId()), true).build();
+ new PushTelemetryRequestData().setClientInstanceId(uuid)
+ .setSubscriptionId(subscription.subscriptionId()), true).build((short) 0);
assertEquals(expectedResult.data(), request.data());
assertEquals(ClientTelemetryState.PUSH_IN_PROGRESS, telemetrySender.state());
@@ -405,16 +406,17 @@ public class ClientTelemetryReporterTest {
assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.PUSH_NEEDED));
ClientTelemetryReporter.ClientTelemetrySubscription subscription = new ClientTelemetryReporter.ClientTelemetrySubscription(
- uuid, 1234, 20000, Collections.singletonList(compressionType), true, null);
+ 1234, 20000, Collections.singletonList(compressionType), true, null);
telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
Optional<AbstractRequest.Builder<?>> requestOptional = telemetrySender.createRequest();
assertNotNull(requestOptional);
assertTrue(requestOptional.isPresent());
assertInstanceOf(PushTelemetryRequest.class, requestOptional.get().build());
- PushTelemetryRequest request = (PushTelemetryRequest) requestOptional.get().build();
+ // Built at v0, where the client instance ID is part of the body rather than the request header.
+ PushTelemetryRequest request = (PushTelemetryRequest) requestOptional.get().build((short) 0);
- assertEquals(subscription.clientInstanceId(), request.data().clientInstanceId());
+ assertEquals(uuid, request.data().clientInstanceId());
assertEquals(subscription.subscriptionId(), request.data().subscriptionId());
assertEquals(compressionType.id, request.data().compressionType());
assertEquals(ClientTelemetryState.PUSH_IN_PROGRESS, telemetrySender.state());
@@ -430,7 +432,7 @@ public class ClientTelemetryReporterTest {
assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.PUSH_NEEDED));
ClientTelemetryReporter.ClientTelemetrySubscription subscription = new ClientTelemetryReporter.ClientTelemetrySubscription(
- uuid, 1234, 20000, Collections.singletonList(CompressionType.GZIP), true, null);
+ 1234, 20000, Collections.singletonList(CompressionType.GZIP), true, null);
telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
Compression.Builder<? extends Compression> failingCompression = compressionFailingWithIOException();
@@ -441,9 +443,10 @@ public class ClientTelemetryReporterTest {
assertNotNull(requestOptional);
assertTrue(requestOptional.isPresent());
assertInstanceOf(PushTelemetryRequest.class, requestOptional.get().build());
- PushTelemetryRequest request = (PushTelemetryRequest) requestOptional.get().build();
+ // Built at v0, where the client instance ID is part of the body rather than the request header.
+ PushTelemetryRequest request = (PushTelemetryRequest) requestOptional.get().build((short) 0);
- assertEquals(subscription.clientInstanceId(), request.data().clientInstanceId());
+ assertEquals(uuid, request.data().clientInstanceId());
assertEquals(subscription.subscriptionId(), request.data().subscriptionId());
// CompressionType.NONE is used when compression fails.
assertEquals(CompressionType.NONE.id, request.data().compressionType());
@@ -462,7 +465,7 @@ public class ClientTelemetryReporterTest {
// Set up subscription with multiple compression types: GZIP -> LZ4 -> SNAPPY
ClientTelemetryReporter.ClientTelemetrySubscription subscription = new ClientTelemetryReporter.ClientTelemetrySubscription(
- uuid, 1234, 20000, List.of(CompressionType.GZIP, CompressionType.LZ4, CompressionType.SNAPPY), true, null);
+ 1234, 20000, List.of(CompressionType.GZIP, CompressionType.LZ4, CompressionType.SNAPPY), true, null);
telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
try (MockedStatic<Compression> mockedCompression = Mockito.mockStatic(Compression.class, new CallsRealMethods())) {
@@ -539,7 +542,7 @@ public class ClientTelemetryReporterTest {
// Set up subscription with ZSTD compression type
ClientTelemetryReporter.ClientTelemetrySubscription subscription = new ClientTelemetryReporter.ClientTelemetrySubscription(
- uuid, 1234, 20000, List.of(CompressionType.ZSTD, CompressionType.LZ4), true, null);
+ 1234, 20000, List.of(CompressionType.ZSTD, CompressionType.LZ4), true, null);
telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
try (MockedStatic<Compression> mockedCompression = Mockito.mockStatic(Compression.class, new CallsRealMethods())) {
@@ -599,10 +602,8 @@ public class ClientTelemetryReporterTest {
ClientTelemetryReporter.DefaultClientTelemetrySender telemetrySender = (ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter.telemetrySender();
assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
- Uuid clientInstanceId = Uuid.randomUuid();
GetTelemetrySubscriptionsResponse response = new GetTelemetrySubscriptionsResponse(
new GetTelemetrySubscriptionsResponseData()
- .setClientInstanceId(clientInstanceId)
.setSubscriptionId(5678)
.setAcceptedCompressionTypes(Collections.singletonList(CompressionType.GZIP.id))
.setPushIntervalMs(20000)
@@ -613,22 +614,27 @@ public class ClientTelemetryReporterTest {
ClientTelemetryReporter.ClientTelemetrySubscription subscription = telemetrySender.subscription();
assertNotNull(subscription);
- assertEquals(clientInstanceId, subscription.clientInstanceId());
+ assertEquals(Optional.of(uuid), telemetrySender.clientInstanceId(Duration.ZERO));
assertEquals(5678, subscription.subscriptionId());
assertEquals(Collections.singletonList(CompressionType.GZIP), subscription.acceptedCompressionTypes());
assertEquals(20000, subscription.pushIntervalMs());
assertEquals(ClientTelemetryUtils.SELECTOR_ALL_METRICS, subscription.selector());
}
+ @Test
+ public void testReporterRejectsInvalidClientInstanceId() {
+ assertThrows(IllegalArgumentException.class, () -> new ClientTelemetryReporter(time, null));
+ assertThrows(IllegalArgumentException.class, () -> new ClientTelemetryReporter(time, Uuid.ZERO_UUID));
+ assertThrows(IllegalArgumentException.class, () -> new ClientTelemetryReporter(time, Uuid.ONE_UUID));
+ }
+
@Test
public void testHandleResponseGetSubscriptionsWithoutMetrics() {
ClientTelemetryReporter.DefaultClientTelemetrySender telemetrySender = (ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter.telemetrySender();
assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
- Uuid clientInstanceId = Uuid.randomUuid();
GetTelemetrySubscriptionsResponse response = new GetTelemetrySubscriptionsResponse(
new GetTelemetrySubscriptionsResponseData()
- .setClientInstanceId(clientInstanceId)
.setSubscriptionId(5678)
.setAcceptedCompressionTypes(Collections.singletonList(CompressionType.GZIP.id))
.setPushIntervalMs(20000));
@@ -639,7 +645,6 @@ public class ClientTelemetryReporterTest {
ClientTelemetryReporter.ClientTelemetrySubscription subscription = telemetrySender.subscription();
assertNotNull(subscription);
- assertEquals(clientInstanceId, subscription.clientInstanceId());
assertEquals(5678, subscription.subscriptionId());
assertEquals(Collections.singletonList(CompressionType.GZIP), subscription.acceptedCompressionTypes());
assertEquals(20000, subscription.pushIntervalMs());
@@ -702,10 +707,8 @@ public class ClientTelemetryReporterTest {
clientTelemetryReporter.metricsCollector(kafkaMetricsCollector);
assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
- Uuid clientInstanceId = Uuid.randomUuid();
GetTelemetrySubscriptionsResponse response = new GetTelemetrySubscriptionsResponse(
new GetTelemetrySubscriptionsResponseData()
- .setClientInstanceId(clientInstanceId)
.setSubscriptionId(15678)
.setAcceptedCompressionTypes(Collections.singletonList(CompressionType.ZSTD.id))
.setPushIntervalMs(10000)
@@ -717,7 +720,6 @@ public class ClientTelemetryReporterTest {
ClientTelemetryReporter.ClientTelemetrySubscription responseSubscription = telemetrySender.subscription();
assertNotNull(responseSubscription);
- assertEquals(clientInstanceId, responseSubscription.clientInstanceId());
assertEquals(15678, responseSubscription.subscriptionId());
assertEquals(Collections.singletonList(CompressionType.ZSTD), responseSubscription.acceptedCompressionTypes());
assertEquals(10000, responseSubscription.pushIntervalMs());
diff --git a/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtilsTest.java b/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtilsTest.java
index 90ec715780a..356ee7f96d3 100644
--- a/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtilsTest.java
+++ b/clients/src/test/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryUtilsTest.java
@@ -98,6 +98,7 @@ public class ClientTelemetryUtilsTest {
public void testValidateClientInstanceId() {
assertThrows(IllegalArgumentException.class, () -> ClientTelemetryUtils.validateClientInstanceId(null));
assertThrows(IllegalArgumentException.class, () -> ClientTelemetryUtils.validateClientInstanceId(Uuid.ZERO_UUID));
+ assertThrows(IllegalArgumentException.class, () -> ClientTelemetryUtils.validateClientInstanceId(Uuid.ONE_UUID));
Uuid uuid = Uuid.randomUuid();
assertEquals(uuid, ClientTelemetryUtils.validateClientInstanceId(uuid));
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 f1cb624458d..082f77f296c 100644
--- a/clients/src/testFixtures/java/org/apache/kafka/clients/MockClient.java
+++ b/clients/src/testFixtures/java/org/apache/kafka/clients/MockClient.java
@@ -589,6 +589,10 @@ public class MockClient implements KafkaClient {
return active;
}
+ public Uuid clientInstanceId() {
+ return clientInstanceId;
+ }
+
@Override
public void close() {
active = false;
diff --git a/clients/src/testFixtures/java/org/apache/kafka/clients/admin/AdminClientUnitTestEnv.java b/clients/src/testFixtures/java/org/apache/kafka/clients/admin/AdminClientUnitTestEnv.java
index c9ec39d4d7d..060d87bb851 100644
--- a/clients/src/testFixtures/java/org/apache/kafka/clients/admin/AdminClientUnitTestEnv.java
+++ b/clients/src/testFixtures/java/org/apache/kafka/clients/admin/AdminClientUnitTestEnv.java
@@ -94,7 +94,7 @@ public class AdminClientUnitTestEnv implements AutoCloseable {
metadataManager.update(cluster, time.milliseconds());
unreachableNodes.forEach(mockClient::setUnreachable);
- this.adminClient = KafkaAdminClient.createInternal(adminClientConfig, metadataManager, mockClient, time);
+ this.adminClient = KafkaAdminClient.createInternal(adminClientConfig, metadataManager, mockClient, mockClient.clientInstanceId(), time);
}
public Time time() {
diff --git a/server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java b/server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java
index 6c0bf390b4b..70ef7067ca0 100644
--- a/server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java
+++ b/server/src/main/java/org/apache/kafka/server/ClientMetricsManager.java
@@ -160,9 +160,18 @@ public class ClientMetricsManager implements AutoCloseable {
GetTelemetrySubscriptionsRequest request, RequestContext requestContext) {
long now = time.milliseconds();
- Uuid clientInstanceId = Optional.ofNullable(request.data().clientInstanceId())
- .filter(id -> !id.equals(Uuid.ZERO_UUID))
- .orElseGet(this::generateNewClientId);
+ Uuid clientInstanceId;
+ if (request.version() >= 1) {
+ // In v1 the ID comes from the request header (KIP-1313); the broker does not generate one.
+ clientInstanceId = requestContext.header.clientInstanceId();
+ if (!isValidClientInstanceId(clientInstanceId)) {
+ return request.getErrorResponse(0, invalidClientInstanceIdException(clientInstanceId));
+ }
+ } else {
+ clientInstanceId = Optional.ofNullable(request.data().clientInstanceId())
+ .filter(id -> !id.equals(Uuid.ZERO_UUID))
+ .orElseGet(this::generateNewClientId);
+ }
/*
Get the client instance from the cache or create a new one. If subscription has changed
@@ -182,16 +191,17 @@ public class ClientMetricsManager implements AutoCloseable {
}
clientInstance.lastKnownError(Errors.NONE);
- return createGetSubscriptionResponse(clientInstanceId, clientInstance);
+ return createGetSubscriptionResponse(clientInstanceId, clientInstance, request.version());
}
public PushTelemetryResponse processPushTelemetryRequest(PushTelemetryRequest request, RequestContext requestContext) {
- Uuid clientInstanceId = request.data().clientInstanceId();
- if (clientInstanceId == null || Uuid.RESERVED.contains(clientInstanceId)) {
- String msg = String.format("Invalid request from the client [%s], invalid client instance id",
- clientInstanceId);
- return request.getErrorResponse(0, new InvalidRequestException(msg));
+ // In v1 the ID comes from the request header (KIP-1313).
+ Uuid clientInstanceId = request.version() >= 1
+ ? requestContext.header.clientInstanceId()
+ : request.data().clientInstanceId();
+ if (!isValidClientInstanceId(clientInstanceId)) {
+ return request.getErrorResponse(0, invalidClientInstanceIdException(clientInstanceId));
}
long now = time.milliseconds();
@@ -216,7 +226,7 @@ public class ClientMetricsManager implements AutoCloseable {
if (metrics != null && metrics.limit() > 0) {
try {
long exportTimeStartMs = time.hiResClockMs();
- clientTelemetryExporterPlugin.exportMetrics(requestContext, request, clientInstance.pushIntervalMs(), clientTelemetryMaxBytes);
+ clientTelemetryExporterPlugin.exportMetrics(requestContext, request, clientInstanceId, clientInstance.pushIntervalMs(), clientTelemetryMaxBytes);
clientMetricsStats.recordPluginExport(clientInstanceId, time.hiResClockMs() - exportTimeStartMs);
} catch (TelemetryTooLargeException exception) {
// The decompressed payload exceeded the configured size limit. This is retryable (the client may
@@ -265,6 +275,15 @@ public class ClientMetricsManager implements AutoCloseable {
subscriptionMap.put(subscriptionName, newSubscription);
}
+ private static boolean isValidClientInstanceId(Uuid clientInstanceId) {
+ return clientInstanceId != null && !Uuid.RESERVED.contains(clientInstanceId);
+ }
+
+ private static InvalidRequestException invalidClientInstanceIdException(Uuid clientInstanceId) {
+ return new InvalidRequestException(String.format(
+ "Invalid request from the client [%s], invalid client instance id", clientInstanceId));
+ }
+
private Uuid generateNewClientId() {
Uuid id = Uuid.randomUuid();
while (clientInstanceCache.get(id) != null) {
@@ -382,10 +401,9 @@ public class ClientMetricsManager implements AutoCloseable {
}
private GetTelemetrySubscriptionsResponse createGetSubscriptionResponse(Uuid clientInstanceId,
- ClientMetricsInstance clientInstance) {
+ ClientMetricsInstance clientInstance, short version) {
GetTelemetrySubscriptionsResponseData data = new GetTelemetrySubscriptionsResponseData()
- .setClientInstanceId(clientInstanceId)
.setSubscriptionId(clientInstance.subscriptionId())
.setRequestedMetrics(new ArrayList<>(clientInstance.metrics()))
.setAcceptedCompressionTypes(SUPPORTED_COMPRESSION_TYPES)
@@ -393,6 +411,11 @@ public class ClientMetricsManager implements AutoCloseable {
.setTelemetryMaxBytes(clientTelemetryMaxBytes)
.setDeltaTemporality(true)
.setErrorCode(Errors.NONE.code());
+ // Only the v0 response has the ID field. It is always set, even when the request supplied the ID,
+ // because a client using v0 of the request may reject a zero ID in the response.
+ if (version == 0) {
+ data.setClientInstanceId(clientInstanceId);
+ }
return new GetTelemetrySubscriptionsResponse(data);
}
@@ -404,7 +427,7 @@ public class ClientMetricsManager implements AutoCloseable {
&& clientInstance.lastKnownError() != Errors.UNSUPPORTED_COMPRESSION_TYPE)) {
clientMetricsStats.recordThrottleCount(clientInstance.clientInstanceId());
String msg = String.format("Request from the client [%s] arrived before the next push interval time",
- request.data().clientInstanceId());
+ clientInstance.clientInstanceId());
throw new ThrottlingQuotaExceededException(msg);
}
}
@@ -414,33 +437,33 @@ public class ClientMetricsManager implements AutoCloseable {
if (clientInstance.terminating()) {
String msg = String.format(
"Client [%s] sent the previous request with state terminating to TRUE, can not accept"
- + "any requests after that", request.data().clientInstanceId());
+ + "any requests after that", clientInstance.clientInstanceId());
throw new InvalidRequestException(msg);
}
if (!clientInstance.maybeUpdatePushRequestTimestamp(timestamp) && !request.data().terminating()) {
clientMetricsStats.recordThrottleCount(clientInstance.clientInstanceId());
String msg = String.format("Request from the client [%s] arrived before the next push interval time",
- request.data().clientInstanceId());
+ clientInstance.clientInstanceId());
throw new ThrottlingQuotaExceededException(msg);
}
if (request.data().subscriptionId() != clientInstance.subscriptionId()) {
clientMetricsStats.recordUnknownSubscriptionCount();
String msg = String.format("Unknown client subscription id for the client [%s]",
- request.data().clientInstanceId());
+ clientInstance.clientInstanceId());
throw new UnknownSubscriptionIdException(msg);
}
if (!isSupportedCompressionType(request.data().compressionType())) {
String msg = String.format("Unknown compression type [%s] is received in telemetry request from [%s]",
- request.data().compressionType(), request.data().clientInstanceId());
+ request.data().compressionType(), clientInstance.clientInstanceId());
throw new UnsupportedCompressionTypeException(msg);
}
if (request.data().metrics() != null && request.data().metrics().limit() > clientTelemetryMaxBytes) {
String msg = String.format("Telemetry request from [%s] is larger than the maximum allowed size [%s]",
- request.data().clientInstanceId(), clientTelemetryMaxBytes);
+ clientInstance.clientInstanceId(), clientTelemetryMaxBytes);
throw new TelemetryTooLargeException(msg);
}
}
diff --git a/server/src/main/java/org/apache/kafka/server/metrics/ClientTelemetryExporterPlugin.java b/server/src/main/java/org/apache/kafka/server/metrics/ClientTelemetryExporterPlugin.java
index 38b0ac5ac47..76f874dacd2 100644
--- a/server/src/main/java/org/apache/kafka/server/metrics/ClientTelemetryExporterPlugin.java
+++ b/server/src/main/java/org/apache/kafka/server/metrics/ClientTelemetryExporterPlugin.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.server.metrics;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.requests.PushTelemetryRequest;
import org.apache.kafka.common.requests.RequestContext;
import org.apache.kafka.server.telemetry.ClientTelemetryExporter;
@@ -53,12 +54,12 @@ public class ClientTelemetryExporterPlugin {
exporters.add(exporter);
}
- public DefaultClientTelemetryPayload getPayLoad(PushTelemetryRequest request, int maxDecompressedBytes) {
- return new DefaultClientTelemetryPayload(request, maxDecompressedBytes);
+ public DefaultClientTelemetryPayload getPayLoad(PushTelemetryRequest request, Uuid clientInstanceId, int maxDecompressedBytes) {
+ return new DefaultClientTelemetryPayload(request, clientInstanceId, maxDecompressedBytes);
}
- public void exportMetrics(RequestContext context, PushTelemetryRequest request, int pushIntervalMs, int maxDecompressedBytes) {
- DefaultClientTelemetryPayload payload = getPayLoad(request, maxDecompressedBytes);
+ public void exportMetrics(RequestContext context, PushTelemetryRequest request, Uuid clientInstanceId, int pushIntervalMs, int maxDecompressedBytes) {
+ DefaultClientTelemetryPayload payload = getPayLoad(request, clientInstanceId, maxDecompressedBytes);
// Export to deprecated receivers
for (ClientTelemetryReceiver receiver : receivers) {
diff --git a/server/src/main/java/org/apache/kafka/server/metrics/DefaultClientTelemetryPayload.java b/server/src/main/java/org/apache/kafka/server/metrics/DefaultClientTelemetryPayload.java
index 0db7fcb6de1..e41f345c37e 100644
--- a/server/src/main/java/org/apache/kafka/server/metrics/DefaultClientTelemetryPayload.java
+++ b/server/src/main/java/org/apache/kafka/server/metrics/DefaultClientTelemetryPayload.java
@@ -32,8 +32,9 @@ public class DefaultClientTelemetryPayload implements ClientTelemetryPayload {
private final String metricsContentType;
private final ByteBuffer metricsData;
- DefaultClientTelemetryPayload(PushTelemetryRequest request, int maxDecompressedBytes) {
- this.clientInstanceId = request.data().clientInstanceId();
+ DefaultClientTelemetryPayload(PushTelemetryRequest request, Uuid clientInstanceId, int maxDecompressedBytes) {
+ // Resolved by the broker: from the request body in v0, from the request header in v1 (KIP-1313).
+ this.clientInstanceId = clientInstanceId;
this.isClientTerminating = request.data().terminating();
this.metricsContentType = request.metricsContentType();
this.metricsData = request.metricsData(maxDecompressedBytes);
diff --git a/server/src/test/java/org/apache/kafka/server/ClientMetricsManagerTest.java b/server/src/test/java/org/apache/kafka/server/ClientMetricsManagerTest.java
index 48aadd82bea..e0110e4050f 100644
--- a/server/src/test/java/org/apache/kafka/server/ClientMetricsManagerTest.java
+++ b/server/src/test/java/org/apache/kafka/server/ClientMetricsManagerTest.java
@@ -24,6 +24,7 @@ import org.apache.kafka.common.message.GetTelemetrySubscriptionsRequestData;
import org.apache.kafka.common.message.PushTelemetryRequestData;
import org.apache.kafka.common.metrics.KafkaMetric;
import org.apache.kafka.common.metrics.Metrics;
+import org.apache.kafka.common.protocol.ApiKeys;
import org.apache.kafka.common.protocol.Errors;
import org.apache.kafka.common.record.internal.CompressionType;
import org.apache.kafka.common.requests.GetTelemetrySubscriptionsRequest;
@@ -196,7 +197,7 @@ public class ClientMetricsManagerTest {
assertEquals(1, clientMetricsManager.subscriptions().size());
GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -239,12 +240,83 @@ public class ClientMetricsManagerTest {
assertEquals(Double.NaN, getMetric(ClientMetricsManager.ClientMetricsStats.PLUGIN_EXPORT_TIME + "-max").metricValue());
}
+ @Test
+ public void testGetTelemetryV1UsesClientInstanceIdFromRequestHeader() throws Exception {
+ clientMetricsManager.updateSubscription("sub-1", ClientMetricsTestUtils.defaultTestProperties());
+ Uuid clientInstanceId = Uuid.randomUuid();
+
+ GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 1);
+ GetTelemetrySubscriptionsResponse response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
+ request, ClientMetricsTestUtils.requestContext(ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS, (short) 1, clientInstanceId));
+
+ assertEquals(Errors.NONE, response.error());
+ // The v1 response carries no client instance ID; the instance is keyed on the header ID (KIP-1313).
+ assertEquals(Uuid.ZERO_UUID, response.data().clientInstanceId());
+ ClientMetricsInstance instance = clientMetricsManager.clientInstance(clientInstanceId);
+ assertNotNull(instance);
+ assertEquals(response.data().subscriptionId(), instance.subscriptionId());
+ assertEquals(Errors.NONE, instance.lastKnownError());
+ }
+
+ @Test
+ public void testGetTelemetryV1RejectsReservedClientInstanceId() throws Exception {
+ for (Uuid clientInstanceId : List.of(Uuid.ZERO_UUID, Uuid.ONE_UUID)) {
+ GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 1);
+ GetTelemetrySubscriptionsResponse response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
+ request, ClientMetricsTestUtils.requestContext(ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS, (short) 1, clientInstanceId));
+
+ assertEquals(Errors.INVALID_REQUEST, response.error());
+ assertNull(clientMetricsManager.clientInstance(clientInstanceId));
+ }
+ }
+
+ @Test
+ public void testPushTelemetryV1UsesClientInstanceIdFromRequestHeader() throws Exception {
+ clientMetricsManager.updateSubscription("sub-1", ClientMetricsTestUtils.defaultTestProperties());
+ ClientMetricsTestUtils.TestClientTelemetryExporter exporter = new ClientMetricsTestUtils.TestClientTelemetryExporter();
+ clientTelemetryExporterPlugin.add(exporter);
+ Uuid clientInstanceId = Uuid.randomUuid();
+
+ GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
+ new GetTelemetrySubscriptionsRequest.Builder(new GetTelemetrySubscriptionsRequestData(), true).build((short) 1),
+ ClientMetricsTestUtils.requestContext(ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS, (short) 1, clientInstanceId));
+ assertEquals(Errors.NONE, subscriptionsResponse.error());
+
+ PushTelemetryRequest request = new PushTelemetryRequest.Builder(
+ new PushTelemetryRequestData()
+ .setSubscriptionId(subscriptionsResponse.data().subscriptionId())
+ .setCompressionType(CompressionType.NONE.id)
+ .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build((short) 1);
+ PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
+ request, ClientMetricsTestUtils.requestContext(ApiKeys.PUSH_TELEMETRY, (short) 1, clientInstanceId));
+
+ assertEquals(Errors.NONE, response.error());
+ assertEquals(Errors.NONE, clientMetricsManager.clientInstance(clientInstanceId).lastKnownError());
+ // The exporter sees the header ID although the v1 body carries none.
+ assertEquals(List.of(clientInstanceId), exporter.clientInstanceIds);
+ }
+
+ @Test
+ public void testPushTelemetryV1RejectsReservedClientInstanceId() throws Exception {
+ for (Uuid clientInstanceId : List.of(Uuid.ZERO_UUID, Uuid.ONE_UUID)) {
+ PushTelemetryRequest request = new PushTelemetryRequest.Builder(
+ new PushTelemetryRequestData().setCompressionType(CompressionType.NONE.id), true).build((short) 1);
+ PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
+ request, ClientMetricsTestUtils.requestContext(ApiKeys.PUSH_TELEMETRY, (short) 1, clientInstanceId));
+
+ assertEquals(Errors.INVALID_REQUEST, response.error());
+ assertNull(clientMetricsManager.clientInstance(clientInstanceId));
+ }
+ }
+
@Test
public void testGetTelemetryWithoutSubscription() throws UnknownHostException {
assertTrue(clientMetricsManager.subscriptions().isEmpty());
GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -266,7 +338,7 @@ public class ClientMetricsManagerTest {
@Test
public void testGetTelemetryAfterPushIntervalTime() throws UnknownHostException {
GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -277,7 +349,7 @@ public class ClientMetricsManagerTest {
time.sleep(ClientMetricsConfigs.INTERVAL_MS_DEFAULT);
request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData().setClientInstanceId(response.data().clientInstanceId()), true).build();
+ new GetTelemetrySubscriptionsRequestData().setClientInstanceId(response.data().clientInstanceId()), true).build((short) 0);
response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -295,7 +367,7 @@ public class ClientMetricsManagerTest {
assertEquals(2, clientMetricsManager.subscriptions().size());
GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -320,7 +392,7 @@ public class ClientMetricsManagerTest {
@Test
public void testGetTelemetrySameClientImmediateRetryFail() throws Exception {
GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -330,7 +402,7 @@ public class ClientMetricsManagerTest {
assertEquals(Errors.NONE, response.error());
request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData().setClientInstanceId(clientInstanceId), true).build();
+ new GetTelemetrySubscriptionsRequestData().setClientInstanceId(clientInstanceId), true).build((short) 0);
response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -344,7 +416,7 @@ public class ClientMetricsManagerTest {
@Test
public void testGetTelemetrySameClientImmediateRetryAfterPushFail() throws Exception {
GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -366,7 +438,7 @@ public class ClientMetricsManagerTest {
.setClientInstanceId(response.data().clientInstanceId())
.setSubscriptionId(response.data().subscriptionId())
.setCompressionType(CompressionType.NONE.id)
- .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build();
+ .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build((short) 0);
PushTelemetryResponse pushResponse = newClientMetricsManager.processPushTelemetryRequest(
pushRequest, ClientMetricsTestUtils.requestContext());
@@ -374,7 +446,7 @@ public class ClientMetricsManagerTest {
assertEquals(Errors.NONE, pushResponse.error());
request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData().setClientInstanceId(clientInstanceId), true).build();
+ new GetTelemetrySubscriptionsRequestData().setClientInstanceId(clientInstanceId), true).build((short) 0);
response = newClientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -392,7 +464,7 @@ public class ClientMetricsManagerTest {
assertEquals(1, clientMetricsManager.subscriptions().size());
GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -410,7 +482,7 @@ public class ClientMetricsManagerTest {
assertEquals(2, clientMetricsManager.subscriptions().size());
request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData().setClientInstanceId(clientInstanceId), true).build();
+ new GetTelemetrySubscriptionsRequestData().setClientInstanceId(clientInstanceId), true).build((short) 0);
response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -424,7 +496,7 @@ public class ClientMetricsManagerTest {
@Test
public void testGetTelemetryConcurrentRequestNewClientInstance() throws Exception {
GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData().setClientInstanceId(Uuid.randomUuid()), true).build();
+ new GetTelemetrySubscriptionsRequestData().setClientInstanceId(Uuid.randomUuid()), true).build((short) 0);
CountDownLatch lock = new CountDownLatch(2);
List<GetTelemetrySubscriptionsResponse> responses = Collections.synchronizedList(new ArrayList<>());
@@ -480,7 +552,7 @@ public class ClientMetricsManagerTest {
@Test
public void testGetTelemetryConcurrentRequestAfterSubscriptionUpdate() throws Exception {
GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData().setClientInstanceId(Uuid.randomUuid()), true).build();
+ new GetTelemetrySubscriptionsRequestData().setClientInstanceId(Uuid.randomUuid()), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -548,7 +620,7 @@ public class ClientMetricsManagerTest {
assertEquals(1, clientMetricsManager.subscriptions().size());
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -561,7 +633,7 @@ public class ClientMetricsManagerTest {
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
.setCompressionType(CompressionType.NONE.id)
- .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build();
+ .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build((short) 0);
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -588,7 +660,7 @@ public class ClientMetricsManagerTest {
@Test
public void testPushTelemetryOnNewServer() throws Exception {
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -605,7 +677,7 @@ public class ClientMetricsManagerTest {
new PushTelemetryRequestData()
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
- .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build();
+ .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build((short) 0);
PushTelemetryResponse response = newClientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -632,7 +704,7 @@ public class ClientMetricsManagerTest {
assertEquals(1, clientMetricsManager.subscriptions().size());
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -642,7 +714,7 @@ public class ClientMetricsManagerTest {
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
.setCompressionType(CompressionType.NONE.id)
- .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build();
+ .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build((short) 0);
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -661,7 +733,7 @@ public class ClientMetricsManagerTest {
public void testPushTelemetryClientInstanceIdInvalid() throws UnknownHostException {
// Null client instance id
PushTelemetryRequest request = new PushTelemetryRequest.Builder(
- new PushTelemetryRequestData().setClientInstanceId(null), true).build();
+ new PushTelemetryRequestData().setClientInstanceId(null), true).build((short) 0);
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -670,7 +742,7 @@ public class ClientMetricsManagerTest {
// Zero client instance id
request = new PushTelemetryRequest.Builder(
- new PushTelemetryRequestData().setClientInstanceId(Uuid.ZERO_UUID), true).build();
+ new PushTelemetryRequestData().setClientInstanceId(Uuid.ZERO_UUID), true).build((short) 0);
response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -681,7 +753,7 @@ public class ClientMetricsManagerTest {
@Test
public void testPushTelemetryThrottleError() throws Exception {
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -690,7 +762,7 @@ public class ClientMetricsManagerTest {
new PushTelemetryRequestData()
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
- .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build();
+ .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build((short) 0);
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -719,7 +791,7 @@ public class ClientMetricsManagerTest {
@Test
public void testPushTelemetryTerminatingFlag() throws Exception {
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -728,7 +800,7 @@ public class ClientMetricsManagerTest {
new PushTelemetryRequestData()
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
- .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build();
+ .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build((short) 0);
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -741,7 +813,7 @@ public class ClientMetricsManagerTest {
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
.setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8)))
- .setTerminating(true), true).build();
+ .setTerminating(true), true).build((short) 0);
response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -763,7 +835,7 @@ public class ClientMetricsManagerTest {
@Test
public void testPushTelemetryNextRequestPostTerminatingFlag() throws UnknownHostException {
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -775,7 +847,7 @@ public class ClientMetricsManagerTest {
new PushTelemetryRequestData()
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
- .setTerminating(true), true).build();
+ .setTerminating(true), true).build((short) 0);
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -788,7 +860,7 @@ public class ClientMetricsManagerTest {
new PushTelemetryRequestData()
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
- .setTerminating(true), true).build();
+ .setTerminating(true), true).build((short) 0);
response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -801,7 +873,7 @@ public class ClientMetricsManagerTest {
@Test
public void testPushTelemetrySubscriptionIdInvalid() throws Exception {
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -813,7 +885,7 @@ public class ClientMetricsManagerTest {
new PushTelemetryRequestData()
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8)))
- .setSubscriptionId(1234), true).build();
+ .setSubscriptionId(1234), true).build((short) 0);
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -832,7 +904,7 @@ public class ClientMetricsManagerTest {
@Test
public void testPushTelemetryCompressionTypeInvalid() throws UnknownHostException {
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -844,7 +916,7 @@ public class ClientMetricsManagerTest {
new PushTelemetryRequestData()
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
- .setCompressionType((byte) 100), true).build();
+ .setCompressionType((byte) 100), true).build((short) 0);
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -857,7 +929,7 @@ public class ClientMetricsManagerTest {
@Test
public void testPushTelemetryNullMetricsData() throws Exception {
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -869,7 +941,7 @@ public class ClientMetricsManagerTest {
new PushTelemetryRequestData()
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
- .setMetrics(null), true).build();
+ .setMetrics(null), true).build((short) 0);
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -894,7 +966,7 @@ public class ClientMetricsManagerTest {
) {
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -909,7 +981,7 @@ public class ClientMetricsManagerTest {
new PushTelemetryRequestData()
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
- .setMetrics(ByteBuffer.wrap(metrics)), true).build();
+ .setMetrics(ByteBuffer.wrap(metrics)), true).build((short) 0);
// Set the max bytes 1 to force the error.
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
@@ -925,7 +997,7 @@ public class ClientMetricsManagerTest {
@Test
public void testPushTelemetryConcurrentRequestNewClientInstance() throws Exception {
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -938,7 +1010,7 @@ public class ClientMetricsManagerTest {
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
.setCompressionType(CompressionType.NONE.id)
- .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build();
+ .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build((short) 0);
CountDownLatch lock = new CountDownLatch(2);
List<PushTelemetryResponse> responses = Collections.synchronizedList(new ArrayList<>());
@@ -1001,7 +1073,7 @@ public class ClientMetricsManagerTest {
@Test
public void testPushTelemetryConcurrentRequestAfterSubscriptionUpdate() throws Exception {
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -1014,7 +1086,7 @@ public class ClientMetricsManagerTest {
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
.setCompressionType(CompressionType.NONE.id)
- .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build();
+ .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build((short) 0);
clientMetricsManager.updateSubscription("sub-1", ClientMetricsTestUtils.defaultTestProperties());
assertEquals(1, clientMetricsManager.subscriptions().size());
@@ -1080,7 +1152,7 @@ public class ClientMetricsManagerTest {
@Test
public void testPushTelemetryPluginException() throws Exception {
ClientTelemetryExporterPlugin receiverPlugin = Mockito.mock(ClientTelemetryExporterPlugin.class);
- Mockito.doThrow(new RuntimeException("test exception")).when(receiverPlugin).exportMetrics(Mockito.any(), Mockito.any(), Mockito.anyInt(), Mockito.anyInt());
+ Mockito.doThrow(new RuntimeException("test exception")).when(receiverPlugin).exportMetrics(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyInt(), Mockito.anyInt());
try (
Metrics kafkaMetrics = new Metrics();
@@ -1091,7 +1163,7 @@ public class ClientMetricsManagerTest {
assertEquals(1, clientMetricsManager.subscriptions().size());
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -1104,7 +1176,7 @@ public class ClientMetricsManagerTest {
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
.setCompressionType(CompressionType.NONE.id)
- .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build();
+ .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build((short) 0);
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -1130,7 +1202,7 @@ public class ClientMetricsManagerTest {
// interval), not INVALID_RECORD (which tells the client to stop pushing telemetry entirely).
ClientTelemetryExporterPlugin receiverPlugin = Mockito.mock(ClientTelemetryExporterPlugin.class);
Mockito.doThrow(new TelemetryTooLargeException("Decompressed telemetry metrics exceed maximum allowed size: 100"))
- .when(receiverPlugin).exportMetrics(Mockito.any(), Mockito.any(), Mockito.anyInt(), Mockito.anyInt());
+ .when(receiverPlugin).exportMetrics(Mockito.any(), Mockito.any(), Mockito.any(), Mockito.anyInt(), Mockito.anyInt());
try (
Metrics kafkaMetrics = new Metrics();
@@ -1141,7 +1213,7 @@ public class ClientMetricsManagerTest {
assertEquals(1, clientMetricsManager.subscriptions().size());
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -1154,7 +1226,7 @@ public class ClientMetricsManagerTest {
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
.setCompressionType(CompressionType.NONE.id)
- .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build();
+ .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build((short) 0);
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -1175,7 +1247,7 @@ public class ClientMetricsManagerTest {
assertEquals(1, clientMetricsManager.subscriptions().size());
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -1190,7 +1262,7 @@ public class ClientMetricsManagerTest {
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
.setCompressionType(CompressionType.NONE.id)
- .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build();
+ .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build((short) 0);
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -1201,7 +1273,7 @@ public class ClientMetricsManagerTest {
assertEquals(Errors.UNKNOWN_SUBSCRIPTION_ID, instance.lastKnownError());
subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData().setClientInstanceId(subscriptionsResponse.data().clientInstanceId()), true).build();
+ new GetTelemetrySubscriptionsRequestData().setClientInstanceId(subscriptionsResponse.data().clientInstanceId()), true).build((short) 0);
subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
assertEquals(Errors.NONE, subscriptionsResponse.error());
@@ -1213,7 +1285,7 @@ public class ClientMetricsManagerTest {
assertEquals(1, clientMetricsManager.subscriptions().size());
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -1223,7 +1295,7 @@ public class ClientMetricsManagerTest {
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
.setCompressionType((byte) 10) // // Invalid compression type
- .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build();
+ .setMetrics(ByteBuffer.wrap("test-data".getBytes(StandardCharsets.UTF_8))), true).build((short) 0);
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -1234,7 +1306,7 @@ public class ClientMetricsManagerTest {
assertEquals(Errors.UNSUPPORTED_COMPRESSION_TYPE, instance.lastKnownError());
subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData().setClientInstanceId(subscriptionsResponse.data().clientInstanceId()), true).build();
+ new GetTelemetrySubscriptionsRequestData().setClientInstanceId(subscriptionsResponse.data().clientInstanceId()), true).build((short) 0);
subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
assertEquals(Errors.NONE, subscriptionsResponse.error());
@@ -1247,7 +1319,7 @@ public class ClientMetricsManagerTest {
ClientMetricsManager clientMetricsManager = new ClientMetricsManager(clientTelemetryExporterPlugin, 1, time, kafkaMetrics)
) {
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -1259,7 +1331,7 @@ public class ClientMetricsManagerTest {
new PushTelemetryRequestData()
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
- .setMetrics(ByteBuffer.wrap(metrics)), true).build();
+ .setMetrics(ByteBuffer.wrap(metrics)), true).build((short) 0);
// Set the max bytes 1 to force the error.
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
@@ -1271,7 +1343,7 @@ public class ClientMetricsManagerTest {
assertEquals(Errors.TELEMETRY_TOO_LARGE, instance.lastKnownError());
subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData().setClientInstanceId(subscriptionsResponse.data().clientInstanceId()), true).build();
+ new GetTelemetrySubscriptionsRequestData().setClientInstanceId(subscriptionsResponse.data().clientInstanceId()), true).build((short) 0);
subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
assertEquals(Errors.THROTTLING_QUOTA_EXCEEDED, subscriptionsResponse.error());
@@ -1286,7 +1358,7 @@ public class ClientMetricsManagerTest {
clientMetricsManager.updateSubscription("sub-1", properties);
GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -1326,7 +1398,7 @@ public class ClientMetricsManagerTest {
clientMetricsManager.updateSubscription("sub-1", properties);
GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse response1 = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -1367,7 +1439,7 @@ public class ClientMetricsManagerTest {
@Test
public void testCacheExpirationTaskCancelledOnInstanceUpdate() throws Exception {
GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContextWithConnectionId("conn-1"));
@@ -1385,7 +1457,7 @@ public class ClientMetricsManagerTest {
assertEquals(1, clientMetricsManager.subscriptions().size());
request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData().setClientInstanceId(clientInstanceId), true).build();
+ new GetTelemetrySubscriptionsRequestData().setClientInstanceId(clientInstanceId), true).build((short) 0);
response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContextWithConnectionId("conn-1"));
@@ -1410,7 +1482,7 @@ public class ClientMetricsManagerTest {
@Test
public void testRemoveConnection() throws Exception {
GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContextWithConnectionId("conn-1"));
@@ -1436,7 +1508,7 @@ public class ClientMetricsManagerTest {
@Test
public void testRemoveConnectionUnknownConnectionId() throws Exception {
GetTelemetrySubscriptionsRequest request = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse response = clientMetricsManager.processGetTelemetrySubscriptionRequest(
request, ClientMetricsTestUtils.requestContextWithConnectionId("conn-1"));
@@ -1464,7 +1536,7 @@ public class ClientMetricsManagerTest {
clientMetricsManager.updateSubscription("sub-1", ClientMetricsTestUtils.defaultTestProperties());
GetTelemetrySubscriptionsRequest subscriptionsRequest = new GetTelemetrySubscriptionsRequest.Builder(
- new GetTelemetrySubscriptionsRequestData(), true).build();
+ new GetTelemetrySubscriptionsRequestData(), true).build((short) 0);
GetTelemetrySubscriptionsResponse subscriptionsResponse = clientMetricsManager.processGetTelemetrySubscriptionRequest(
subscriptionsRequest, ClientMetricsTestUtils.requestContext());
@@ -1480,7 +1552,7 @@ public class ClientMetricsManagerTest {
new PushTelemetryRequestData()
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(1234) // wrong subscription id
- .setTerminating(true), true).build();
+ .setTerminating(true), true).build((short) 0);
PushTelemetryResponse response = clientMetricsManager.processPushTelemetryRequest(
request, ClientMetricsTestUtils.requestContext());
@@ -1497,7 +1569,7 @@ public class ClientMetricsManagerTest {
.setClientInstanceId(subscriptionsResponse.data().clientInstanceId())
.setSubscriptionId(subscriptionsResponse.data().subscriptionId())
.setCompressionType(CompressionType.NONE.id)
- .setTerminating(true), true).build();
+ .setTerminating(true), true).build((short) 0);
PushTelemetryResponse validResponse = clientMetricsManager.processPushTelemetryRequest(
validRequest, ClientMetricsTestUtils.requestContext());
diff --git a/server/src/test/java/org/apache/kafka/server/metrics/ClientMetricsTelemetryPluginTest.java b/server/src/test/java/org/apache/kafka/server/metrics/ClientMetricsTelemetryPluginTest.java
index f2803883cc6..294756c20eb 100644
--- a/server/src/test/java/org/apache/kafka/server/metrics/ClientMetricsTelemetryPluginTest.java
+++ b/server/src/test/java/org/apache/kafka/server/metrics/ClientMetricsTelemetryPluginTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.server.metrics;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.message.PushTelemetryRequestData;
import org.apache.kafka.common.requests.PushTelemetryRequest;
import org.apache.kafka.server.metrics.ClientMetricsTestUtils.TestClientMetricsReceiver;
@@ -27,6 +28,7 @@ import org.junit.jupiter.api.Test;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
+import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -56,12 +58,14 @@ public class ClientMetricsTelemetryPluginTest {
assertTrue(telemetryReceiver.metricsData.isEmpty());
byte[] metrics = "test-metrics".getBytes(StandardCharsets.UTF_8);
+ Uuid clientInstanceId = Uuid.randomUuid();
clientTelemetryExporterPlugin.exportMetrics(ClientMetricsTestUtils.requestContext(),
- new PushTelemetryRequest.Builder(new PushTelemetryRequestData().setMetrics(ByteBuffer.wrap(metrics)), true).build(), 5000, MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_DEFAULT);
+ new PushTelemetryRequest.Builder(new PushTelemetryRequestData().setMetrics(ByteBuffer.wrap(metrics)), true).build(), clientInstanceId, 5000, MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_DEFAULT);
assertEquals(1, telemetryReceiver.exportMetricsInvokedCount);
assertEquals(1, telemetryReceiver.metricsData.size());
assertEquals(metrics, telemetryReceiver.metricsData.get(0).array());
+ assertEquals(List.of(clientInstanceId), telemetryReceiver.clientInstanceIds);
}
@Test
@@ -76,14 +80,16 @@ public class ClientMetricsTelemetryPluginTest {
byte[] metrics = "test-metrics-new".getBytes(StandardCharsets.UTF_8);
int pushIntervalMs = 10000;
+ Uuid clientInstanceId = Uuid.randomUuid();
clientTelemetryExporterPlugin.exportMetrics(ClientMetricsTestUtils.requestContext(),
- new PushTelemetryRequest.Builder(new PushTelemetryRequestData().setMetrics(ByteBuffer.wrap(metrics)), true).build(), pushIntervalMs, MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_DEFAULT);
+ new PushTelemetryRequest.Builder(new PushTelemetryRequestData().setMetrics(ByteBuffer.wrap(metrics)), true).build(), clientInstanceId, pushIntervalMs, MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_DEFAULT);
assertEquals(1, telemetryExporter.exportMetricsInvokedCount);
assertEquals(1, telemetryExporter.metricsData.size());
assertEquals(ByteBuffer.wrap(metrics), telemetryExporter.metricsData.get(0));
assertEquals(1, telemetryExporter.pushIntervals.size());
assertEquals(pushIntervalMs, telemetryExporter.pushIntervals.get(0));
+ assertEquals(List.of(clientInstanceId), telemetryExporter.clientInstanceIds);
}
@Test
@@ -95,8 +101,9 @@ public class ClientMetricsTelemetryPluginTest {
byte[] metrics = "test-metrics-both".getBytes(StandardCharsets.UTF_8);
int pushIntervalMs = 15000;
+ Uuid clientInstanceId = Uuid.randomUuid();
clientTelemetryExporterPlugin.exportMetrics(ClientMetricsTestUtils.requestContext(),
- new PushTelemetryRequest.Builder(new PushTelemetryRequestData().setMetrics(ByteBuffer.wrap(metrics)), true).build(), pushIntervalMs, MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_DEFAULT);
+ new PushTelemetryRequest.Builder(new PushTelemetryRequestData().setMetrics(ByteBuffer.wrap(metrics)), true).build(), clientInstanceId, pushIntervalMs, MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_DEFAULT);
// Both should be called since they are separate objects
assertEquals(1, telemetryReceiver.exportMetricsInvokedCount);
@@ -105,6 +112,8 @@ public class ClientMetricsTelemetryPluginTest {
assertEquals(ByteBuffer.wrap(metrics), telemetryReceiver.metricsData.get(0));
assertEquals(ByteBuffer.wrap(metrics), telemetryExporter.metricsData.get(0));
assertEquals(pushIntervalMs, telemetryExporter.pushIntervals.get(0));
+ assertEquals(List.of(clientInstanceId), telemetryReceiver.clientInstanceIds);
+ assertEquals(List.of(clientInstanceId), telemetryExporter.clientInstanceIds);
}
@Test
@@ -121,8 +130,9 @@ public class ClientMetricsTelemetryPluginTest {
byte[] metrics = "test-metrics-dual".getBytes(StandardCharsets.UTF_8);
int pushIntervalMs = 12000;
+ Uuid clientInstanceId = Uuid.randomUuid();
clientTelemetryExporterPlugin.exportMetrics(ClientMetricsTestUtils.requestContext(),
- new PushTelemetryRequest.Builder(new PushTelemetryRequestData().setMetrics(ByteBuffer.wrap(metrics)), true).build(), pushIntervalMs, MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_DEFAULT);
+ new PushTelemetryRequest.Builder(new PushTelemetryRequestData().setMetrics(ByteBuffer.wrap(metrics)), true).build(), clientInstanceId, pushIntervalMs, MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_DEFAULT);
// Only the exporter should be called (receiver should not be invoked)
assertEquals(0, dualImpl.getReceiver().exportMetricsInvokedCount);
@@ -130,5 +140,6 @@ public class ClientMetricsTelemetryPluginTest {
assertEquals(ByteBuffer.wrap(metrics), dualImpl.getExporter().metricsData.get(0));
assertEquals(pushIntervalMs, dualImpl.getExporter().pushIntervals.get(0));
+ assertEquals(List.of(clientInstanceId), dualImpl.getExporter().clientInstanceIds);
}
}
diff --git a/server/src/test/java/org/apache/kafka/server/metrics/ClientTelemetryTest.java b/server/src/test/java/org/apache/kafka/server/metrics/ClientTelemetryTest.java
index dcda4b5f19f..a7a1590226e 100644
--- a/server/src/test/java/org/apache/kafka/server/metrics/ClientTelemetryTest.java
+++ b/server/src/test/java/org/apache/kafka/server/metrics/ClientTelemetryTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.server.metrics;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.message.PushTelemetryRequestData;
import org.apache.kafka.common.requests.PushTelemetryRequest;
import org.apache.kafka.server.metrics.ClientMetricsTestUtils.TestClientMetricsReceiver;
@@ -27,6 +28,7 @@ import org.junit.jupiter.api.Test;
import java.net.UnknownHostException;
import java.nio.ByteBuffer;
import java.nio.charset.StandardCharsets;
+import java.util.List;
import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -49,15 +51,18 @@ public class ClientTelemetryTest {
clientTelemetryExporterPlugin.add(receiver2);
byte[] metrics = "test-metrics-multiple".getBytes(StandardCharsets.UTF_8);
+ Uuid clientInstanceId = Uuid.randomUuid();
clientTelemetryExporterPlugin.exportMetrics(ClientMetricsTestUtils.requestContext(),
new PushTelemetryRequest.Builder(new PushTelemetryRequestData().setMetrics(ByteBuffer.wrap(metrics)), true).build(),
- 5000, MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_DEFAULT);
+ clientInstanceId, 5000, MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_DEFAULT);
// Verify both receivers were called
assertEquals(1, receiver1.exportMetricsInvokedCount);
assertEquals(1, receiver2.exportMetricsInvokedCount);
assertEquals(ByteBuffer.wrap(metrics), receiver1.metricsData.get(0));
assertEquals(ByteBuffer.wrap(metrics), receiver2.metricsData.get(0));
+ assertEquals(List.of(clientInstanceId), receiver1.clientInstanceIds);
+ assertEquals(List.of(clientInstanceId), receiver2.clientInstanceIds);
}
@Test
@@ -71,9 +76,10 @@ public class ClientTelemetryTest {
byte[] metrics = "test-metrics-multiple-new".getBytes(StandardCharsets.UTF_8);
int pushIntervalMs = 20000;
+ Uuid clientInstanceId = Uuid.randomUuid();
clientTelemetryExporterPlugin.exportMetrics(ClientMetricsTestUtils.requestContext(),
new PushTelemetryRequest.Builder(new PushTelemetryRequestData().setMetrics(ByteBuffer.wrap(metrics)), true).build(),
- pushIntervalMs, MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_DEFAULT);
+ clientInstanceId, pushIntervalMs, MetricConfigs.CLIENT_TELEMETRY_MAX_BYTES_DEFAULT);
// Verify both exporters were called
assertEquals(1, exporter1.exportMetricsInvokedCount);
@@ -82,5 +88,7 @@ public class ClientTelemetryTest {
assertEquals(ByteBuffer.wrap(metrics), exporter2.metricsData.get(0));
assertEquals(pushIntervalMs, exporter1.pushIntervals.get(0));
assertEquals(pushIntervalMs, exporter2.pushIntervals.get(0));
+ assertEquals(List.of(clientInstanceId), exporter1.clientInstanceIds);
+ assertEquals(List.of(clientInstanceId), exporter2.clientInstanceIds);
}
}
diff --git a/server/src/testFixtures/java/org/apache/kafka/server/metrics/ClientMetricsTestUtils.java b/server/src/testFixtures/java/org/apache/kafka/server/metrics/ClientMetricsTestUtils.java
index 8e8bb21b7da..74c58774372 100644
--- a/server/src/testFixtures/java/org/apache/kafka/server/metrics/ClientMetricsTestUtils.java
+++ b/server/src/testFixtures/java/org/apache/kafka/server/metrics/ClientMetricsTestUtils.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.server.metrics;
+import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.network.ClientInformation;
import org.apache.kafka.common.network.ListenerName;
import org.apache.kafka.common.protocol.ApiKeys;
@@ -72,6 +73,19 @@ public class ClientMetricsTestUtils {
false);
}
+ public static RequestContext requestContext(ApiKeys apiKey, short apiVersion, Uuid clientInstanceId) throws UnknownHostException {
+ return new RequestContext(
+ new RequestHeader(apiKey, apiVersion, "producer-1", clientInstanceId, 0),
+ TestUtils.randomString(5),
+ InetAddress.getLocalHost(),
+ Optional.of(CLIENT_PORT),
+ KafkaPrincipal.ANONYMOUS,
+ ListenerName.forSecurityProtocol(SecurityProtocol.PLAINTEXT),
+ SecurityProtocol.PLAINTEXT,
+ new ClientInformation("apache-kafka-java", "3.5.2"),
+ false);
+ }
+
public static RequestContext requestContextWithNullClientInfo() throws UnknownHostException {
return new RequestContext(
new RequestHeader(ApiKeys.GET_TELEMETRY_SUBSCRIPTIONS, (short) 0, "producer-1", 0),
@@ -102,10 +116,12 @@ public class ClientMetricsTestUtils {
public static class TestClientMetricsReceiver implements ClientTelemetryReceiver {
public int exportMetricsInvokedCount = 0;
public List<ByteBuffer> metricsData = new ArrayList<>();
+ public List<Uuid> clientInstanceIds = new ArrayList<>();
public void exportMetrics(AuthorizableRequestContext context, ClientTelemetryPayload payload) {
exportMetricsInvokedCount += 1;
metricsData.add(payload.data());
+ clientInstanceIds.add(payload.clientInstanceId());
}
}
@@ -113,12 +129,14 @@ public class ClientMetricsTestUtils {
public int exportMetricsInvokedCount = 0;
public List<ByteBuffer> metricsData = new ArrayList<>();
public List<Integer> pushIntervals = new ArrayList<>();
+ public List<Uuid> clientInstanceIds = new ArrayList<>();
@Override
public void exportMetrics(ClientTelemetryContext context, ClientTelemetryPayload payload) {
exportMetricsInvokedCount += 1;
metricsData.add(payload.data());
pushIntervals.add(context.pushIntervalMs());
+ clientInstanceIds.add(payload.clientInstanceId());
}
}