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

Code before:
                Optional<Integer> errorIntervalMsOpt = ClientTelemetryUtils.maybeFetchErrorIntervalMs(data.errorCode(),
                    subscription.pushIntervalMs());
                /*
                 If the error code indicates that the interval ms needs to be updated as per the error
                 code then update the interval ms and state so that the subscription can be re-fetched,
                 and the push retried.
                */
                if (errorIntervalMsOpt.isPresent()) {
                    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) {

Review comment (suggestion):
nit: Personally prefers to write private method after public methods hence I would moved it later in the class. Feel free to ignore as that's my personal preference for readability.

Code after:
                Optional<Integer> errorIntervalMsOpt = ClientTelemetryUtils.maybeFetchErrorIntervalMs(data.errorCode(),
                    subscription.pushIntervalMs());
                /*
                 If the error code indicates that the interval ms needs to be updated as per the error
                 code then update the interval ms and state so that the subscription can be re-fetched,
                 and the push retried.
                */
                if (errorIntervalMsOpt.isPresent()) {
                    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)) {