private static Collection<String> createTopics(Logger log, Admin adminClient,
                                                   Collection<NewTopic> topics) throws Throwable {
        long startMs = Time.SYSTEM.milliseconds();
        int tries = 0;
        List<String> existingTopics = new ArrayList<>();

        Map<String, NewTopic> newTopics = new HashMap<>();
        for (NewTopic newTopic : topics) {
            newTopics.put(newTopic.name(), newTopic);
        }
        List<String> topicsToCreate = new ArrayList<>(newTopics.keySet());
        while (true) {
            log.info("Attempting to create {} topics (try {})...", topicsToCreate.size(), ++tries);
            Map<String, Future<Void>> creations = new HashMap<>();
            while (!topicsToCreate.isEmpty()) {
                List<NewTopic> newTopicsBatch = new ArrayList<>();
                for (int i = 0; (i < MAX_CREATE_TOPICS_BATCH_SIZE) &&
                                !topicsToCreate.isEmpty(); i++) {
                    String topicName = topicsToCreate.remove(0);
                    newTopicsBatch.add(newTopics.get(topicName));
                }
                creations.putAll(adminClient.createTopics(newTopicsBatch).values());
            }
            // We retry cases where the topic creation failed with a
            // timeout.  This is a workaround for KAFKA-6368.
            for (Map.Entry<String, Future<Void>> entry : creations.entrySet()) {
                String topicName = entry.getKey();
                Future<Void> future = entry.getValue();
                try {
                    future.get();
                    log.debug("Successfully created {}.", topicName);
                } catch (Exception e) {
                    if ((e.getCause() instanceof TimeoutException)
                        || (e.getCause() instanceof NotEnoughReplicasException)) {
                        log.warn("Attempt to create topic `{}` failed: {}", topicName,
                                 e.getCause().getMessage());
                        topicsToCreate.add(topicName);
                    } else if (e.getCause() instanceof TopicExistsException) {
                        log.info("Topic {} already exists.", topicName);
                        existingTopics.add(topicName);
                    } else {
                        log.warn("Failed to create {}", topicName, e.getCause());
                        throw e.getCause();
                    }
                }
            }
            if (topicsToCreate.isEmpty()) {
                break;
            }
            if (Time.SYSTEM.milliseconds() > startMs + CREATE_TOPICS_CALL_TIMEOUT) {
                String str = "Unable to create topic(s): " +
                             String.join(", ", topicsToCreate) + "after " + tries + " attempt(s)";
                log.warn(str);
                throw new TimeoutException(str);
            }
        }
        return existingTopics;
    }
Creates Kafka topics and returns a list of topics that already exist
@param log             The logger to use
@param adminClient     AdminClient
@param topics          List of topics to create
@return                Collection of topics names that already exist.
@throws Throwable if creation of one or more topics fails (except for topic exists case).
trogdor/src/main/java/org/apache/kafka/trogdor/common/WorkerUtils.java