public List<Topic> createTopics(TopicsConfig config) {
        List<NewTopic> newTopics = IntStream.range(0, config.topics)
                .mapToObj(i -> generateTopicName(config.topicPrefix, config.partitionsPerTopic, i))
                .map(name -> new NewTopic(name, config.partitionsPerTopic, (short) 1).configs(config.topicConfigs))
                .collect(Collectors.toList());

        int topicsPerBatch = MAX_PARTITIONS_PER_BATCH / config.partitionsPerTopic;
        List<List<NewTopic>> requests = Lists.partition(newTopics, topicsPerBatch);

        Map<String, KafkaFuture<Void>> results = requests.stream()
                .map(admin::createTopics)
                .map(CreateTopicsResult::values)
                .flatMap(map -> map.entrySet().stream())
                .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
        results.forEach(this::waitTopicCreated);

        return results.keySet().stream()
                .map(name -> new Topic(name, config.partitionsPerTopic))
                .collect(Collectors.toList());
    }
Create topics with the given configuration.
Note: If the topic already exists, it will not be created again.
tools/src/main/java/org/apache/kafka/tools/automq/perf/TopicService.java