private Function<String, List<byte[]>> payloads(PerfConfig config, List<Topic> topics) {
        if (Strings.isNullOrEmpty(config.valueSchema)) {
            List<byte[]> payloads = randomPayloads(config.recordSize, config.randomRatio, config.randomPoolSize);
            return topic -> payloads;
        } else {
            // The producer configs should contain:
            // - schema.registry.url: http://localhost:8081
            Map<String, List<byte[]>> topic2payloads = new HashMap<>();
            topics.forEach(topic -> {
                topic2payloads.put(topic.name(), schemaPayloads(topic.name(), config.valueSchema, config.valuesFile, config.producerConfigs));
            });
            return topic2payloads::get;
        }
    }
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