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

Code before:

        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);
        String brokerUrl = optionSet.valueOf(options.brokerOpt);
        int topicCount = optionSet.valueOf(options.topicsOpt);
        int sleepSecs = optionSet.valueOf(options.sleepSecsOpt);

        CompressionType.forName(compressionType.toLowerCase(Locale.ROOT));

        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");

Review comment (suggestion):
`compressionType` could be changed from `string` to `CompressionType`. That also validates the string value

Code after:
        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);
        CompressionType compressionType = CompressionType.forName(optionSet.valueOf(options.messageCompressionOpt));
        Integer compressionLevel = optionSet.valueOf(options.compressionLevelOpt);
        int percentDeletes = optionSet.valueOf(options.percentDeletesOpt);
        int dups = optionSet.valueOf(options.numDupsOpt);
        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 {