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 testExistingTopicsMustHaveRequestedNumberOfPartitions() {
        List<TopicPartitionInfo> tpInfo = new ArrayList<>();
        tpInfo.add(new TopicPartitionInfo(0, broker1, singleReplica, Collections.emptyList()));
        tpInfo.add(new TopicPartitionInfo(1, broker2, singleReplica, Collections.emptyList()));
        adminClient.addTopic(
            false,
            TEST_TOPIC,
            tpInfo,
            null);

        assertThrows(RuntimeException.class, () -> WorkerUtils.createTopics(
            log, adminClient, Collections.singletonMap(TEST_TOPIC, NEW_TEST_TOPIC), false));
    }