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

Code before:
        }

        String normalizedType = compressionType.toLowerCase(Locale.ROOT);
        switch (normalizedType) {
            case "none":
            case "snappy":
                System.err.println("Compression level not supported for " + compressionType);
                System.err.println("Remove --compression-level option");
                Exit.exit(1);
                break;
            case "gzip":
                validateLevelRange(compressionLevel, CompressionType.GZIP, "gzip");
                break;
            case "lz4":
                validateLevelRange(compressionLevel, CompressionType.LZ4, "lz4");
                break;
            case "zstd":
                validateLevelRange(compressionLevel, CompressionType.ZSTD, "zstd");
                break;
            default:
                // This should never happen since compression type is validated first
                throw new IllegalStateException("Unknown compression type: " + compressionType);
        }
    }

    private static void validateLevelRange(int level, CompressionType type, String typeName) {
        if (level < type.minLevel() || level > type.maxLevel()) {
            System.err.println("Invalid " + typeName + " compression level: " + level);
            System.err.println("Valid " + typeName + " levels: " + type.minLevel() + "-" + type.maxLevel());
            Exit.exit(1);
        }
    }

    public static void main(String[] args) throws Exception {

        OptionParser parser = new OptionParser(false);
        Options options = new Options(parser);

        OptionSet optionSet = parser.parse(args);
        if (args.length == 0) {
            CommandLineUtils.printUsageAndExit(parser,
                    "A tool to test log compaction. Valid options are: ");
        }

        CommandLineUtils.checkRequiredArgs(parser, optionSet, options.brokerOpt, options.numMessagesOpt);

        long messages = optionSet.valueOf(options.numMessagesOpt);
        String compressionType = optionSet.valueOf(options.messageCompressionOpt);
        Integer compressionLevel = optionSet.valueOf(options.compressionLevelOpt);
        int percentDeletes = optionSet.valueOf(options.percentDeletesOpt);
        int dups = optionSet.valueOf(options.numDupsOpt);

Review comment (refactor):
this validation is good, but `CompressionType` already cover the checks. Perhaps the duplicate checks are not necessary.

Code after:
        String brokerUrl = optionSet.valueOf(options.brokerOpt);
        int topicCount = optionSet.valueOf(options.topicsOpt);
        int sleepSecs = optionSet.valueOf(options.sleepSecsOpt);

        long testId = RANDOM.nextLong();
        Set<String> topics = IntStream.range(0, topicCount)
                .mapToObj(i -> "log-cleaner-test-" + testId + "-" + i)
                .collect(toCollection(LinkedHashSet::new));
        createTopics(brokerUrl, topics);

        System.out.println("Producing " + messages + " messages..to topics " + String.join(",", topics));
        Path producedDataFilePath = produceMessages(
                brokerUrl, topics, messages,
                compressionType, compressionLevel,
                dups, percentDeletes);
        System.out.println("Sleeping for " + sleepSecs + "seconds...");
        TimeUnit.MILLISECONDS.sleep(sleepSecs * 1000L);
        System.out.println("Consuming messages...");
        Path consumedDataFilePath = consumeMessages(brokerUrl, topics);

        long producedLines = lineCount(producedDataFilePath);
        long consumedLines = lineCount(consumedDataFilePath);
        double reduction = 100 * (1.0 - (double) consumedLines / producedLines);

        System.out.printf(
            "%d rows of data produced, %d rows of data consumed (%.1f%% reduction).%n",
            producedLines, consumedLines, reduction);

        System.out.println("De-duplicating and validating output files...");
        validateOutput(producedDataFilePath.toFile(), consumedDataFilePath.toFile());

        Files.deleteIfExists(producedDataFilePath);
        Files.deleteIfExists(consumedDataFilePath);
        // if you change this line, we need to update test_log_compaction_tool.py system test
        System.out.println("Data verification is completed");
    }


    private static void createTopics(String brokerUrl, Set<String> topics) throws Exception {
        Properties adminConfig = new Properties();
        adminConfig.put(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);

        try (Admin adminClient = Admin.create(adminConfig)) {
            Map<String, String> topicConfigs = Map.of(
                    TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT
            );
            List<NewTopic> newTopics = topics.stream()
                    .map(name -> new NewTopic(name, 1, (short) 1).configs(topicConfigs)).toList();
            adminClient.createTopics(newTopics).all().get();

            final List<String> pendingTopics = new ArrayList<>();