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

Code before:
     *
     * 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,
                    "MakeReady timeout",
                    String.format("Could not create topics within %d milliseconds. This can happen if the Kafka cluster is temporarily not available.", retryTimeoutMs),
                    null
                ));

            }
        }
        log.debug("Completed validating internal topics and created {}", newlyCreatedTopics);

        return newlyCreatedTopics;
    }

Review comment (suggestion):
Here we compute `topicToCreate`, but for all existing topics, we don't return them, but we also don't remove existing topics from `topicsNotReady`.

I think, we need to add a step `topicsNotReady.retainAll(notReadyTopicsMap.keySet())` after this line, plus modify `notReadyTopicsMap` inside `computeTopicsToCreate` and remove all entries for the topics we successfully verify.

Code after:
     * 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);
            }
            if (!topicsNotReady.isEmpty()) {
                maybeThrowTimeout(new TimeoutContext(
                    Collections.singleton("makeReadyCheck"), // dummy collection just to trigger if `topicsNotReady` is non-empty
                    deadlineMs,
                    "MakeReady timeout",
                    String.format("Could not create topics within %d milliseconds. This can happen if the Kafka cluster is temporarily not available.", retryTimeoutMs),
                    null
                ));

            }
        }