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

Code before:
                    }
                } else {
                    cb = new PerfCallback(sendStartMs, payload.length, stats);
                }
                producer.send(record, cb);

                currentTransactionSize++;
                if (config.transactionsEnabled && config.transactionDurationMs <= (sendStartMs - transactionStartTime)) {
                    producer.commitTransaction();
                    currentTransactionSize = 0;
                }

                if (throttler.shouldThrottle(i, sendStartMs)) {
                    throttler.throttle();
                }
            }

            if (config.transactionsEnabled && currentTransactionSize != 0)
                producer.commitTransaction();

            if (!config.shouldPrintMetrics) {
                producer.close();

                /* print warmup stats if relevant */
                if (warmupStats != null) {
                    new Stats(warmupStats, stats).printTotal();
                }
                /* print final results */
                stats.printTotal();
            } else {
                // Make sure all messages are sent before printing out the stats and the metrics
                // We need to do this in a different branch for now since tests/kafkatest/sanity_checks/test_performance_services.py
                // expects this class to work with older versions of the client jar that don't support flush().
                producer.flush();

                /* print warmup stats if relevant */
                if (warmupStats != null) {
                    new Stats(warmupStats, stats).printTotal();
                }
                /* print final results */
                stats.printTotal();

                /* print out metrics */
                ToolsUtils.printMetrics(producer.metrics());
                producer.close();
            }
        } catch (ArgumentParserException e) {
            if (args.length == 0) {
                parser.printHelp();
                Exit.exit(0);
            } else {

Review comment (suggestion):
Maybe we should have a state for 'all' and another state for 'steady-only'. With this approach, we can avoid this kind of magic merging.

Code after:
                    currentTransactionSize = 0;
                }

                if (throttler.shouldThrottle(i, sendStartMs)) {
                    throttler.throttle();
                }
            }

            if (config.transactionsEnabled && currentTransactionSize != 0)
                producer.commitTransaction();

            if (!config.shouldPrintMetrics) {
                producer.close();

                /* print final results */
                stats.printTotal();
                /* print steady-state stats if relevant */
                if (steadyStateStats != null) {
                    steadyStateStats.printTotal();
                }
            } else {
                // Make sure all messages are sent before printing out the stats and the metrics
                // We need to do this in a different branch for now since tests/kafkatest/sanity_checks/test_performance_services.py
                // expects this class to work with older versions of the client jar that don't support flush().
                producer.flush();

                /* print final results */
                stats.printTotal();
                /* print steady-state stats if relevant */
                if (steadyStateStats != null) {
                    steadyStateStats.printTotal();
                }

                /* print out metrics */
                ToolsUtils.printMetrics(producer.metrics());
                producer.close();
            }
        } catch (ArgumentParserException e) {
            if (args.length == 0) {
                parser.printHelp();
                Exit.exit(0);
            } else {
                parser.handleError(e);
                Exit.exit(1);
            }
        }

    }

    KafkaProducer<byte[], byte[]> createKafkaProducer(Properties props) {
        return new KafkaProducer<>(props);