You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by bb...@apache.org on 2025/10/09 20:49:25 UTC
(kafka) branch 4.1 updated: KAFKA-19747: Update ClientTelemetryReporter telemetry push error handling (#20661)
This is an automated email from the ASF dual-hosted git repository.
bbejeck pushed a commit to branch 4.1
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/4.1 by this push:
new aceb32d15e6 KAFKA-19747: Update ClientTelemetryReporter telemetry push error handling (#20661)
aceb32d15e6 is described below
commit aceb32d15e621c33df7c45220d12520c5416596a
Author: Bill Bejeck <bb...@apache.org>
AuthorDate: Thu Oct 9 16:26:39 2025 -0400
KAFKA-19747: Update ClientTelemetryReporter telemetry push error handling (#20661)
When a failure occurs with a push telemetry request, any exception is
treated as fatal, increasing the time interval to `Integer.MAX_VALUE`
effectively turning telemetry off. This PR updates the error handling
to check if the exception is a transient one with expected recovery and
keeps the telemetry interval value the same in those cases since a
recovery is expected.
Reviewers: Apoorv Mittal <ap...@gmail.com>, Matthias
Sax<mj...@apache.org>
---
checkstyle/suppressions.xml | 2 +-
.../internals/ClientTelemetryReporter.java | 11 +-
.../internals/ClientTelemetryReporterTest.java | 163 +++++++++++++++++++++
3 files changed, 173 insertions(+), 3 deletions(-)
diff --git a/checkstyle/suppressions.xml b/checkstyle/suppressions.xml
index ca1c3d65763..7c02cf3adf9 100644
--- a/checkstyle/suppressions.xml
+++ b/checkstyle/suppressions.xml
@@ -87,7 +87,7 @@
files="ClientUtils.java"/>
<suppress checks="ClassDataAbstractionCoupling"
- files="(KafkaConsumer|ConsumerCoordinator|AbstractFetch|KafkaProducer|AbstractRequest|AbstractResponse|TransactionManager|Admin|KafkaAdminClient|MockAdminClient|KafkaNetworkChannelTest).java"/>
+ files="(KafkaConsumer|ConsumerCoordinator|AbstractFetch|KafkaProducer|AbstractRequest|AbstractResponse|TransactionManager|Admin|KafkaAdminClient|MockAdminClient|KafkaNetworkChannelTest|ClientTelemetryReporterTest).java"/>
<suppress checks="ClassDataAbstractionCoupling"
files="(Errors|SaslAuthenticatorTest|AgentTest|CoordinatorTest|NetworkClientTest).java"/>
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 e0491943fef..ae60aef66c9 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
@@ -20,6 +20,7 @@ import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.errors.InterruptException;
+import org.apache.kafka.common.errors.RetriableException;
import org.apache.kafka.common.message.GetTelemetrySubscriptionsRequestData;
import org.apache.kafka.common.message.GetTelemetrySubscriptionsResponseData;
import org.apache.kafka.common.message.PushTelemetryRequestData;
@@ -526,13 +527,13 @@ public class ClientTelemetryReporter implements MetricsReporter {
@Override
public void handleFailedGetTelemetrySubscriptionsRequest(KafkaException maybeFatalException) {
log.debug("The broker generated an error for the get telemetry network API request", maybeFatalException);
- handleFailedRequest(maybeFatalException != null);
+ handleFailedRequest(isRetryable(maybeFatalException));
}
@Override
public void handleFailedPushTelemetryRequest(KafkaException maybeFatalException) {
log.debug("The broker generated an error for the push telemetry network API request", maybeFatalException);
- handleFailedRequest(maybeFatalException != null);
+ handleFailedRequest(isRetryable(maybeFatalException));
}
@Override
@@ -626,6 +627,12 @@ public class ClientTelemetryReporter implements MetricsReporter {
}
}
+ private boolean isRetryable(final KafkaException maybeFatalException) {
+ return maybeFatalException == null ||
+ (maybeFatalException instanceof RetriableException) ||
+ (maybeFatalException.getCause() != null && maybeFatalException.getCause() instanceof RetriableException);
+ }
+
private Optional<Builder<?>> createSubscriptionRequest(ClientTelemetrySubscription localSubscription) {
/*
If we've previously retrieved a subscription, it will contain the client instance ID
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 b708b4eeb60..003061563b7 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
@@ -19,7 +19,13 @@ package org.apache.kafka.common.telemetry.internals;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
+import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Uuid;
+import org.apache.kafka.common.errors.AuthorizationException;
+import org.apache.kafka.common.errors.DisconnectException;
+import org.apache.kafka.common.errors.NetworkException;
+import org.apache.kafka.common.errors.TimeoutException;
+import org.apache.kafka.common.errors.UnsupportedVersionException;
import org.apache.kafka.common.message.GetTelemetrySubscriptionsRequestData;
import org.apache.kafka.common.message.GetTelemetrySubscriptionsResponseData;
import org.apache.kafka.common.message.PushTelemetryRequestData;
@@ -770,6 +776,163 @@ public class ClientTelemetryReporterTest {
.telemetrySender()).state());
}
+ @Test
+ public void testHandleFailedGetTelemetrySubscriptionsRequestWithRetriableException() {
+ ClientTelemetryReporter.DefaultClientTelemetrySender telemetrySender = (ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter.telemetrySender();
+ telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
+
+ KafkaException retriableException = new TimeoutException("Request timed out");
+ telemetrySender.handleFailedGetTelemetrySubscriptionsRequest(retriableException);
+
+ assertEquals(ClientTelemetryState.SUBSCRIPTION_NEEDED, telemetrySender.state());
+ assertEquals(ClientTelemetryReporter.DEFAULT_PUSH_INTERVAL_MS, telemetrySender.intervalMs());
+ assertTrue(telemetrySender.enabled());
+ }
+
+ @Test
+ public void testHandleFailedGetTelemetrySubscriptionsRequestWithWrappedRetriableException() {
+ ClientTelemetryReporter.DefaultClientTelemetrySender telemetrySender = (ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter.telemetrySender();
+ telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
+
+ KafkaException wrappedException = new KafkaException(new DisconnectException("Connection lost"));
+ telemetrySender.handleFailedGetTelemetrySubscriptionsRequest(wrappedException);
+
+ assertEquals(ClientTelemetryState.SUBSCRIPTION_NEEDED, telemetrySender.state());
+ assertEquals(ClientTelemetryReporter.DEFAULT_PUSH_INTERVAL_MS, telemetrySender.intervalMs());
+ assertTrue(telemetrySender.enabled());
+ }
+
+ @Test
+ public void testHandleFailedGetTelemetrySubscriptionsRequestWithFatalException() {
+ ClientTelemetryReporter.DefaultClientTelemetrySender telemetrySender = (ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter.telemetrySender();
+ telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
+
+ KafkaException fatalException = new AuthorizationException("Not authorized for telemetry");
+ telemetrySender.handleFailedGetTelemetrySubscriptionsRequest(fatalException);
+
+ assertEquals(ClientTelemetryState.SUBSCRIPTION_NEEDED, telemetrySender.state());
+ assertEquals(Integer.MAX_VALUE, telemetrySender.intervalMs());
+ assertFalse(telemetrySender.enabled());
+ }
+
+ @Test
+ public void testHandleFailedGetTelemetrySubscriptionsRequestWithWrappedFatalException() {
+ ClientTelemetryReporter.DefaultClientTelemetrySender telemetrySender = (ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter.telemetrySender();
+ telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
+
+ KafkaException wrappedException = new KafkaException("Version check failed",
+ new UnsupportedVersionException("Broker doesn't support telemetry"));
+ telemetrySender.handleFailedGetTelemetrySubscriptionsRequest(wrappedException);
+
+ assertEquals(ClientTelemetryState.SUBSCRIPTION_NEEDED, telemetrySender.state());
+ assertEquals(Integer.MAX_VALUE, telemetrySender.intervalMs());
+ assertFalse(telemetrySender.enabled());
+ }
+
+ @Test
+ public void testHandleFailedPushTelemetryRequestWithRetriableException() {
+ ClientTelemetryReporter.DefaultClientTelemetrySender telemetrySender = (ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter.telemetrySender();
+ telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.PUSH_NEEDED));
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.PUSH_IN_PROGRESS));
+
+ KafkaException networkException = new NetworkException("Network failure");
+ telemetrySender.handleFailedPushTelemetryRequest(networkException);
+
+ assertEquals(ClientTelemetryState.SUBSCRIPTION_NEEDED, telemetrySender.state());
+ assertEquals(ClientTelemetryReporter.DEFAULT_PUSH_INTERVAL_MS, telemetrySender.intervalMs());
+ assertTrue(telemetrySender.enabled());
+ }
+
+ @Test
+ public void testHandleFailedPushTelemetryRequestWithFatalException() {
+ ClientTelemetryReporter.DefaultClientTelemetrySender telemetrySender = (ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter.telemetrySender();
+ telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.PUSH_NEEDED));
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.PUSH_IN_PROGRESS));
+
+ KafkaException authException = new AuthorizationException("Not authorized to push telemetry");
+ telemetrySender.handleFailedPushTelemetryRequest(authException);
+
+ assertEquals(ClientTelemetryState.SUBSCRIPTION_NEEDED, telemetrySender.state());
+ assertEquals(Integer.MAX_VALUE, telemetrySender.intervalMs());
+ assertFalse(telemetrySender.enabled());
+ }
+
+ @Test
+ public void testHandleFailedRequestWithMultipleRetriableExceptionsInChain() {
+ ClientTelemetryReporter.DefaultClientTelemetrySender telemetrySender = (ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter.telemetrySender();
+ telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
+
+ KafkaException chainedException = new TimeoutException("Outer timeout",
+ new DisconnectException("Inner disconnect"));
+ telemetrySender.handleFailedGetTelemetrySubscriptionsRequest(chainedException);
+
+ assertEquals(ClientTelemetryState.SUBSCRIPTION_NEEDED, telemetrySender.state());
+ assertEquals(ClientTelemetryReporter.DEFAULT_PUSH_INTERVAL_MS, telemetrySender.intervalMs());
+ assertTrue(telemetrySender.enabled());
+ }
+
+ @Test
+ public void testHandleFailedRequestWithGenericKafkaException() {
+ ClientTelemetryReporter.DefaultClientTelemetrySender telemetrySender = (ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter.telemetrySender();
+ telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
+
+ KafkaException genericException = new KafkaException("Unknown error");
+ telemetrySender.handleFailedGetTelemetrySubscriptionsRequest(genericException);
+
+ assertEquals(ClientTelemetryState.SUBSCRIPTION_NEEDED, telemetrySender.state());
+ assertEquals(Integer.MAX_VALUE, telemetrySender.intervalMs());
+ assertFalse(telemetrySender.enabled());
+ }
+
+ @Test
+ public void testHandleFailedRequestDuringTermination() {
+ ClientTelemetryReporter.DefaultClientTelemetrySender telemetrySender = (ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter.telemetrySender();
+ telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.PUSH_NEEDED));
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.TERMINATING_PUSH_NEEDED));
+
+ KafkaException exception = new TimeoutException("Timeout");
+ telemetrySender.handleFailedPushTelemetryRequest(exception);
+
+ assertEquals(ClientTelemetryState.TERMINATING_PUSH_NEEDED, telemetrySender.state());
+ assertTrue(telemetrySender.enabled());
+ }
+
+ @Test
+ public void testSequentialFailuresWithDifferentExceptionTypes() {
+ ClientTelemetryReporter.DefaultClientTelemetrySender telemetrySender = (ClientTelemetryReporter.DefaultClientTelemetrySender) clientTelemetryReporter.telemetrySender();
+ telemetrySender.updateSubscriptionResult(subscription, time.milliseconds());
+
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
+ telemetrySender.handleFailedGetTelemetrySubscriptionsRequest(
+ new TimeoutException("Timeout 1"));
+ assertEquals(ClientTelemetryState.SUBSCRIPTION_NEEDED, telemetrySender.state());
+ assertTrue(telemetrySender.enabled());
+
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
+ telemetrySender.handleFailedGetTelemetrySubscriptionsRequest(
+ new DisconnectException("Disconnect"));
+ assertEquals(ClientTelemetryState.SUBSCRIPTION_NEEDED, telemetrySender.state());
+ assertTrue(telemetrySender.enabled());
+
+ assertTrue(telemetrySender.maybeSetState(ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS));
+ telemetrySender.handleFailedGetTelemetrySubscriptionsRequest(
+ new UnsupportedVersionException("Version not supported"));
+ assertEquals(ClientTelemetryState.SUBSCRIPTION_NEEDED, telemetrySender.state());
+ assertFalse(telemetrySender.enabled());
+ }
+
@AfterEach
public void tearDown() {
clientTelemetryReporter.close();