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

Code before:
                    deadlineMs - currentWallClockMs
                );
                Utils.sleep(retryBackOffMs);
            }
        }
        log.debug("Completed describing topics");
        return topicPartitionInfo;
    }

    /**
     * Prepares a set of given internal topics.
     *
     * If a topic does not exist creates a new topic.
     * If a topic with the correct number of partitions exists ignores it.
     * If a topic exists already but has different number of partitions we fail and throw exception requesting user to reset the app before restarting again.
     * @return the set of topics which had to be newly created
     */
    public Set<String> makeReady(final Map<String, InternalTopicConfig> topics) {
        // we will do the validation / topic-creation in a loop, until we have confirmed all topics
        // have existed with the expected number of partitions, or some create topic returns fatal errors.
        log.debug("Starting to validate internal topics {} in partition assignor.", topics);

        final long currentWallClockMs = time.milliseconds();
        final long deadlineMs = currentWallClockMs + retryTimeoutMs;

        final Set<String> topicsNotReady = new HashSet<>(topics.keySet());
        final Set<String> newlyCreatedTopics = new HashSet<>();

        while (!topicsNotReady.isEmpty()) {
            final Set<String> tempUnknownTopics = new HashSet<>();

            // Only consider the not-ready subset on each iteration
            final Map<String, InternalTopicConfig> notReadyTopicsMap = topics.entrySet().stream()
                    .filter(e -> topicsNotReady.contains(e.getKey()))
                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
            final Set<NewTopic> topicsToCreate = computeTopicsToCreate(notReadyTopicsMap, tempUnknownTopics);
            final boolean noTopicsToCreate = topicsToCreate.isEmpty() && tempUnknownTopics.isEmpty();

            if (noTopicsToCreate) {
                // Nothing left to create - all remaining topics were already ready
                break;
            }
            if (!topicsToCreate.isEmpty()) {
                final Set<String> createdTopics = createTopics(topicsToCreate, topicsNotReady, deadlineMs);
                topicsNotReady.removeAll(createdTopics);
                newlyCreatedTopics.addAll(createdTopics);
            }
            if (!topicsNotReady.isEmpty()) {
                maybeThrowTimeout(new TimeoutContext(
                    Collections.singleton("makeReadyCheck"), // dummy collection just to trigger if `topicsNotReady` is non-empty
                    deadlineMs,

Review comment (suggestion):
Cf my other commend on the test. Here, he old code update `topicNotReady` and the new map only contains `topic_2` after the first validation try. -- Below in the new code, we do not make this update though.

Code after:
                    retryBackOffMs,
                    deadlineMs - currentWallClockMs
                );
                Utils.sleep(retryBackOffMs);
            }
        }
        log.debug("Completed describing topics");
        return topicPartitionInfo;
    }

    /**
     * Prepares a set of given internal topics.
     *
     * If a topic does not exist creates a new topic.
     * If a topic with the correct number of partitions exists ignores it.
     * If a topic exists already but has different number of partitions we fail and throw exception requesting user to reset the app before restarting again.
     * @return the set of topics which had to be newly created
     */
    public Set<String> makeReady(final Map<String, InternalTopicConfig> topics) {
        // we will do the validation / topic-creation in a loop, until we have confirmed all topics
        // have existed with the expected number of partitions, or some create topic returns fatal errors.
        log.debug("Starting to validate internal topics {} in partition assignor.", topics);

        final long currentWallClockMs = time.milliseconds();
        final long deadlineMs = currentWallClockMs + retryTimeoutMs;

        final Set<String> topicsNotReady = new HashSet<>(topics.keySet());
        final Set<String> newlyCreatedTopics = new HashSet<>();

        while (!topicsNotReady.isEmpty()) {
            final Set<String> tempUnknownTopics = new HashSet<>();

            // Only consider the not-ready subset on each iteration
            final Map<String, InternalTopicConfig> notReadyTopicsMap = topics.entrySet().stream()
                    .filter(e -> topicsNotReady.contains(e.getKey()))
                    .collect(Collectors.toMap(Map.Entry::getKey, Map.Entry::getValue));
            final Set<NewTopic> topicsToCreate = computeTopicsToCreate(notReadyTopicsMap, tempUnknownTopics);

            topicsNotReady.retainAll(notReadyTopicsMap.keySet());

            final boolean noTopicsToCreate = topicsToCreate.isEmpty() && tempUnknownTopics.isEmpty();

            if (noTopicsToCreate) {
                // Nothing left to create - all remaining topics were already ready
                break;
            }
            if (!topicsToCreate.isEmpty()) {
                final Set<String> createdTopics = createTopics(topicsToCreate, topicsNotReady, deadlineMs);
                topicsNotReady.removeAll(createdTopics);
                newlyCreatedTopics.addAll(createdTopics);
            }