static void verifyTopics(
        Logger log, Admin adminClient,
        Collection<String> topicsToVerify, Map<String, NewTopic> topicsInfo, int retryCount, long retryBackoffMs) throws Throwable {

        Map<String, TopicDescription> topicDescriptionMap = topicDescriptions(topicsToVerify, adminClient,
                retryCount, retryBackoffMs);

        for (TopicDescription desc: topicDescriptionMap.values()) {
            // map will always contain the topic since all topics in 'topicsExists' are in given
            // 'topics' map
            int partitions = topicsInfo.get(desc.name()).numPartitions();
            if (partitions != CreateTopicsRequest.NO_NUM_PARTITIONS && desc.partitions().size() != partitions) {
                String str = "Topic '" + desc.name() + "' exists, but has "
                             + desc.partitions().size() + " partitions, while requested "
                             + " number of partitions is " + partitions;
                log.warn(str);
                throw new RuntimeException(str);
            }
        }
    }
@Test
    public void testVerifyTopics() throws Throwable {
        Map<String, NewTopic> newTopics = Collections.singletonMap(TEST_TOPIC, NEW_TEST_TOPIC);
        WorkerUtils.createTopics(log, adminClient, newTopics, true);
        adminClient.setFetchesRemainingUntilVisible(TEST_TOPIC, 2);
        WorkerUtils.verifyTopics(log, adminClient, Collections.singleton(TEST_TOPIC),
            Collections.singletonMap(TEST_TOPIC, NEW_TEST_TOPIC), 3, 1);
        adminClient.setFetchesRemainingUntilVisible(TEST_TOPIC, 100);
        assertThrows(UnknownTopicOrPartitionException.class, () ->
            WorkerUtils.verifyTopics(log, adminClient, Collections.singleton(TEST_TOPIC),
                Collections.singletonMap(TEST_TOPIC, NEW_TEST_TOPIC), 2, 1));
    }