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

Code before:

        String val = (String) m.get(key);
        if (val.isEmpty()) {
            log.trace("{} is empty string. value = {} map {}", key, val, m);
            return false;
        }
        return true;
    }

    public static boolean validateRequiredResourceLabels(Map<String, String> metadata) {
        return validateResourceLabel(metadata, MetricsContext.NAMESPACE);
    }

    /**
     * Determines the preferred compression type from broker-accepted types, avoiding unsupported ones.
     * 
     * @param acceptedCompressionTypes the list of compression types accepted by the broker in order 
     *                                of preference, or null/empty if no compression is accepted
     * @param unsupportedCompressionTypes the set of compression types that should be avoided due to 
     *                                   missing libraries or previous failures (must not be null)
     * @return the preferred compression type to use, or {@link CompressionType#NONE} if no acceptable
     *         compression type is available
     * @throws NullPointerException if unsupportedCompressionTypes is null
     */
    public static CompressionType preferredCompressionType(List<CompressionType> acceptedCompressionTypes, Set<CompressionType> unsupportedCompressionTypes) {
        if (acceptedCompressionTypes == null) {
            return CompressionType.NONE;
        }

        // Broker is providing the compression types in order of preference. Grab the
        // first one that's not unsupported.
        return acceptedCompressionTypes.stream()
                .filter(t -> !unsupportedCompressionTypes.contains(t))
                .findFirst()
                .orElse(CompressionType.NONE);
    }

    public static ByteBuffer compress(MetricsData metrics, CompressionType compressionType) throws IOException {
        try (ByteBufferOutputStream compressedOut = new ByteBufferOutputStream(512)) {
            Compression compression = Compression.of(compressionType).build();
            try (OutputStream out = compression.wrapForOutput(compressedOut, RecordBatch.CURRENT_MAGIC_VALUE)) {
                metrics.writeTo(out);
            }
            compressedOut.buffer().flip();
            return compressedOut.buffer();
        }
    }

    public static ByteBuffer decompress(ByteBuffer metrics, CompressionType compressionType) {
        Compression compression = Compression.of(compressionType).build();
        try (InputStream in = compression.wrapForInput(metrics, RecordBatch.CURRENT_MAGIC_VALUE, BufferSupplier.create());

Review comment (suggestion):
According to the source code (https://github.com/apache/kafka/blob/9a2f202a1e99a807b6ba91bb3c4b9148089f0b02/clients/src/main/java/org/apache/kafka/common/telemetry/internals/ClientTelemetryReporter.java#L953), it appears the `acceptedCompressionTypes` could never be null.

Code after:

        String val = (String) m.get(key);
        if (val.isEmpty()) {
            log.trace("{} is empty string. value = {} map {}", key, val, m);
            return false;
        }
        return true;
    }

    public static boolean validateRequiredResourceLabels(Map<String, String> metadata) {
        return validateResourceLabel(metadata, MetricsContext.NAMESPACE);
    }

    /**
     * Determines the preferred compression type from broker-accepted types, avoiding unsupported ones.
     * 
     * @param acceptedCompressionTypes the list of compression types accepted by the broker in order 
     *                                of preference (must not be null, use empty list if no compression is accepted)
     * @param unsupportedCompressionTypes the set of compression types that should be avoided due to 
     *                                   missing libraries or previous failures (must not be null)
     * @return the preferred compression type to use, or {@link CompressionType#NONE} if no acceptable
     *         compression type is available
     */
    public static CompressionType preferredCompressionType(List<CompressionType> acceptedCompressionTypes, Set<CompressionType> unsupportedCompressionTypes) {
        // Broker is providing the compression types in order of preference. Grab the
        // first one that's supported.
        return acceptedCompressionTypes.stream()
                .filter(t -> !unsupportedCompressionTypes.contains(t))
                .findFirst()
                .orElse(CompressionType.NONE);
    }

    public static ByteBuffer compress(MetricsData metrics, CompressionType compressionType) throws IOException {
        try (ByteBufferOutputStream compressedOut = new ByteBufferOutputStream(512)) {
            Compression compression = Compression.of(compressionType).build();
            try (OutputStream out = compression.wrapForOutput(compressedOut, RecordBatch.CURRENT_MAGIC_VALUE)) {
                metrics.writeTo(out);
            }
            compressedOut.buffer().flip();
            return compressedOut.buffer();
        }
    }

    public static ByteBuffer decompress(ByteBuffer metrics, CompressionType compressionType) {
        Compression compression = Compression.of(compressionType).build();
        try (InputStream in = compression.wrapForInput(metrics, RecordBatch.CURRENT_MAGIC_VALUE, BufferSupplier.create());
            ByteBufferOutputStream out = new ByteBufferOutputStream(512)) {
            byte[] bytes = new byte[metrics.limit() * 2];
            int nRead;
            while ((nRead = in.read(bytes, 0, bytes.length)) != -1) {
                out.write(bytes, 0, nRead);