Repository: apache/kafka
File: tools/src/main/java/org/apache/kafka/tools/LogCompactionTester.java (Java)

Code before:
        );
    }

    private static long lineCount(Path filePath) throws IOException {
        try (Stream<String> lines = Files.lines(filePath)) {
            return lines.count();
        }
    }

    private static void require(boolean requirement, String message) {
        if (!requirement) {
            System.err.println("Data validation failed : " + message);
            Exit.exit(1);
        }
    }

    private static Path produceMessages(String brokerUrl, Set<String> topics, long messages,
                                        CompressionType compressionType, Integer compressionLevel,
                                        int dups, int percentDeletes) throws IOException {
        Map<String, Object> producerProps = new HashMap<>();
        producerProps.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, String.valueOf(Long.MAX_VALUE));
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
        producerProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compressionType.name);
        
        if (compressionLevel != null) {
            switch (compressionType.name.toLowerCase(Locale.ROOT)) {
                case "gzip":
                    producerProps.put(ProducerConfig.COMPRESSION_GZIP_LEVEL_CONFIG, compressionLevel);
                    break;
                case "lz4":
                    producerProps.put(ProducerConfig.COMPRESSION_LZ4_LEVEL_CONFIG, compressionLevel);
                    break;
                case "zstd":
                    producerProps.put(ProducerConfig.COMPRESSION_ZSTD_LEVEL_CONFIG, compressionLevel);
                    break;
                default:
                    System.out.println("Warning: Compression level " + compressionLevel + " is ignored for compression type " + compressionType.name + ". Only gzip, lz4, and zstd support compression levels.");
                    break;
            }
        }

        try (KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(
                producerProps, new ByteArraySerializer(), new ByteArraySerializer())) {
            int keyCount = (int) (messages / dups);
            Path producedFilePath = Files.createTempFile("kafka-log-cleaner-produced-", ".txt");
            System.out.println("Logging produce requests to " + producedFilePath);

            try (BufferedWriter producedWriter = Files.newBufferedWriter(
                    producedFilePath, StandardCharsets.UTF_8)) {
                List<String> topicsList = List.copyOf(topics);
                int size = topicsList.size();

Review comment (suggestion):
```java
        if (compressionLevel != null) {
            switch (compressionType) {
                case GZIP -> producerProps.put(ProducerConfig.COMPRESSION_GZIP_LEVEL_CONFIG, compressionLevel);
                case LZ4 -> producerProps.put(ProducerConfig.COMPRESSION_LZ4_LEVEL_CONFIG, compressionLevel);
                case ZSTD -> producerProps.put(ProducerConfig.COMPRESSION_ZSTD_LEVEL_CONFIG, compressionLevel);
                default -> System.out.println("Warning: Compression level " + compressionLevel + " is ignored for compression type "
                        + compressionType.name + ". Only gzip, lz4, and zstd support compression levels.");
            }
        }
```

Code after:
    }

    private static long lineCount(Path filePath) throws IOException {
        try (Stream<String> lines = Files.lines(filePath)) {
            return lines.count();
        }
    }

    private static void require(boolean requirement, String message) {
        if (!requirement) {
            System.err.println("Data validation failed : " + message);
            Exit.exit(1);
        }
    }

    private static Path produceMessages(String brokerUrl, Set<String> topics, long messages,
                                        CompressionType compressionType, Integer compressionLevel,
                                        int dups, int percentDeletes) throws IOException {
        Map<String, Object> producerProps = new HashMap<>();
        producerProps.put(ProducerConfig.MAX_BLOCK_MS_CONFIG, String.valueOf(Long.MAX_VALUE));
        producerProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
        producerProps.put(ProducerConfig.COMPRESSION_TYPE_CONFIG, compressionType.name);
        
        if (compressionLevel != null) {
            switch (compressionType) {
                case GZIP -> producerProps.put(ProducerConfig.COMPRESSION_GZIP_LEVEL_CONFIG, compressionLevel);
                case LZ4 -> producerProps.put(ProducerConfig.COMPRESSION_LZ4_LEVEL_CONFIG, compressionLevel);
                case ZSTD -> producerProps.put(ProducerConfig.COMPRESSION_ZSTD_LEVEL_CONFIG, compressionLevel);
                default -> System.out.println("Warning: Compression level " + compressionLevel + " is ignored for compression type "
                    + compressionType.name + ". Only gzip, lz4, and zstd support compression levels.");
            }
        }

        try (KafkaProducer<byte[], byte[]> producer = new KafkaProducer<>(
                producerProps, new ByteArraySerializer(), new ByteArraySerializer())) {
            int keyCount = (int) (messages / dups);
            Path producedFilePath = Files.createTempFile("kafka-log-cleaner-produced-", ".txt");
            System.out.println("Logging produce requests to " + producedFilePath);

            try (BufferedWriter producedWriter = Files.newBufferedWriter(
                    producedFilePath, StandardCharsets.UTF_8)) {
                List<String> topicsList = List.copyOf(topics);
                int size = topicsList.size();
                for (long i = 0; i < messages * size; i++) {
                    String topic = topicsList.get((int) (i % size));
                    int key = RANDOM.nextInt(keyCount);
                    boolean delete = (i % 100) < percentDeletes;
                    ProducerRecord<byte[], byte[]> record;
                    if (delete) {
                        record = new ProducerRecord<>(topic,
                                String.valueOf(key).getBytes(StandardCharsets.UTF_8), null);