Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java (Java)

Code before:
                    if (!maybeSetState(ClientTelemetryState.SUBSCRIPTION_NEEDED)) {
                        log.warn("Unable to transition state after failed push telemetry from state {}", state);
                    }
                    updateErrorResult(errorIntervalMsOpt.get(), nowMs);
                    return;
                }

                lastRequestMs = nowMs;
                intervalMs = subscription.pushIntervalMs();
                if (!maybeSetState(ClientTelemetryState.PUSH_NEEDED)) {
                    log.warn("Unable to transition state after successful push telemetry from state {}", state);
                }
            } finally {
                lock.writeLock().unlock();
            }
        }
        
        private boolean isRetryable(final KafkaException maybeFatalException) {
            if (maybeFatalException == null) {
                return true;
            }

            Throwable cause;
            if (maybeFatalException.getClass() == KafkaException.class) {
                if (maybeFatalException.getCause() == null) {
                    return false;
                } else {
                    cause = maybeFatalException.getCause();
                }
            } else {
                cause = maybeFatalException;
            }
            while (cause != null) {
                if (!(cause instanceof RetriableException)) {
                    return false;
                }
                cause = cause.getCause();
            }
            return true;
        }

        @Override
        public void handleFailedGetTelemetrySubscriptionsRequest(KafkaException maybeFatalException) {
            log.debug("The broker generated an error for the get telemetry network API request", maybeFatalException);
            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(isRetryable(maybeFatalException));

Review comment (suggestion):
With a plain `KafkaException` we don't know what this is, so treat it as fatal.

Code after:
                    if (!maybeSetState(ClientTelemetryState.SUBSCRIPTION_NEEDED)) {
                        log.warn("Unable to transition state after failed push telemetry from state {}", state);
                    }
                    updateErrorResult(errorIntervalMsOpt.get(), nowMs);
                    return;
                }

                lastRequestMs = nowMs;
                intervalMs = subscription.pushIntervalMs();
                if (!maybeSetState(ClientTelemetryState.PUSH_NEEDED)) {
                    log.warn("Unable to transition state after successful push telemetry from state {}", state);
                }
            } finally {
                lock.writeLock().unlock();
            }
        }

        @Override
        public void handleFailedGetTelemetrySubscriptionsRequest(KafkaException maybeFatalException) {
            log.debug("The broker generated an error for the get telemetry network API request", maybeFatalException);
            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(isRetryable(maybeFatalException));
        }

        @Override
        public Optional<Uuid> clientInstanceId(Duration timeout) {
            final long timeoutMs = timeout.toMillis();
            if (timeoutMs < 0) {
                throw new IllegalArgumentException("The timeout cannot be negative for fetching client instance id.");
            }

            lock.writeLock().lock();
            try {
                if (subscription == null) {
                    // If we have a non-negative timeout and no-subscription, let's wait for one to be retrieved.
                    log.debug("Waiting for telemetry subscription containing the client instance ID with timeoutMillis = {} ms.", timeoutMs);
                    try {
                        if (!subscriptionLoaded.await(timeoutMs, TimeUnit.MILLISECONDS)) {
                            log.debug("Wait for telemetry subscription elapsed; may not have actually loaded it");
                        }
                    } catch (InterruptedException e) {
                        throw new InterruptException(e);
                    }
                }

                if (subscription == null) {