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

Code before:
                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 {}", newTopics);

        return newTopics;
    }

    private Set<NewTopic> computeTopicsToCreate(final Map<String, InternalTopicConfig> topics,
                                                      final Set<String> topicsNotReady,
                                                      final Set<String> newTopics) {
        final Set<String> tempUnknownTopics = new HashSet<>();
        final Set<String> validatedTopics = validateTopics(topicsNotReady, topics, tempUnknownTopics);

        newTopics.addAll(validatedTopics);

        final Set<NewTopic> topicsToCreate = new HashSet<>();

        for (final String topicName : topicsNotReady) {
            if (tempUnknownTopics.contains(topicName)) {
                // for the tempUnknownTopics, don't create topic for them
                // we'll check again later if remaining retries > 0
                continue;
            }
            final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
            final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);

            log.debug("Going to create topic {} with {} partitions and config {}.",
                    internalTopicConfig.name(),
                    internalTopicConfig.numberOfPartitions(),
                    topicConfig);

            topicsToCreate.add(
                    new NewTopic(
                            internalTopicConfig.name(),
                            internalTopicConfig.numberOfPartitions(),
                            Optional.of(replicationFactor))
                            .configs(topicConfig));
        }
        return topicsToCreate;
    }

    private void readyTopics(final Set<NewTopic> topicsToCreate,
                             final Set<String> topicsNotReady) {

Review comment (suggestion):
Cf my other comment further below: I think we need to add the retryBackOff sleep here, if we where not able to create all topics yet, but did not yet hit the timeout.

Code after:
                    .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;
    }

    private Set<NewTopic> computeTopicsToCreate(final Map<String, InternalTopicConfig> topics, final Set<String> tempUnknownTopics) {
        final Set<String> topicsNotYetCreated = identifyTopicsNotCreated(topics, tempUnknownTopics);

        final Set<NewTopic> topicsToCreate = new HashSet<>();
        
        for (final String topicName : topicsNotYetCreated) {
            // Topic already exists or non-deterministic result
            if (tempUnknownTopics.contains(topicName) || topics.get(topicName) == null) {
                // for the tempUnknownTopics, don't create topic for them
                // we'll check again later if remaining retries > 0
                continue;
            }
            final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
            final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);

            log.debug("Going to create topic {} with {} partitions and config {}.",
                    internalTopicConfig.name(),
                    internalTopicConfig.numberOfPartitions(),