k-apol commented on code in PR #20326:
URL: https://github.com/apache/kafka/pull/20326#discussion_r2284888555
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,120 +466,119 @@ 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;
- Set<String> topicsNotReady = new HashSet<>(topics.keySet());
- final Set<String> newlyCreatedTopics = new HashSet<>();
+ final Set<String> topicsNotReady = new HashSet<>(topics.keySet());
+ final Set<String> newTopics = new HashSet<>();
Review Comment:
> Re-reading the code again, it seems the control flow is still to
convoluted?
>
> We have method `createTopics(...)` in which we pass `topicsToCreate` (this
makes sense), but we also pass in `topicsNotReady` and modify it as a side
effect -- this does not seem to be clean? It might be better to let
`createTopics(...)` return the set of topic it could create, and do the overall
bookkeeping outside of `createTopics(...)` (we could not need to pass
`topicsNotReady` into `createTopics(...)` at all for this case, but do:
>
> ```
> final Set<String> createdTopics = createTopics(topicsToCreate, deadlineMs);
> topicsNotReady.removeAll(createdTopics);
> ```
I agree, this is hard to reason about. I propose we pass `topicsNotReady`
into `createTopics(topicsToCreate, topicsNotReady, deadlineMs)`, and just
abstract the bookeeping step away inside the method. It feels okay to modify
the `topicsNotReady` inside here, since it logically tracks that as we create a
topic, we have one less that is 'not ready'.
--
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
