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

Code before:
                    ProducerRecord<byte[], byte[]> record;
                    if (delete) {
                        record = new ProducerRecord<>(topic,
                                String.valueOf(key).getBytes(StandardCharsets.UTF_8), null);
                    } else {
                        record = new ProducerRecord<>(topic,
                                String.valueOf(key).getBytes(StandardCharsets.UTF_8),
                                String.valueOf(i).getBytes(StandardCharsets.UTF_8));
                    }
                    producer.send(record);
                    producedWriter.write(new TestRecord(topic, key, i, delete).toString());
                    producedWriter.newLine();
                }
            }
            return producedFilePath;
        }
    }

    private static Path consumeMessages(String brokerUrl, Set<String> topics) throws IOException {

        Path consumedFilePath = Files.createTempFile("kafka-log-cleaner-consumed-", ".txt");
        System.out.println("Logging consumed messages to " + consumedFilePath);

        try (Consumer<String, String> consumer = createConsumer(brokerUrl);
             BufferedWriter consumedWriter = Files.newBufferedWriter(consumedFilePath, StandardCharsets.UTF_8)) {
            consumer.subscribe(new ArrayList<>(topics));
            while (true) {
                ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(20));
                if (consumerRecords.isEmpty()) return consumedFilePath;
                consumerRecords.forEach(
                    record -> {
                        try {
                            boolean delete = record.value() == null;
                            long value = delete ? -1L : Long.parseLong(record.value());
                            TestRecord testRecord = new TestRecord(
                                    record.topic(), Integer.parseInt(record.key()), value, delete);
                            consumedWriter.write(testRecord.toString());
                            consumedWriter.newLine();
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                    }
                );
            }
        }
    }

    private static Consumer<String, String> createConsumer(String brokerUrl) {
        Map<String, Object> consumerProps = Map.of(
                ConsumerConfig.GROUP_ID_CONFIG, "log-cleaner-test-" + RANDOM.nextInt(Integer.MAX_VALUE),
                ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl,

Review comment (suggestion):
We can use `consumer.subscribe(topics)` directly;


Code after:
                    boolean delete = (i % 100) < percentDeletes;
                    ProducerRecord<byte[], byte[]> record;
                    if (delete) {
                        record = new ProducerRecord<>(topic,
                                String.valueOf(key).getBytes(StandardCharsets.UTF_8), null);
                    } else {
                        record = new ProducerRecord<>(topic,
                                String.valueOf(key).getBytes(StandardCharsets.UTF_8),
                                String.valueOf(i).getBytes(StandardCharsets.UTF_8));
                    }
                    producer.send(record);
                    producedWriter.write(new TestRecord(topic, key, i, delete).toString());
                    producedWriter.newLine();
                }
            }
            return producedFilePath;
        }
    }

    private static Path consumeMessages(String brokerUrl, Set<String> topics) throws IOException {

        Path consumedFilePath = Files.createTempFile("kafka-log-cleaner-consumed-", ".txt");
        System.out.println("Logging consumed messages to " + consumedFilePath);

        try (Consumer<String, String> consumer = createConsumer(brokerUrl);
             BufferedWriter consumedWriter = Files.newBufferedWriter(consumedFilePath, StandardCharsets.UTF_8)) {
            consumer.subscribe(topics);
            while (true) {
                ConsumerRecords<String, String> consumerRecords = consumer.poll(Duration.ofSeconds(20));
                if (consumerRecords.isEmpty()) return consumedFilePath;
                consumerRecords.forEach(
                    record -> {
                        try {
                            boolean delete = record.value() == null;
                            long value = delete ? -1L : Long.parseLong(record.value());
                            TestRecord testRecord = new TestRecord(
                                    record.topic(), Integer.parseInt(record.key()), value, delete);
                            consumedWriter.write(testRecord.toString());
                            consumedWriter.newLine();
                        } catch (IOException e) {
                            throw new RuntimeException(e);
                        }
                    }
                );
            }
        }
    }

    private static Consumer<String, String> createConsumer(String brokerUrl) {
        Map<String, Object> consumerProps = Map.of(
                ConsumerConfig.GROUP_ID_CONFIG, "log-cleaner-test-" + RANDOM.nextInt(Integer.MAX_VALUE),