lihaosky commented on code in PR #13851:
URL: https://github.com/apache/kafka/pull/13851#discussion_r1255324708
##########
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:
I feel this log is ok since it logs internal `retryTimeoutMs` here which
makes it clear. Caller will have to find it out to log it.
--
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
