protected Map<String, Integer> getNumPartitions(final Set<String> topics,
                                                    final Set<String> tempUnknownTopics) {
        log.debug("Trying to check if topics {} have been created with expected number of partitions.", topics);

        final Map<String, List<TopicPartitionInfo>> topicPartitionInfo = getTopicPartitionInfo(topics, tempUnknownTopics);
        return topicPartitionInfo.entrySet().stream().collect(Collectors.toMap(
            Entry::getKey, e -> e.getValue().size()));
    }
@Test
    public void shouldThrowTimeoutExceptionIfGetNumPartitionsHasTopicDescriptionTimeout() {
        mockAdminClient.timeoutNextRequest(1);

        final InternalTopicManager internalTopicManager =
                new InternalTopicManager(time, mockAdminClient, new StreamsConfig(config));
        try {
            final Set<String> topic1set = new HashSet<String>(Arrays.asList(topic1));
            final Set<String> topic2set = new HashSet<String>(Arrays.asList(topic2));

            internalTopicManager.getNumPartitions(topic1set, topic2set);

        } catch (final TimeoutException expected) {
            assertEquals(TimeoutException.class, expected.getCause().getClass());
        }

        mockAdminClient.timeoutNextRequest(1);

        try {
            final Set<String> topic1set = new HashSet<String>(Arrays.asList(topic1));
            final Set<String> topic2set = new HashSet<String>(Arrays.asList(topic2));

            internalTopicManager.getNumPartitions(topic1set, topic2set);

        } catch (final TimeoutException expected) {
            assertEquals(TimeoutException.class, expected.getCause().getClass());
        }
    }