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

Code before:
    public static void main(String[] args) throws Exception {
        ProducerPerformance perf = new ProducerPerformance();
        perf.start(args);
    }

    void start(String[] args) throws IOException {
        ArgumentParser parser = argParser();

        try {
            ConfigPostProcessor config = new ConfigPostProcessor(parser, args);
            KafkaProducer<byte[], byte[]> producer = createKafkaProducer(config.producerProps);

            if (config.transactionsEnabled)
                producer.initTransactions();

            /* setup perf test */
            byte[] payload = null;
            if (config.recordSize != null) {
                payload = new byte[config.recordSize];
            }
            // not thread-safe, do not share with other threads
            SplittableRandom random = new SplittableRandom(0);
            ProducerRecord<byte[], byte[]> record;
            if (config.warmupRecords > 0) {
                // TODO: Keep this message? Maybe unnecessary
                System.out.println("Warmup first " + config.warmupRecords + " records. Steady-state results will print after the complete-test summary.");
                this.warmupStats = new Stats(config.warmupRecords, 5000);
            } else {
                stats = new Stats(config.numRecords, 5000);
            }
            long startMs = System.currentTimeMillis();

            ThroughputThrottler throttler = new ThroughputThrottler(config.throughput, startMs);

            int currentTransactionSize = 0;
            long transactionStartTime = 0;
            for (long i = 0; i < config.numRecords; i++) {

                payload = generateRandomPayload(config.recordSize, config.payloadByteList, payload, random, config.payloadMonotonic, i);

                if (config.transactionsEnabled && currentTransactionSize == 0) {
                    producer.beginTransaction();
                    transactionStartTime = System.currentTimeMillis();
                }

                record = new ProducerRecord<>(config.topicName, payload);

                long sendStartMs = System.currentTimeMillis();
                if (warmupStats != null) {
                    if (i < config.warmupRecords) {
                        cb = new PerfCallback(sendStartMs, payload.length, warmupStats);

Review comment (suggestion):
Using the tool I changed my mind about this. It's useful to know that the tool is running the warmup, but we would also need a similar log for the steady state phase, otherwise you wouldn't be able to distinguish if the periodic log is about warmup or steady state.

Code after:

    public static void main(String[] args) throws Exception {
        ProducerPerformance perf = new ProducerPerformance();
        perf.start(args);
    }

    void start(String[] args) throws IOException {
        ArgumentParser parser = argParser();

        try {
            ConfigPostProcessor config = new ConfigPostProcessor(parser, args);
            KafkaProducer<byte[], byte[]> producer = createKafkaProducer(config.producerProps);

            if (config.transactionsEnabled)
                producer.initTransactions();

            /* setup perf test */
            byte[] payload = null;
            if (config.recordSize != null) {
                payload = new byte[config.recordSize];
            }
            // not thread-safe, do not share with other threads
            SplittableRandom random = new SplittableRandom(0);
            ProducerRecord<byte[], byte[]> record;

            if (config.warmupRecords > 0) {
                System.out.println("Warmup first " + config.warmupRecords + " records. Steady state results will print after the complete test summary.");
            }
            boolean isSteadyState = false;
            stats = new Stats(config.numRecords, isSteadyState);
            long startMs = System.currentTimeMillis();

            ThroughputThrottler throttler = new ThroughputThrottler(config.throughput, startMs);

            int currentTransactionSize = 0;
            long transactionStartTime = 0;
            for (long i = 0; i < config.numRecords; i++) {

                payload = generateRandomPayload(config.recordSize, config.payloadByteList, payload, random, config.payloadMonotonic, i);

                if (config.transactionsEnabled && currentTransactionSize == 0) {
                    producer.beginTransaction();
                    transactionStartTime = System.currentTimeMillis();
                }

                record = new ProducerRecord<>(config.topicName, payload);

                long sendStartMs = System.currentTimeMillis();
                if ((isSteadyState = config.warmupRecords > 0) && i == config.warmupRecords) {
                    steadyStateStats = new Stats(config.numRecords - config.warmupRecords, isSteadyState);
                    stats.suppressPrinting();