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

Code before:
            log.trace("{} is not a string. map {}", key, m);
            return false;
        }

        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
     * @throws NullPointerException if acceptedCompressionTypes or unsupportedCompressionTypes is null
     */
    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 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());
            ByteBufferOutputStream out = new ByteBufferOutputStream(512)) {

Review comment (bug):
Having NPE in @throws and being documented seem a bit inappropriate. You have already mentioned that non-null values are expected for these params.

Code after:
            log.trace("{} is not a string. map {}", key, m);
            return false;
        }

        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];