protected Map<String, List<TopicPartitionInfo>> getTopicPartitionInfo(final Set<String> topics,
                                                                          final Set<String> tempUnknownTopics) {
        final DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(topics);
        final Map<String, KafkaFuture<TopicDescription>> futures = describeTopicsResult.topicNameValues();

        final Map<String, List<TopicPartitionInfo>> topicPartitionInfo = new HashMap<>();
        for (final Map.Entry<String, KafkaFuture<TopicDescription>> topicFuture : futures.entrySet()) {
            final String topicName = topicFuture.getKey();
            try {
                final TopicDescription topicDescription = topicFuture.getValue().get();
                topicPartitionInfo.put(topicName, topicDescription.partitions());
            } catch (final InterruptedException fatalException) {
                // this should not happen; if it ever happens it indicate a bug
                Thread.currentThread().interrupt();
                log.error(INTERRUPTED_ERROR_MESSAGE, fatalException);
                throw new IllegalStateException(INTERRUPTED_ERROR_MESSAGE, fatalException);
            } catch (final ExecutionException couldNotDescribeTopicException) {
                final Throwable cause = couldNotDescribeTopicException.getCause();
                if (cause instanceof UnknownTopicOrPartitionException) {
                    // This topic didn't exist
                    log.debug("Topic {} is unknown or not found, hence not existed yet.\n" +
                        "Error message was: {}", topicName, cause.toString());
                } else if (cause instanceof LeaderNotAvailableException) {
                    if (tempUnknownTopics != null) {
                        tempUnknownTopics.add(topicName);
                    }
                    log.debug("The leader of topic {} is not available.\n" +
                        "Error message was: {}", topicName, cause.toString());
                } else if (cause instanceof TimeoutException) {
                    if (tempUnknownTopics != null) {
                        tempUnknownTopics.add(topicName);
                    }
                    log.debug("Describing topic {} (to get number of partitions) timed out.\n" +
                            "Error message was: {}", topicName, cause.toString());
                } else {
                    log.error("Unexpected error during topic description for {}.\n" +
                        "Error message was: {}", topicName, cause.toString());
                    throw new StreamsException(String.format("Could not create topic %s.", topicName), cause);
                }
            }
        }

        return topicPartitionInfo;
    }
Try to get the partition information for the given topics; return the partition info for topics that already exists.

Topics that were not able to get its description will simply not be returned
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java