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

Code before:
     * 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(tempUnknownTopics);
            topicsNotReady.retainAll(notReadyTopicsMap.keySet());
            topicsToCreate.forEach(topic -> topicsNotReady.add(topic.name()));

            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);


Review comment (performance):
This seems to be unnecessarily complex? We first remove topics, and in the third line add back topics... So it seem we are removing topic we should not have remove, just to add them back?

In the end, we should only remove the topics from `topicNotReady` which do already exist.

The issue seems to be, that we want to retain both `tempUnknownTopics` and `notReadyTopicsMap.keySet()`, but calling `retainAll` twice does not do this, but we would need a call `topicsNotReady.retainAll(union(tempUnknownTopics, notReadyTopicsMap.keySet()))`?

Or we rewrite this using `removeIf` instead of the three lines you added (I believe [but not 100% sure] this line would do the trick?
```
topicsNotReady.removeIf(t -> !topicsToCreate.contains(t) && !tempUnknownTopics.contains(t));
```

Or maybe my original comment could help (or would this not work?) https://github.com/apache/kafka/pull/20326#discussion_r2512435233

I think, if we modify `topicsNotReady` there (ie, `topicMap` which is the corresponding parameter name inside the `computeTopicsToCreate` for which we pass in `topicsNotReady`), it should fix it, allowing us to avoid the code complexity here?

Code after:
     * @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
                ));

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

        return newlyCreatedTopics;
    }