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

Code before:
import static org.apache.kafka.server.util.CommandLineUtils.parseKeyValueArgs;

public class ShareConsumerPerformance {
    private static final Logger LOG = LoggerFactory.getLogger(ShareConsumerPerformance.class);

    public static void main(String[] args) {
        run(args, KafkaShareConsumer::new);
    }

    static void run(String[] args, Function<Properties, ShareConsumer<byte[], byte[]>> shareConsumerCreator) {
        try {
            LOG.info("Starting share consumer/consumers...");
            ShareConsumerPerfOptions options = new ShareConsumerPerfOptions(args);
            AtomicLong totalRecordsRead = new AtomicLong(0);
            AtomicLong totalBytesRead = new AtomicLong(0);

            if (!options.hideHeader())
                printHeader();

            List<ShareConsumer<byte[], byte[]>> shareConsumers = new ArrayList<>();
            List<String> clientIds = new ArrayList<>();
            for (int i = 0; i < options.threads(); i++) {
                if (options.threads() == 1) {
                    clientIds.add(options.props().getProperty(ConsumerConfig.CLIENT_ID_CONFIG));
                    shareConsumers.add(shareConsumerCreator.apply(options.props()));
                    continue;
                }
                Properties shareConsumerProps = options.props();
                String shareConsumerClientId = options.props().getProperty(ConsumerConfig.CLIENT_ID_CONFIG) + "-" + (i + 1);
                shareConsumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, shareConsumerClientId);
                clientIds.add(shareConsumerClientId);
                shareConsumers.add(shareConsumerCreator.apply(shareConsumerProps));
            }
            long startMs = System.currentTimeMillis();
            consume(shareConsumers, options, totalRecordsRead, totalBytesRead, startMs, clientIds);
            long endMs = System.currentTimeMillis();

            List<Map<MetricName, ? extends Metric>> shareConsumersMetrics = new ArrayList<>();
            if (options.printMetrics()) {
                shareConsumers.forEach(shareConsumer -> shareConsumersMetrics.add(shareConsumer.metrics()));
            }
            shareConsumers.forEach(shareConsumer -> {
                @SuppressWarnings("UnusedLocalVariable")
                Map<TopicIdPartition, Optional<KafkaException>> ignored = shareConsumer.commitSync();
            });

            // Print final stats for share group.
            double elapsedSec = (endMs - startMs) / 1_000.0;
            long fetchTimeInMs = endMs - startMs;
            printStatsForShareGroup(totalBytesRead.get(), totalRecordsRead.get(), elapsedSec, fetchTimeInMs, startMs,
                endMs, options.dateFormat());

Review comment (suggestion):
Although both achieve the same behaviour in this specific case, perhaps `break` is the more suitable choice here

Code after:
import static org.apache.kafka.server.util.CommandLineUtils.parseKeyValueArgs;

public class ShareConsumerPerformance {
    private static final Logger LOG = LoggerFactory.getLogger(ShareConsumerPerformance.class);

    public static void main(String[] args) {
        run(args, KafkaShareConsumer::new);
    }

    static void run(String[] args, Function<Properties, ShareConsumer<byte[], byte[]>> shareConsumerCreator) {
        try {
            LOG.info("Starting share consumer/consumers...");
            ShareConsumerPerfOptions options = new ShareConsumerPerfOptions(args);
            AtomicLong totalRecordsRead = new AtomicLong(0);
            AtomicLong totalBytesRead = new AtomicLong(0);

            if (!options.hideHeader())
                printHeader();

            List<ShareConsumer<byte[], byte[]>> shareConsumers = new ArrayList<>();
            List<String> clientIds = new ArrayList<>();
            for (int i = 0; i < options.threads(); i++) {
                if (options.threads() == 1) {
                    clientIds.add(options.props().getProperty(ConsumerConfig.CLIENT_ID_CONFIG));
                    shareConsumers.add(shareConsumerCreator.apply(options.props()));
                    break;
                }
                Properties shareConsumerProps = options.props();
                String shareConsumerClientId = options.props().getProperty(ConsumerConfig.CLIENT_ID_CONFIG) + "-" + (i + 1);
                shareConsumerProps.put(ConsumerConfig.CLIENT_ID_CONFIG, shareConsumerClientId);
                clientIds.add(shareConsumerClientId);
                shareConsumers.add(shareConsumerCreator.apply(shareConsumerProps));
            }
            long startMs = System.currentTimeMillis();
            consume(shareConsumers, options, totalRecordsRead, totalBytesRead, startMs, clientIds);
            long endMs = System.currentTimeMillis();

            List<Map<MetricName, ? extends Metric>> shareConsumersMetrics = new ArrayList<>();
            if (options.printMetrics()) {
                shareConsumers.forEach(shareConsumer -> shareConsumersMetrics.add(shareConsumer.metrics()));
            }
            shareConsumers.forEach(shareConsumer -> {
                @SuppressWarnings("UnusedLocalVariable")
                Map<TopicIdPartition, Optional<KafkaException>> ignored = shareConsumer.commitSync();
            });

            // Print final stats for share group.
            double elapsedSec = (endMs - startMs) / 1_000.0;
            long fetchTimeInMs = endMs - startMs;
            printStatsForShareGroup(totalBytesRead.get(), totalRecordsRead.get(), elapsedSec, fetchTimeInMs, startMs,
                endMs, options.dateFormat());