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

Code before:
                        throw new StreamsException(
                                String.format("Could not create topic %s.", topicName),
                                cause
                        );
                    }
                }
            }

            if (!topicsNotReady.isEmpty()) {
                maybeThrowTimeout(new TimeoutContext(
                        topicsNotReady,
                        deadlineMs,
                        "createTopics timeout",
                        String.format(
                                "Could not create topics within %d milliseconds. This can happen if the Kafka cluster is temporarily not available.",
                                retryTimeoutMs),
                        null));
                log.info(
                    "Topics {} could not be made ready. Will retry in {} milliseconds. Remaining time in milliseconds: {}",
                    topicsNotReady,
                    retryBackOffMs,
                    deadlineMs - time.milliseconds()
                );
                Utils.sleep(retryBackOffMs);
            } else {
                return;
            }
        }
    } 
        

    /**
     * Try to get the partition information for the given topics; return the partition info for topics that already exists.
     *
     * Topics that were not able to get its description will simply not be returned
     */
    // visible for testing
    protected Map<String, List<TopicPartitionInfo>> getTopicPartitionInfo(final Set<String> topics,
                                                                          final Set<String> tempUnknownTopics) {
        final DescribeTopicsResult describeTopicsResult = adminClient.describeTopics(topics);
        final Map<String, KafkaFuture<TopicDescription>> futures = describeTopicsResult.topicNameValues();

        final Map<String, List<TopicPartitionInfo>> topicPartitionInfo = new HashMap<>();
        for (final Map.Entry<String, KafkaFuture<TopicDescription>> topicFuture : futures.entrySet()) {
            final String topicName = topicFuture.getKey();
            try {
                final TopicDescription topicDescription = topicFuture.getValue().get();
                topicPartitionInfo.put(topicName, topicDescription.partitions());
            } catch (final InterruptedException fatalException) {
                // this should not happen; if it ever happens it indicate a bug
                Thread.currentThread().interrupt();

Review comment (performance):
Should this go after the for-loop? I don't think we need to retry-backoff after verifying a single topic, but should be sufficient to check once for the whole "batch" of topics?

Code after:
                            "Error message was: {}", topicName, cause.toString());

                    if (cause instanceof UnsupportedVersionException) {
                        final String errorMessage = cause.getMessage();
                        if (errorMessage != null &&
                                errorMessage.startsWith("Creating topics with default partitions/replication factor are only supported in CreateTopicRequest version 4+")) {

                            throw new StreamsException(String.format(
                                    "Could not create topic %s, because brokers don't support configuration replication.factor=-1."
                                            + " You can change the replication.factor config or upgrade your brokers to version 2.4 or newer to avoid this error.",
                                    topicName)
                            );
                        }
                    } else if (cause instanceof TimeoutException) {
                        log.error("Creating topic {} timed out.\n" +
                                "Error message was: {}", topicName, cause.toString());
                    } else {
                        throw new StreamsException(
                                String.format("Could not create topic %s.", topicName),
                                cause
                        );
                    }
                }
            }

            if (!topicsNotReady.isEmpty()) {
                maybeThrowTimeout(new TimeoutContext(
                        topicsNotReady,
                        deadlineMs,
                        "createTopics timeout",
                        String.format(
                                "Could not create topics within %d milliseconds. This can happen if the Kafka cluster is temporarily not available.",
                                retryTimeoutMs),
                        null));
                log.info(
                    "Topics {} could not be made ready. Will retry in {} milliseconds. Remaining time in milliseconds: {}",
                    topicsNotReady,
                    retryBackOffMs,
                    deadlineMs - time.milliseconds()
                );
                Utils.sleep(retryBackOffMs);
            } else {
                continue;
            }
        } 
        return createdTopics;
    } 
        

    /**
     * Try to get the partition information for the given topics; return the partition info for topics that already exists.