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

Code before:
                enabled = false;
                return false;
            } finally {
                lock.writeLock().unlock();
            }
        }

        private void handleFailedRequest(KafkaException maybeFatalException) {
            final long nowMs = time.milliseconds();
            lock.writeLock().lock();
            try {
                if (isTerminatingState()) {
                    return;
                }
                if (state != ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS && state != ClientTelemetryState.PUSH_IN_PROGRESS) {
                    log.warn("Could not transition state after failed telemetry from state {}, disabling telemetry", state);
                    updateErrorResult(Integer.MAX_VALUE, nowMs);
                    return;
                }

                /*
                 The broker might not support telemetry. Let's sleep for a while before trying request
                 again. We may disconnect from the broker and connect to a broker that supports client
                 telemetry.
                */
                boolean shouldWait = isRetryable(maybeFatalException);
                if (shouldWait) {
                    updateErrorResult(DEFAULT_PUSH_INTERVAL_MS, nowMs);
                } else {
                    boolean shouldLog = shouldLog(maybeFatalException);
                    if (shouldLog) {
                        log.warn("Received unrecoverable error from broker, disabling telemetry");
                    }
                    updateErrorResult(Integer.MAX_VALUE, nowMs);
                }

                if (!maybeSetState(ClientTelemetryState.SUBSCRIPTION_NEEDED)) {
                    log.warn("Could not transition state after failed telemetry from state {}", state);
                }
            } finally {
                lock.writeLock().unlock();
            }
        }

        private MetricsData createPayload(List<SinglePointMetric> emittedMetrics) {
            MetricsData.Builder builder = MetricsData.newBuilder();
            emittedMetrics.forEach(metric -> {
                Metric m = metric.builder().build();
                ResourceMetrics rm = buildMetric(m);
                builder.addResourceMetrics(rm);
            });

Review comment (suggestion):
Personally, I'd prefer to see
```
if (isRetryable(maybeFatalException))
```

There's no need for a shouldWait variable.

Code after:

        private void handleFailedRequest(KafkaException maybeFatalException) {
            final long nowMs = time.milliseconds();
            lock.writeLock().lock();
            try {
                if (isTerminatingState()) {
                    return;
                }
                if (state != ClientTelemetryState.SUBSCRIPTION_IN_PROGRESS && state != ClientTelemetryState.PUSH_IN_PROGRESS) {
                    log.warn("Could not transition state after failed telemetry from state {}, disabling telemetry", state);
                    updateErrorResult(Integer.MAX_VALUE, nowMs);
                    return;
                }

                /*
                 The broker might not support telemetry. Let's sleep for a while before trying request
                 again. We may disconnect from the broker and connect to a broker that supports client
                 telemetry.
                */
                if (isRetryable(maybeFatalException)) {
                    updateErrorResult(DEFAULT_PUSH_INTERVAL_MS, nowMs);
                } else {
                    if (!(maybeFatalException instanceof UnsupportedVersionException)) {
                        log.warn("Received unrecoverable error from broker, disabling telemetry");
                    }
                    updateErrorResult(Integer.MAX_VALUE, nowMs);
                }

                if (!maybeSetState(ClientTelemetryState.SUBSCRIPTION_NEEDED)) {
                    log.warn("Could not transition state after failed telemetry from state {}", state);
                }
            } finally {
                lock.writeLock().unlock();
            }
        }

        private MetricsData createPayload(List<SinglePointMetric> emittedMetrics) {
            MetricsData.Builder builder = MetricsData.newBuilder();
            emittedMetrics.forEach(metric -> {
                Metric m = metric.builder().build();
                ResourceMetrics rm = buildMetric(m);
                builder.addResourceMetrics(rm);
            });
            return builder.build();
        }

        // Visible for testing
        ClientTelemetrySubscription subscription() {
            lock.readLock().lock();
            try {
                return subscription;