private void run() {
        LOGGER.info("Starting perf test with config: {}", jsonStringify(config));
        TimerUtil timer = new TimerUtil();

        if (config.reset) {
            LOGGER.info("Deleting all test topics...");
            int deleted = topicService.deleteTopics();
            LOGGER.info("Deleted all test topics ({} in total), took {} ms", deleted, timer.elapsedAndResetAs(TimeUnit.MILLISECONDS));
        }

        List<Topic> topics;
        if (config.catchupTopicPrefix != null && !config.catchupTopicPrefix.isEmpty()) {
            LOGGER.info("Looking for catch-up topics with prefix: {}", config.catchupTopicPrefix);
            topics = topicService.findExistingTopicsByPrefix(config.catchupTopicPrefix);
            if (topics.isEmpty()) {
                throw new RuntimeException("No catch-up topics found with prefix: " + config.catchupTopicPrefix);
            }
            LOGGER.info("Found {} catch-up topics with prefix '{}'.", topics.size(), config.catchupTopicPrefix);
        } else if (config.reuseTopics) {
            LOGGER.info("Reusing existing topics with prefix: {}", config.topicPrefix);
            topics = topicService.findExistingTopicsByPrefix(config.topicPrefix);
            if (topics.isEmpty()) {
                LOGGER.warn("No existing topics found with prefix '{}', creating new topics instead.", config.topicPrefix);
                topics = topicService.createTopics(config.topicsConfig());
                LOGGER.info("Created {} topics, took {} ms", topics.size(), timer.elapsedAndResetAs(TimeUnit.MILLISECONDS));
            } else {
                LOGGER.info("Found {} existing topics with prefix '{}'.", topics.size(), config.topicPrefix);
            }
        } else {
            LOGGER.info("Creating topics...");
            topics = topicService.createTopics(config.topicsConfig());
            LOGGER.info("Created {} topics, took {} ms", topics.size(), timer.elapsedAndResetAs(TimeUnit.MILLISECONDS));
        }

        LOGGER.info("Creating consumers...");
        int consumers = consumerService.createConsumers(topics, config.consumersConfig());
        consumerService.start(this::messageReceived, config.maxConsumeRecordRate);
        LOGGER.info("Created {} consumers, took {} ms", consumers, timer.elapsedAndResetAs(TimeUnit.MILLISECONDS));

        LOGGER.info("Creating producers...");
        int producers = producerService.createProducers(topics, config.producersConfig(), this::messageSent);
        LOGGER.info("Created {} producers, took {} ms", producers, timer.elapsedAndResetAs(TimeUnit.MILLISECONDS));

        if (config.catchupTopicPrefix != null && !config.catchupTopicPrefix.isEmpty()) {
            LOGGER.info("Using catch-up topics, skipping message accumulation phase");
            preparing = false; // Directly start consuming without accumulation
        } else if (config.awaitTopicReady) {
            LOGGER.info("Waiting for topics to be ready...");
            waitTopicsReady(consumerService.consumerCount() > 0);
            LOGGER.info("Topics are ready, took {} ms", timer.elapsedAndResetAs(TimeUnit.MILLISECONDS));

            Function<String, List<byte[]>> payloads = payloads(config, topics);
            producerService.start(payloads, config.sendRate);
            preparing = false;

            if (config.warmupDurationMinutes > 0) {
                LOGGER.info("Warming up for {} minutes...", config.warmupDurationMinutes);
                long warmupStart = System.nanoTime();
                long warmupMiddle = warmupStart + TimeUnit.MINUTES.toNanos(config.warmupDurationMinutes) / 2;
                producerService.adjustRate(warmupStart, ProducerService.MIN_RATE);
                producerService.adjustRate(warmupMiddle, config.sendRate);
                collectStats(Duration.ofMinutes(config.warmupDurationMinutes));
            }
        } else {
            // If not waiting for topic ready and not using catchup topics, start producing immediately.
            Function<String, List<byte[]>> payloads = payloads(config, topics);
            producerService.start(payloads, config.sendRate);
            preparing = false;
        }

        Result result;
        if (config.backlogDurationSeconds > 0) {
            LOGGER.info("Pausing consumers for {} seconds to build up backlog...", config.backlogDurationSeconds);
            consumerService.pause();
            long backlogStart = System.currentTimeMillis();
            collectStats(Duration.ofSeconds(config.backlogDurationSeconds));
            long backlogEnd = System.nanoTime();

            LOGGER.info("Resetting consumer offsets and resuming...");
            consumerService.resetOffset(backlogStart, TimeUnit.SECONDS.toMillis(config.groupStartDelaySeconds));

            // Select topics for catch-up
            int numTopics = topics.size();
            int topicsToResume = (int) Math.ceil(numTopics * (config.consumersDuringCatchupPercentage / 100.0));
            topicsToResume = Math.max(1, Math.min(numTopics, topicsToResume));
            List<String> allTopicNames = topics.stream().map(t -> t.name()).collect(java.util.stream.Collectors.toList());
            java.util.Collections.shuffle(allTopicNames);
            List<String> resumeTopics = allTopicNames.subList(0, topicsToResume);

            // No need to pause all again, they're already paused from line 172
            // Just resume the selected topics
            consumerService.resumeTopics(resumeTopics);
            LOGGER.info("Resuming consumers for topics: {} ({} out of {})", resumeTopics, topicsToResume, numTopics);
            // No need to pause specific topics, they're already paused
            LOGGER.info("Keeping remaining consumers paused ({} topics)", numTopics - topicsToResume);

            stats.reset();
            producerService.adjustRate(config.sendRateDuringCatchup);
            result = collectStats(backlogEnd);
        } else {
            LOGGER.info("Running test for {} minutes...", config.testDurationMinutes);
            stats.reset();
            result = collectStats(Duration.ofMinutes(config.testDurationMinutes));
        }
        LOGGER.info("Saving results to {}", saveResult(result));

        running = false;
    }
Partitions that are ready to be consumed.
Only used during the initial topic readiness check, which is, {@link #preparing} is true.
tools/src/main/java/org/apache/kafka/tools/automq/PerfCommand.java