mjsax commented on code in PR #13851:
URL: https://github.com/apache/kafka/pull/13851#discussion_r1253718618
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -409,6 +411,43 @@ private String getBrokerSideConfigValue(final Config
brokerSideTopicConfig,
return brokerSideConfigEntry.value();
}
+ public Map<String, List<TopicPartitionInfo>> getTopicPartitionInfo(final
Set<String> topics) {
+ log.debug("Starting to describe topics {} in partition assignor.",
topics);
+
+ long currentWallClockMs = time.milliseconds();
+ final long deadlineMs = currentWallClockMs + retryTimeoutMs;
+
+ Set<String> topicsToDescribe = new HashSet<>(topics);
+ final Map<String, List<TopicPartitionInfo>> topicPartitionInfo = new
HashMap<>();
+
+ while (!topicsToDescribe.isEmpty()) {
+ final Map<String, List<TopicPartitionInfo>> existed =
getTopicPartitionInfo(topicsToDescribe, null);
+ topicPartitionInfo.putAll(existed);
+ topicsToDescribe.removeAll(topicPartitionInfo.keySet());
+ if (!topicsToDescribe.isEmpty()) {
+ currentWallClockMs = time.milliseconds();
+
+ if (currentWallClockMs >= deadlineMs) {
+ final String timeoutError = String.format(
+ "Could not create topics within %d milliseconds. " +
+ "This can happen if the Kafka cluster is
temporarily not available.",
+ retryTimeoutMs);
+ log.error(timeoutError);
Review Comment:
Seem `makeReady` is not clean for this case -- should we instead try to
cleanup `makeReady`?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
