Repository: apache/kafka
File: streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java (Java)

Code before:
        final Set<String> topicsToIdentify = new HashSet<>(topicsMap.keySet());
        final Map<String, Integer> existedTopicPartition = getNumPartitions(topicsToIdentify, tempUnknownTopics);

        final Set<String> topicsNotCreated = new HashSet<>();
        for (final String topicName : topicsToIdentify) {
            final Optional<Integer> numberOfPartitions = topicsMap.get(topicName).numberOfPartitions();
            if (numberOfPartitions.isEmpty()) {
                log.error("Found undefined number of partitions for topic {}", topicName);
                throw new StreamsException("Topic " + topicName + " number of partitions not defined");
            }
            if (existedTopicPartition.containsKey(topicName)) {
                if (!existedTopicPartition.get(topicName).equals(numberOfPartitions.get())) {
                    final String errorMsg = String.format("Existing internal topic %s has invalid partitions: " +
                            "expected: %d; actual: %d. " +
                            "Use 'org.apache.kafka.tools.StreamsResetter' tool to clean up invalid topics before processing.",
                        topicName, numberOfPartitions.get(), existedTopicPartition.get(topicName));
                    log.error(errorMsg);
                    throw new StreamsException(errorMsg);
                }
            } else {
                topicsNotCreated.add(topicName);
            }
        }

        return topicsNotCreated;
    }

    /**
     * Sets up internal topics.
     *
     * Either the given topic are all created or the method fails with an exception.
     *
     * @param topicConfigs internal topics to setup
     */
    public void setup(final Map<String, InternalTopicConfig> topicConfigs) {
        log.info("Starting to setup internal topics {}.", topicConfigs.keySet());

        final long now = time.milliseconds();
        final long deadline = now + retryTimeoutMs;

        final Map<String, Map<String, String>> streamsSideTopicConfigs = topicConfigs.values().stream()
            .collect(Collectors.toMap(
                InternalTopicConfig::name,
                topicConfig -> topicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention)
            ));
        final Set<String> createdTopics = new HashSet<>();
        final Set<String> topicStillToCreate = new HashSet<>(topicConfigs.keySet());
        while (!topicStillToCreate.isEmpty()) {
            final Set<NewTopic> newTopics = topicStillToCreate.stream()
                .map(topicName -> new NewTopic(
                        topicName,

Review comment (suggestion):
If the `if` above does not pass, ie, the number of partitions is as expected and we don't throw an error, we can remove the move the current topic from `topicsMap`, as we successfully remove it. This is the side effect which avoid that we query for the same topic multiple times.

Code after:
     * Check the existing topics to have correct number of partitions; and return the remaining topics that needs to be created
     */
    private Set<String> identifyTopicsNotCreated(final Map<String, InternalTopicConfig> topicsMap,
                                                 final Set<String> tempUnknownTopics) {

        final Set<String> topicsToIdentify = new HashSet<>(topicsMap.keySet());
        final Map<String, Integer> existedTopicPartition = getNumPartitions(topicsToIdentify, tempUnknownTopics);

        final Set<String> topicsNotCreated = new HashSet<>();
        for (final String topicName : topicsToIdentify) {
            final Optional<Integer> numberOfPartitions = topicsMap.get(topicName).numberOfPartitions();
            if (numberOfPartitions.isEmpty()) {
                log.error("Found undefined number of partitions for topic {}", topicName);
                throw new StreamsException("Topic " + topicName + " number of partitions not defined");
            }
            if (existedTopicPartition.containsKey(topicName)) {
                if (!existedTopicPartition.get(topicName).equals(numberOfPartitions.get())) {
                    final String errorMsg = String.format("Existing internal topic %s has invalid partitions: " +
                            "expected: %d; actual: %d. " +
                            "Use 'org.apache.kafka.tools.StreamsResetter' tool to clean up invalid topics before processing.",
                        topicName, numberOfPartitions.get(), existedTopicPartition.get(topicName));
                    log.error(errorMsg);
                    throw new StreamsException(errorMsg);
                } else {
                    topicsMap.remove(topicName);
                }
            } else {
                topicsNotCreated.add(topicName);
            }
        }

        return topicsNotCreated;
    }

    /**
     * Sets up internal topics.
     *
     * Either the given topic are all created or the method fails with an exception.
     *
     * @param topicConfigs internal topics to setup
     */
    public void setup(final Map<String, InternalTopicConfig> topicConfigs) {
        log.info("Starting to setup internal topics {}.", topicConfigs.keySet());

        final long now = time.milliseconds();
        final long deadline = now + retryTimeoutMs;

        final Map<String, Map<String, String>> streamsSideTopicConfigs = topicConfigs.values().stream()
            .collect(Collectors.toMap(
                InternalTopicConfig::name,
                topicConfig -> topicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention)