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

Code before:
                joinStartMs, joinTimeMsInSingleRound);
            long endMs = System.currentTimeMillis();

            Map<MetricName, ? extends Metric> metrics = null;
            if (options.printMetrics())
                metrics = consumer.metrics();
            consumer.close();

            // print final stats
            double elapsedSec = (endMs - startMs) / 1_000.0;
            long fetchTimeInMs = (endMs - startMs) - joinTimeMs.get();
            if (!options.showDetailedStats()) {
                double totalMbRead = (totalBytesRead.get() * 1.0) / (1024 * 1024);
                System.out.printf("%s, %s, %.4f, %.4f, %d, %.4f, %d, %d, %.4f, %.4f%n",
                    options.dateFormat().format(startMs),
                    options.dateFormat().format(endMs),
                    totalMbRead,
                    totalMbRead / elapsedSec,
                    totalMessagesRead.get(),
                    totalMessagesRead.get() / elapsedSec,
                    joinTimeMs.get(),
                    fetchTimeInMs,
                    totalMbRead / (fetchTimeInMs / 1000.0),
                    totalMessagesRead.get() / (fetchTimeInMs / 1000.0)
                );
            }

            if (metrics != null)
                ToolsUtils.printMetrics(metrics);
        } catch (Throwable e) {
            System.err.println(e.getMessage());
            System.err.println(Utils.stackTrace(e));
            Exit.exit(1);
        }
    }

    protected static void printHeader(boolean showDetailedStats) {
        String newFieldsInHeader = ", rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec";
        if (!showDetailedStats)
            System.out.printf("start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec%s%n", newFieldsInHeader);
        else
            System.out.printf("time, threadId, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec%s%n", newFieldsInHeader);
    }

    private static void consume(KafkaConsumer<byte[], byte[]> consumer,
                                ConsumerPerfOptions options,
                                AtomicLong totalMessagesRead,
                                AtomicLong totalBytesRead,
                                AtomicLong joinTimeMs,
                                long bytesRead,
                                long messagesRead,

Review comment (performance):
Great find. It looks like `ShareConsumerPerformance` has a similar issue. @kirktrue do you have some time to fix it as well

https://github.com/apache/kafka/blob/4271fd8c8b17cc7f31be027a7c25fd31ea11a7d7/tools/src/main/java/org/apache/kafka/tools/ShareConsumerPerformance.java#L86

Code after:
                long startMs = currentTimeMs;
                consume(consumer, options, totalMessagesRead, totalBytesRead, joinTimeMs,
                    bytesRead, messagesRead, lastBytesRead, lastMessagesRead,
                    joinStartMs, joinTimeMsInSingleRound);
                long endMs = System.currentTimeMillis();

                // print final stats
                double elapsedSec = (endMs - startMs) / 1_000.0;
                long fetchTimeInMs = (endMs - startMs) - joinTimeMs.get();
                if (!options.showDetailedStats()) {
                    double totalMbRead = (totalBytesRead.get() * 1.0) / (1024 * 1024);
                    System.out.printf("%s, %s, %.4f, %.4f, %d, %.4f, %d, %d, %.4f, %.4f%n",
                        options.dateFormat().format(startMs),
                        options.dateFormat().format(endMs),
                        totalMbRead,
                        totalMbRead / elapsedSec,
                        totalMessagesRead.get(),
                        totalMessagesRead.get() / elapsedSec,
                        joinTimeMs.get(),
                        fetchTimeInMs,
                        totalMbRead / (fetchTimeInMs / 1000.0),
                        totalMessagesRead.get() / (fetchTimeInMs / 1000.0)
                    );
                }

                if (options.printMetrics()) {
                    ToolsUtils.printMetrics(consumer.metrics());
                }
            }
        } catch (Throwable e) {
            System.err.println(e.getMessage());
            System.err.println(Utils.stackTrace(e));
            Exit.exit(1);
        }
    }

    protected static void printHeader(boolean showDetailedStats) {
        String newFieldsInHeader = ", rebalance.time.ms, fetch.time.ms, fetch.MB.sec, fetch.nMsg.sec";
        if (!showDetailedStats)
            System.out.printf("start.time, end.time, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec%s%n", newFieldsInHeader);
        else
            System.out.printf("time, threadId, data.consumed.in.MB, MB.sec, data.consumed.in.nMsg, nMsg.sec%s%n", newFieldsInHeader);
    }

    private static void consume(Consumer<byte[], byte[]> consumer,
                                ConsumerPerfOptions options,
                                AtomicLong totalMessagesRead,
                                AtomicLong totalBytesRead,
                                AtomicLong joinTimeMs,
                                long bytesRead,
                                long messagesRead,