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

Code before:
                return Optional.empty();
            }

            CompressionType compressionType = ClientTelemetryUtils.preferredCompressionType(localSubscription.acceptedCompressionTypes());
            ByteBuffer compressedPayload;
            try {
                compressedPayload = ClientTelemetryUtils.compress(payload, compressionType);
            } catch (IOException | NoClassDefFoundError e) {
                log.debug("Failed to compress telemetry payload for compression: {}, sending uncompressed data", compressionType, e);
                compressedPayload = ByteBuffer.wrap(payload.toByteArray());
                compressionType = CompressionType.NONE;
            }

            AbstractRequest.Builder<?> requestBuilder = new PushTelemetryRequest.Builder(
                new PushTelemetryRequestData()
                    .setClientInstanceId(localSubscription.clientInstanceId())
                    .setSubscriptionId(localSubscription.subscriptionId())
                    .setTerminating(terminating)
                    .setCompressionType(compressionType.id)
                    .setMetrics(compressedPayload), true);

            return Optional.of(requestBuilder);
        }

        /**
         * Updates the {@link ClientTelemetrySubscription}, {@link #intervalMs}, and
         * {@link #lastRequestMs}.
         * <p>
         * After the update, the {@link #subscriptionLoaded} condition is signaled so any threads
         * waiting on the subscription can be unblocked.
         *
         * @param subscription Updated subscription that will replace any current subscription
         * @param timeMs       Time in milliseconds representing the current time
         */
        // Visible for testing
        void updateSubscriptionResult(ClientTelemetrySubscription subscription, long timeMs) {
            lock.writeLock().lock();
            try {
                this.subscription = Objects.requireNonNull(subscription);
                /*
                 If the subscription is updated for the client, we want to attempt to spread out the push
                 requests between 50% and 150% of the push interval value from the broker. This helps us
                 to avoid the case where multiple clients are started at the same time and end up sending
                 all their data at the same time.
                */
                if (state == ClientTelemetryState.PUSH_NEEDED) {
                    intervalMs = computeStaggeredIntervalMs(subscription.pushIntervalMs(), INITIAL_PUSH_JITTER_LOWER, INITIAL_PUSH_JITTER_UPPER);
                } else {
                    intervalMs = subscription.pushIntervalMs();
                }
                lastRequestMs = timeMs;

Review comment (suggestion):
Perhaps we could have an `unsupportedTypes` collection to filter out types according to `NoClassDefFoundError`? for example:
```java

                if (e instanceof NoClassDefFoundError) unsupportedTypes.add(compressionType);
```
`ClientTelemetryUtils.preferredCompressionType` could leverage the collection `compressionType` to avoid using  unsupported compression. for example:
```java
    public static CompressionType preferredCompressionType(List<CompressionType> acceptedCompressionTypes, Set<CompressionType> unsupportedTypes) {
        if (acceptedCompressionTypes == null) return CompressionType.NONE;
        // Broker is providing the compression types in order of preference. Grab the
        // first one.
        return acceptedCompressionTypes.stream()
                .filter(t -> !unsupportedTypes.contains(t))
                .findFirst()
                .orElse(CompressionType.NONE);
    }
````

Code after:
                updateErrorResult(localSubscription.pushIntervalMs, time.milliseconds());
                return Optional.empty();
            }

            CompressionType compressionType = ClientTelemetryUtils.preferredCompressionType(localSubscription.acceptedCompressionTypes(), unsupportedCompressionTypes);
            ByteBuffer compressedPayload;
            try {
                compressedPayload = ClientTelemetryUtils.compress(payload, compressionType);
            } catch (Throwable e) {
                // Distinguish between recoverable errors (NoClassDefFoundError for missing compression libs) 
                // and fatal errors (OutOfMemoryError, etc.) that should terminate telemetry.
                if (e instanceof Error && !(e instanceof NoClassDefFoundError) && !(e.getCause() instanceof NoClassDefFoundError)) {
                    lock.writeLock().lock();
                    try {
                        state = ClientTelemetryState.TERMINATED;
                    } finally {
                        lock.writeLock().unlock();
                    }
                    log.error("Unexpected error occurred while compressing telemetry payload for compression: {}, stopping client telemetry", compressionType, e);
                    throw new KafkaException("Unexpected compression error", e);
                }

                log.debug("Failed to compress telemetry payload for compression: {}, sending uncompressed data", compressionType, e);
                unsupportedCompressionTypes.add(compressionType);
                compressedPayload = ByteBuffer.wrap(payload.toByteArray());
                compressionType = CompressionType.NONE;
            }

            AbstractRequest.Builder<?> requestBuilder = new PushTelemetryRequest.Builder(
                new PushTelemetryRequestData()
                    .setClientInstanceId(localSubscription.clientInstanceId())
                    .setSubscriptionId(localSubscription.subscriptionId())
                    .setTerminating(terminating)
                    .setCompressionType(compressionType.id)
                    .setMetrics(compressedPayload), true);

            return Optional.of(requestBuilder);
        }

        /**
         * Updates the {@link ClientTelemetrySubscription}, {@link #intervalMs}, and
         * {@link #lastRequestMs}.
         * <p>
         * After the update, the {@link #subscriptionLoaded} condition is signaled so any threads
         * waiting on the subscription can be unblocked.
         *
         * @param subscription Updated subscription that will replace any current subscription
         * @param timeMs       Time in milliseconds representing the current time
         */
        // Visible for testing
        void updateSubscriptionResult(ClientTelemetrySubscription subscription, long timeMs) {