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;
        }
    }
@Test
    public void testCreateTopicsFailsIfAtLeastOneTopicExists() {
        adminClient.addTopic(
            false,
            TEST_TOPIC,
            Collections.singletonList(new TopicPartitionInfo(0, broker1, singleReplica, Collections.emptyList())),
            null);

        Map<String, NewTopic> newTopics = new HashMap<>();
        newTopics.put(TEST_TOPIC, NEW_TEST_TOPIC);
        newTopics.put("another-topic",
                      new NewTopic("another-topic", TEST_PARTITIONS, TEST_REPLICATION_FACTOR));
        newTopics.put("one-more-topic",
                      new NewTopic("one-more-topic", TEST_PARTITIONS, TEST_REPLICATION_FACTOR));

        assertThrows(TopicExistsException.class, () -> WorkerUtils.createTopics(log, adminClient, newTopics, true));
    }