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 testCreateOneTopic() throws Throwable {
        Map<String, NewTopic> newTopics = Collections.singletonMap(TEST_TOPIC, NEW_TEST_TOPIC);

        WorkerUtils.createTopics(log, adminClient, newTopics, true);
        assertEquals(Collections.singleton(TEST_TOPIC), adminClient.listTopics().names().get());
        assertEquals(
            new TopicDescription(
                TEST_TOPIC, false,
                Collections.singletonList(
                    new TopicPartitionInfo(0, broker1, singleReplica, Collections.emptyList(), Collections.emptyList(), Collections.emptyList()))),
            adminClient.describeTopics(
                Collections.singleton(TEST_TOPIC)).topicNameValues().get(TEST_TOPIC).get()
        );
    }