mjsax commented on code in PR #19913:
URL: https://github.com/apache/kafka/pull/19913#discussion_r2162804621
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,15 +469,22 @@ public Set<String> makeReady(final Map<String,
InternalTopicConfig> 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);
- long currentWallClockMs = time.milliseconds();
+ final long currentWallClockMs = time.milliseconds();
final long deadlineMs = currentWallClockMs + retryTimeoutMs;
+ final long initDeadlineMs = currentWallClockMs +
this.initTimeout.toMillis();
Set<String> topicsNotReady = new HashSet<>(topics.keySet());
final Set<String> newlyCreatedTopics = new HashSet<>();
while (!topicsNotReady.isEmpty()) {
final Set<String> tempUnknownTopics = new HashSet<>();
topicsNotReady = validateTopics(topicsNotReady, topics,
tempUnknownTopics);
+
+ if (this.isManualInternalTopicConfig && !this.isInitializing) {
Review Comment:
> if (!topicsNotReady.isEmpty()) evaluates to true (the break condition for
the while loop), we are saying there are some topics to create.. but, if I
understand your point, that decision is not actually determined until validate
finishes executing?
Correct. When we enter this method for the very first time and reach the
`while`-loop, `topicsNotReady` contains _all_ internal topics that we may need
to create. Only after the first check which topics already exist, we get the
actual set of topics that we still need to create.
> Validate could return a result that says '**there's nothing to create', **
but the way this check stands, it would fail prematurely
Exactly.
--
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
