public static void createTopics(
        Logger log, String bootstrapServers, Map<String, String> commonClientConf,
        Map<String, String> adminClientConf,
        Map<String, NewTopic> topics, boolean failOnExisting) throws Throwable {
        // this method wraps the call to createTopics() that takes admin client, so that we can
        // unit test the functionality with MockAdminClient. The exception is caught and
        // re-thrown so that admin client is closed when the method returns.
        try (Admin adminClient
                 = createAdminClient(bootstrapServers, commonClientConf, adminClientConf)) {
            createTopics(log, adminClient, topics, failOnExisting);
        } catch (Exception e) {
            log.warn("Failed to create or verify topics {}", topics, e);
            throw e;
        }
    }
Create some Kafka topics.

@param log               The logger to use.
@param bootstrapServers  The bootstrap server list.
@param commonClientConf  Common client config
@param adminClientConf   AdminClient config. This config has precedence over fields in
                         common client config.
@param topics            Maps topic names to partition assignments.
@param failOnExisting    If true, the method will throw TopicExistsException if one or
                         more topics already exist. Otherwise, the existing topics are
                         verified for number of partitions. In this case, if number of
                         partitions of an existing topic does not match the requested
                         number of partitions, the method throws RuntimeException.
trogdor/src/main/java/org/apache/kafka/trogdor/common/WorkerUtils.java