You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "k-apol (via GitHub)" <gi...@apache.org> on 2025/06/28 23:28:53 UTC
[PR] KAFKA-10357 Refactor makeReady to reduce complexity and support upcoming KIP-698 changes [kafka]
k-apol opened a new pull request, #20063:
URL: https://github.com/apache/kafka/pull/20063
This PR reduces the cyclomatic complexity of the makeReady method by extracting parts of its logic into smaller, well-named helper methods. This structural cleanup makes the code easier to understand and maintain.
These changes are intended to prepare for upcoming work related to [KIP-698](https://cwiki.apache.org/confluence/display/KAFKA/KIP-698), which will introduce additional branching logic to makeReady. Refactoring now ensures that future modifications can be made in a more modular and testable way.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10357 Refactor makeReady to reduce complexity and support upcoming KIP-698 changes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax closed pull request #20063: KAFKA-10357 Refactor makeReady to reduce complexity and support upcoming KIP-698 changes
URL: https://github.com/apache/kafka/pull/20063
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10357 Refactor makeReady to reduce complexity and support upcoming KIP-698 changes [kafka]
Posted by "k-apol (via GitHub)" <gi...@apache.org>.
k-apol commented on code in PR #20063:
URL: https://github.com/apache/kafka/pull/20063#discussion_r2247730573
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,120 +461,125 @@ 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> 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);
- newlyCreatedTopics.addAll(topicsNotReady);
-
+ final Set<NewTopic> validatedTopicObjects = createValidatedTopicObjects(topics, topicsNotReady, newlyCreatedTopics);
+ if (!validatedTopicObjects.isEmpty()) {
+ setupValidatedTopics(validatedTopicObjects, topicsNotReady);
Review Comment:
This is the same name as the method we call on the admin client.. I feel that could possibly cause some confusion. I feel `readyTopics` communicates the same intent but without the overlap. Let me know what you think
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10357 Refactor makeReady to reduce complexity and support upcoming KIP-698 changes [kafka]
Posted by "k-apol (via GitHub)" <gi...@apache.org>.
k-apol commented on code in PR #20063:
URL: https://github.com/apache/kafka/pull/20063#discussion_r2247841707
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,120 +461,125 @@ 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> 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);
- newlyCreatedTopics.addAll(topicsNotReady);
-
+ final Set<NewTopic> validatedTopicObjects = createValidatedTopicObjects(topics, topicsNotReady, newlyCreatedTopics);
+ if (!validatedTopicObjects.isEmpty()) {
+ setupValidatedTopics(validatedTopicObjects, topicsNotReady);
+ }
if (!topicsNotReady.isEmpty()) {
- final Set<NewTopic> newTopics = new HashSet<>();
+ maybeThrowTimeoutExceptionDuringMakeReady(deadlineMs);
+ }
+ }
+ log.debug("Completed validating internal topics and created {}", newlyCreatedTopics);
- for (final String topicName : topicsNotReady) {
- if (tempUnknownTopics.contains(topicName)) {
- // for the tempUnknownTopics, don't create topic for them
- // we'll check again later if remaining retries > 0
- continue;
- }
- final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
- final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
+ return newlyCreatedTopics;
+ }
+
+ private Set<NewTopic> createValidatedTopicObjects(final Map<String, InternalTopicConfig> topics,
+ Set<String> topicsNotReady,
+ final Set<String> newlyCreatedTopics) {
+ final Set<String> tempUnknownTopics = new HashSet<>();
+
+ topicsNotReady = validateTopics(topicsNotReady, topics, tempUnknownTopics);
+ newlyCreatedTopics.addAll(topicsNotReady);
+
+ final Set<NewTopic> validatedTopicObjects = new HashSet<>();
+
+ for (final String topicName : topicsNotReady) {
+ if (tempUnknownTopics.contains(topicName)) {
+ // for the tempUnknownTopics, don't create topic for them
+ // we'll check again later if remaining retries > 0
+ continue;
+ }
+ final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
+ final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
- log.debug("Going to create topic {} with {} partitions and config {}.",
- internalTopicConfig.name(),
- internalTopicConfig.numberOfPartitions(),
- topicConfig);
+ log.debug("Going to create topic {} with {} partitions and config {}.",
+ internalTopicConfig.name(),
+ internalTopicConfig.numberOfPartitions(),
+ topicConfig);
- newTopics.add(
- new NewTopic(
+ validatedTopicObjects.add(
+ new NewTopic(
internalTopicConfig.name(),
internalTopicConfig.numberOfPartitions(),
Optional.of(replicationFactor))
.configs(topicConfig));
- }
+ }
+ return validatedTopicObjects;
+ }
- // it's possible that although some topics are not ready yet because they
- // are temporarily not available, not that they do not exist; in this case
- // the new topics to create may be empty and hence we can skip here
- if (!newTopics.isEmpty()) {
- final CreateTopicsResult createTopicsResult = adminClient.createTopics(newTopics);
+ private void setupValidatedTopics(final Set<NewTopic> validatedTopicObjects,
+ final Set<String> topicsNotReady) {
+ final CreateTopicsResult createTopicsResult = adminClient.createTopics(validatedTopicObjects);
- for (final Map.Entry<String, KafkaFuture<Void>> createTopicResult : createTopicsResult.values().entrySet()) {
- final String topicName = createTopicResult.getKey();
- try {
- createTopicResult.getValue().get();
- topicsNotReady.remove(topicName);
- } catch (final InterruptedException fatalException) {
- // this should not happen; if it ever happens it indicate a bug
- Thread.currentThread().interrupt();
- log.error(INTERRUPTED_ERROR_MESSAGE, fatalException);
- throw new IllegalStateException(INTERRUPTED_ERROR_MESSAGE, fatalException);
- } catch (final ExecutionException executionException) {
- final Throwable cause = executionException.getCause();
- if (cause instanceof TopicExistsException) {
- // This topic didn't exist earlier or its leader not known before; just retain it for next round of validation.
- log.info(
- "Could not create topic {}. Topic is probably marked for deletion (number of partitions is unknown).\n"
- +
- "Will retry to create this topic in {} ms (to let broker finish async delete operation first).\n"
- +
- "Error message was: {}", topicName, retryBackOffMs,
- cause.toString());
- } else {
- log.error("Unexpected error during topic creation for {}.\n" +
- "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
- );
- }
- }
+ for (final Map.Entry<String, KafkaFuture<Void>> createTopicResult : createTopicsResult.values().entrySet()) {
+ final String topicName = createTopicResult.getKey();
+ try {
+ createTopicResult.getValue().get();
+ topicsNotReady.remove(topicName);
+ } catch (final InterruptedException fatalException) {
+ // this should not happen; if it ever happens it indicate a bug
+ Thread.currentThread().interrupt();
+ log.error(INTERRUPTED_ERROR_MESSAGE, fatalException);
+ throw new IllegalStateException(INTERRUPTED_ERROR_MESSAGE, fatalException);
+ } catch (final ExecutionException executionException) {
+ final Throwable cause = executionException.getCause();
+ if (cause instanceof TopicExistsException) {
+ // This topic didn't exist earlier or its leader not known before; just retain it for next round of validation.
+ log.info(
+ "Could not create topic {}. Topic is probably marked for deletion (number of partitions is unknown).\n"
+ +
+ "Will retry to create this topic in {} ms (to let broker finish async delete operation first).\n"
+ +
+ "Error message was: {}", topicName, retryBackOffMs,
+ cause.toString());
+ } else {
+ log.error("Unexpected error during topic creation for {}.\n" +
+ "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()) {
- currentWallClockMs = time.milliseconds();
+ private void maybeThrowTimeoutExceptionDuringMakeReady(final long deadlineMs) {
Review Comment:
I think we have a chance to refactor here, there are multiple implementations of a timeout that do mostly the same thing. I am working on a common implementation to use in this function that should solve for the problem you outlined
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10357 Refactor makeReady to reduce complexity and support upcoming KIP-698 changes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #20063:
URL: https://github.com/apache/kafka/pull/20063#issuecomment-3229922214
Closing this PR in favor of https://github.com/apache/kafka/pull/20326
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10357 Refactor makeReady to reduce complexity and support upcoming KIP-698 changes [kafka]
Posted by "k-apol (via GitHub)" <gi...@apache.org>.
k-apol commented on code in PR #20063:
URL: https://github.com/apache/kafka/pull/20063#discussion_r2247678846
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -113,7 +113,7 @@ public InternalTopicManager(final Time time,
}
}
- static class ValidationResult {
+ public static class ValidationResult {
Review Comment:
I did this because I will need to [access it from KakfaStreams.java when I implement the new `doInit()`](https://github.com/apache/kafka/pull/19913/files#diff-ab27af136b0c45ed402ec44368a91380b018bf06f1a9722324fe6be8d5220f7dR406) method.
Alternatively, I could make it `protected static class ...` instead to make it less available. But, my current understanding is that if I want to access this guy from KafkaStreams.java, I can't just leave it as it, right?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10357 Refactor makeReady to reduce complexity and support upcoming KIP-698 changes [kafka]
Posted by "k-apol (via GitHub)" <gi...@apache.org>.
k-apol commented on code in PR #20063:
URL: https://github.com/apache/kafka/pull/20063#discussion_r2247844632
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,120 +461,125 @@ 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> 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);
- newlyCreatedTopics.addAll(topicsNotReady);
-
+ final Set<NewTopic> validatedTopicObjects = createValidatedTopicObjects(topics, topicsNotReady, newlyCreatedTopics);
+ if (!validatedTopicObjects.isEmpty()) {
+ setupValidatedTopics(validatedTopicObjects, topicsNotReady);
+ }
if (!topicsNotReady.isEmpty()) {
- final Set<NewTopic> newTopics = new HashSet<>();
+ maybeThrowTimeoutExceptionDuringMakeReady(deadlineMs);
+ }
+ }
+ log.debug("Completed validating internal topics and created {}", newlyCreatedTopics);
- for (final String topicName : topicsNotReady) {
- if (tempUnknownTopics.contains(topicName)) {
- // for the tempUnknownTopics, don't create topic for them
- // we'll check again later if remaining retries > 0
- continue;
- }
- final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
- final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
+ return newlyCreatedTopics;
+ }
+
+ private Set<NewTopic> createValidatedTopicObjects(final Map<String, InternalTopicConfig> topics,
+ Set<String> topicsNotReady,
+ final Set<String> newlyCreatedTopics) {
+ final Set<String> tempUnknownTopics = new HashSet<>();
+
+ topicsNotReady = validateTopics(topicsNotReady, topics, tempUnknownTopics);
+ newlyCreatedTopics.addAll(topicsNotReady);
+
+ final Set<NewTopic> validatedTopicObjects = new HashSet<>();
+
+ for (final String topicName : topicsNotReady) {
+ if (tempUnknownTopics.contains(topicName)) {
+ // for the tempUnknownTopics, don't create topic for them
+ // we'll check again later if remaining retries > 0
+ continue;
+ }
+ final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
+ final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
- log.debug("Going to create topic {} with {} partitions and config {}.",
- internalTopicConfig.name(),
- internalTopicConfig.numberOfPartitions(),
- topicConfig);
+ log.debug("Going to create topic {} with {} partitions and config {}.",
+ internalTopicConfig.name(),
+ internalTopicConfig.numberOfPartitions(),
+ topicConfig);
- newTopics.add(
- new NewTopic(
+ validatedTopicObjects.add(
+ new NewTopic(
internalTopicConfig.name(),
internalTopicConfig.numberOfPartitions(),
Optional.of(replicationFactor))
.configs(topicConfig));
- }
+ }
+ return validatedTopicObjects;
+ }
- // it's possible that although some topics are not ready yet because they
- // are temporarily not available, not that they do not exist; in this case
- // the new topics to create may be empty and hence we can skip here
- if (!newTopics.isEmpty()) {
- final CreateTopicsResult createTopicsResult = adminClient.createTopics(newTopics);
+ private void setupValidatedTopics(final Set<NewTopic> validatedTopicObjects,
+ final Set<String> topicsNotReady) {
+ final CreateTopicsResult createTopicsResult = adminClient.createTopics(validatedTopicObjects);
- for (final Map.Entry<String, KafkaFuture<Void>> createTopicResult : createTopicsResult.values().entrySet()) {
- final String topicName = createTopicResult.getKey();
- try {
- createTopicResult.getValue().get();
- topicsNotReady.remove(topicName);
- } catch (final InterruptedException fatalException) {
- // this should not happen; if it ever happens it indicate a bug
- Thread.currentThread().interrupt();
- log.error(INTERRUPTED_ERROR_MESSAGE, fatalException);
- throw new IllegalStateException(INTERRUPTED_ERROR_MESSAGE, fatalException);
- } catch (final ExecutionException executionException) {
- final Throwable cause = executionException.getCause();
- if (cause instanceof TopicExistsException) {
- // This topic didn't exist earlier or its leader not known before; just retain it for next round of validation.
- log.info(
- "Could not create topic {}. Topic is probably marked for deletion (number of partitions is unknown).\n"
- +
- "Will retry to create this topic in {} ms (to let broker finish async delete operation first).\n"
- +
- "Error message was: {}", topicName, retryBackOffMs,
- cause.toString());
- } else {
- log.error("Unexpected error during topic creation for {}.\n" +
- "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
- );
- }
- }
+ for (final Map.Entry<String, KafkaFuture<Void>> createTopicResult : createTopicsResult.values().entrySet()) {
+ final String topicName = createTopicResult.getKey();
+ try {
+ createTopicResult.getValue().get();
+ topicsNotReady.remove(topicName);
+ } catch (final InterruptedException fatalException) {
+ // this should not happen; if it ever happens it indicate a bug
+ Thread.currentThread().interrupt();
+ log.error(INTERRUPTED_ERROR_MESSAGE, fatalException);
+ throw new IllegalStateException(INTERRUPTED_ERROR_MESSAGE, fatalException);
+ } catch (final ExecutionException executionException) {
+ final Throwable cause = executionException.getCause();
+ if (cause instanceof TopicExistsException) {
+ // This topic didn't exist earlier or its leader not known before; just retain it for next round of validation.
+ log.info(
+ "Could not create topic {}. Topic is probably marked for deletion (number of partitions is unknown).\n"
+ +
+ "Will retry to create this topic in {} ms (to let broker finish async delete operation first).\n"
+ +
+ "Error message was: {}", topicName, retryBackOffMs,
+ cause.toString());
+ } else {
+ log.error("Unexpected error during topic creation for {}.\n" +
+ "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()) {
- currentWallClockMs = time.milliseconds();
+ private void maybeThrowTimeoutExceptionDuringMakeReady(final long deadlineMs) {
+ final long 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);
- 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);
- throw new TimeoutException(timeoutError);
- }
- log.info(
Review Comment:
I will add it back where its needed, see above comment on refactoring timeouts in this class
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10357 Refactor makeReady to reduce complexity and support upcoming KIP-698 changes [kafka]
Posted by "k-apol (via GitHub)" <gi...@apache.org>.
k-apol commented on code in PR #20063:
URL: https://github.com/apache/kafka/pull/20063#discussion_r2247716436
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,120 +461,125 @@ 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> 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);
- newlyCreatedTopics.addAll(topicsNotReady);
-
+ final Set<NewTopic> validatedTopicObjects = createValidatedTopicObjects(topics, topicsNotReady, newlyCreatedTopics);
Review Comment:
I thought it would be easier to reason about the current state of this method by writing it this way. The motivation for this was that I got a bit confused reading through the old code because they are not really 'topics' officially until we call the admin client. It was easier for me to reason about by being more explicit about their current state at this point in the method, I thought other maintainers would feel the same.
But, I see your point, maybe it's excessive and actually adding more confusion.. I will defer to your judgment here and update this
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10357 Refactor makeReady to reduce complexity and support upcoming KIP-698 changes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #20063:
URL: https://github.com/apache/kafka/pull/20063#issuecomment-3190752585
Seems replace this PR with https://github.com/apache/kafka/pull/20326 -- should this one be closed?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10357 Refactor makeReady to reduce complexity and support upcoming KIP-698 changes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #20063:
URL: https://github.com/apache/kafka/pull/20063#discussion_r2241346902
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,120 +461,125 @@ 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> 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);
- newlyCreatedTopics.addAll(topicsNotReady);
-
+ final Set<NewTopic> validatedTopicObjects = createValidatedTopicObjects(topics, topicsNotReady, newlyCreatedTopics);
+ if (!validatedTopicObjects.isEmpty()) {
+ setupValidatedTopics(validatedTopicObjects, topicsNotReady);
+ }
if (!topicsNotReady.isEmpty()) {
- final Set<NewTopic> newTopics = new HashSet<>();
+ maybeThrowTimeoutExceptionDuringMakeReady(deadlineMs);
+ }
+ }
+ log.debug("Completed validating internal topics and created {}", newlyCreatedTopics);
- for (final String topicName : topicsNotReady) {
- if (tempUnknownTopics.contains(topicName)) {
- // for the tempUnknownTopics, don't create topic for them
- // we'll check again later if remaining retries > 0
- continue;
- }
- final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
- final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
+ return newlyCreatedTopics;
+ }
+
+ private Set<NewTopic> createValidatedTopicObjects(final Map<String, InternalTopicConfig> topics,
+ Set<String> topicsNotReady,
+ final Set<String> newlyCreatedTopics) {
+ final Set<String> tempUnknownTopics = new HashSet<>();
+
+ topicsNotReady = validateTopics(topicsNotReady, topics, tempUnknownTopics);
Review Comment:
It does not sound right to overwrite input parameter `topicsNotReady` -- in the original code, this was ok as it was the same variable, but now that we copy a reference when calling `createValidatedTopicObjects` this change does not reflect back to the caller.
If we really need to modify the content of `topicsNotReady`, we need to apply corresponding set operations to modify the collection.
But I am not sure if the overall code structure is ideal anyway -- it might be better to also change `validateTopics` and only return topic we want to create? For this case, we would introduce a new variable here `final topicToBeCreated = ...`.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,120 +461,125 @@ 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> 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);
- newlyCreatedTopics.addAll(topicsNotReady);
-
+ final Set<NewTopic> validatedTopicObjects = createValidatedTopicObjects(topics, topicsNotReady, newlyCreatedTopics);
+ if (!validatedTopicObjects.isEmpty()) {
+ setupValidatedTopics(validatedTopicObjects, topicsNotReady);
+ }
if (!topicsNotReady.isEmpty()) {
- final Set<NewTopic> newTopics = new HashSet<>();
+ maybeThrowTimeoutExceptionDuringMakeReady(deadlineMs);
+ }
+ }
+ log.debug("Completed validating internal topics and created {}", newlyCreatedTopics);
- for (final String topicName : topicsNotReady) {
- if (tempUnknownTopics.contains(topicName)) {
- // for the tempUnknownTopics, don't create topic for them
- // we'll check again later if remaining retries > 0
- continue;
- }
- final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
- final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
+ return newlyCreatedTopics;
+ }
+
+ private Set<NewTopic> createValidatedTopicObjects(final Map<String, InternalTopicConfig> topics,
+ Set<String> topicsNotReady,
+ final Set<String> newlyCreatedTopics) {
+ final Set<String> tempUnknownTopics = new HashSet<>();
+
+ topicsNotReady = validateTopics(topicsNotReady, topics, tempUnknownTopics);
+ newlyCreatedTopics.addAll(topicsNotReady);
+
+ final Set<NewTopic> validatedTopicObjects = new HashSet<>();
+
+ for (final String topicName : topicsNotReady) {
+ if (tempUnknownTopics.contains(topicName)) {
Review Comment:
If we change `validateTopics` (cf other comment above) to only return topic we want to create, we won't need this check.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,120 +461,125 @@ 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> 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);
- newlyCreatedTopics.addAll(topicsNotReady);
-
+ final Set<NewTopic> validatedTopicObjects = createValidatedTopicObjects(topics, topicsNotReady, newlyCreatedTopics);
+ if (!validatedTopicObjects.isEmpty()) {
+ setupValidatedTopics(validatedTopicObjects, topicsNotReady);
Review Comment:
Rename `setupValidatedTopics` -> `createTopics` ?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,120 +461,125 @@ 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> 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);
- newlyCreatedTopics.addAll(topicsNotReady);
-
+ final Set<NewTopic> validatedTopicObjects = createValidatedTopicObjects(topics, topicsNotReady, newlyCreatedTopics);
Review Comment:
`validatedTopicObjects` are the topics we want to create, right? Not sure what "validated" mean? Why not rename to `topicsToBeCreated` ?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -113,7 +113,7 @@ public InternalTopicManager(final Time time,
}
}
- static class ValidationResult {
+ public static class ValidationResult {
Review Comment:
Why does this class need to become public?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,120 +461,125 @@ 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> 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);
- newlyCreatedTopics.addAll(topicsNotReady);
-
+ final Set<NewTopic> validatedTopicObjects = createValidatedTopicObjects(topics, topicsNotReady, newlyCreatedTopics);
+ if (!validatedTopicObjects.isEmpty()) {
+ setupValidatedTopics(validatedTopicObjects, topicsNotReady);
+ }
if (!topicsNotReady.isEmpty()) {
- final Set<NewTopic> newTopics = new HashSet<>();
+ maybeThrowTimeoutExceptionDuringMakeReady(deadlineMs);
+ }
+ }
+ log.debug("Completed validating internal topics and created {}", newlyCreatedTopics);
- for (final String topicName : topicsNotReady) {
- if (tempUnknownTopics.contains(topicName)) {
- // for the tempUnknownTopics, don't create topic for them
- // we'll check again later if remaining retries > 0
- continue;
- }
- final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
- final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
+ return newlyCreatedTopics;
+ }
+
+ private Set<NewTopic> createValidatedTopicObjects(final Map<String, InternalTopicConfig> topics,
+ Set<String> topicsNotReady,
+ final Set<String> newlyCreatedTopics) {
+ final Set<String> tempUnknownTopics = new HashSet<>();
+
+ topicsNotReady = validateTopics(topicsNotReady, topics, tempUnknownTopics);
+ newlyCreatedTopics.addAll(topicsNotReady);
+
+ final Set<NewTopic> validatedTopicObjects = new HashSet<>();
+
+ for (final String topicName : topicsNotReady) {
+ if (tempUnknownTopics.contains(topicName)) {
+ // for the tempUnknownTopics, don't create topic for them
+ // we'll check again later if remaining retries > 0
+ continue;
+ }
+ final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
+ final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
- log.debug("Going to create topic {} with {} partitions and config {}.",
- internalTopicConfig.name(),
- internalTopicConfig.numberOfPartitions(),
- topicConfig);
+ log.debug("Going to create topic {} with {} partitions and config {}.",
+ internalTopicConfig.name(),
+ internalTopicConfig.numberOfPartitions(),
+ topicConfig);
- newTopics.add(
- new NewTopic(
+ validatedTopicObjects.add(
+ new NewTopic(
internalTopicConfig.name(),
internalTopicConfig.numberOfPartitions(),
Optional.of(replicationFactor))
.configs(topicConfig));
- }
+ }
+ return validatedTopicObjects;
+ }
- // it's possible that although some topics are not ready yet because they
- // are temporarily not available, not that they do not exist; in this case
- // the new topics to create may be empty and hence we can skip here
- if (!newTopics.isEmpty()) {
- final CreateTopicsResult createTopicsResult = adminClient.createTopics(newTopics);
+ private void setupValidatedTopics(final Set<NewTopic> validatedTopicObjects,
Review Comment:
Not sure why we use "validated" in so many name? This method does create new topic, so why no just call it `createTopics` ?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,120 +461,125 @@ 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> 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);
- newlyCreatedTopics.addAll(topicsNotReady);
-
+ final Set<NewTopic> validatedTopicObjects = createValidatedTopicObjects(topics, topicsNotReady, newlyCreatedTopics);
+ if (!validatedTopicObjects.isEmpty()) {
+ setupValidatedTopics(validatedTopicObjects, topicsNotReady);
+ }
if (!topicsNotReady.isEmpty()) {
- final Set<NewTopic> newTopics = new HashSet<>();
+ maybeThrowTimeoutExceptionDuringMakeReady(deadlineMs);
+ }
+ }
+ log.debug("Completed validating internal topics and created {}", newlyCreatedTopics);
- for (final String topicName : topicsNotReady) {
- if (tempUnknownTopics.contains(topicName)) {
- // for the tempUnknownTopics, don't create topic for them
- // we'll check again later if remaining retries > 0
- continue;
- }
- final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
- final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
+ return newlyCreatedTopics;
+ }
+
+ private Set<NewTopic> createValidatedTopicObjects(final Map<String, InternalTopicConfig> topics,
+ Set<String> topicsNotReady,
+ final Set<String> newlyCreatedTopics) {
+ final Set<String> tempUnknownTopics = new HashSet<>();
+
+ topicsNotReady = validateTopics(topicsNotReady, topics, tempUnknownTopics);
+ newlyCreatedTopics.addAll(topicsNotReady);
+
+ final Set<NewTopic> validatedTopicObjects = new HashSet<>();
Review Comment:
These are the topic we want to create, right? So maybe better name it `topicsToBeCreated` ?
And also rename the method name accordingly, maybe `computeTopicToBeCreated`?
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,120 +461,125 @@ 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> 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);
- newlyCreatedTopics.addAll(topicsNotReady);
-
+ final Set<NewTopic> validatedTopicObjects = createValidatedTopicObjects(topics, topicsNotReady, newlyCreatedTopics);
+ if (!validatedTopicObjects.isEmpty()) {
+ setupValidatedTopics(validatedTopicObjects, topicsNotReady);
+ }
if (!topicsNotReady.isEmpty()) {
- final Set<NewTopic> newTopics = new HashSet<>();
+ maybeThrowTimeoutExceptionDuringMakeReady(deadlineMs);
+ }
+ }
+ log.debug("Completed validating internal topics and created {}", newlyCreatedTopics);
- for (final String topicName : topicsNotReady) {
- if (tempUnknownTopics.contains(topicName)) {
- // for the tempUnknownTopics, don't create topic for them
- // we'll check again later if remaining retries > 0
- continue;
- }
- final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
- final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
+ return newlyCreatedTopics;
+ }
+
+ private Set<NewTopic> createValidatedTopicObjects(final Map<String, InternalTopicConfig> topics,
+ Set<String> topicsNotReady,
+ final Set<String> newlyCreatedTopics) {
Review Comment:
nit: fix indention -- also all parameters should be final -- cf my comment below about `topicsNotReady`
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,120 +461,125 @@ 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> 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);
- newlyCreatedTopics.addAll(topicsNotReady);
-
+ final Set<NewTopic> validatedTopicObjects = createValidatedTopicObjects(topics, topicsNotReady, newlyCreatedTopics);
+ if (!validatedTopicObjects.isEmpty()) {
+ setupValidatedTopics(validatedTopicObjects, topicsNotReady);
+ }
if (!topicsNotReady.isEmpty()) {
- final Set<NewTopic> newTopics = new HashSet<>();
+ maybeThrowTimeoutExceptionDuringMakeReady(deadlineMs);
+ }
+ }
+ log.debug("Completed validating internal topics and created {}", newlyCreatedTopics);
- for (final String topicName : topicsNotReady) {
- if (tempUnknownTopics.contains(topicName)) {
- // for the tempUnknownTopics, don't create topic for them
- // we'll check again later if remaining retries > 0
- continue;
- }
- final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
- final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
+ return newlyCreatedTopics;
+ }
+
+ private Set<NewTopic> createValidatedTopicObjects(final Map<String, InternalTopicConfig> topics,
+ Set<String> topicsNotReady,
+ final Set<String> newlyCreatedTopics) {
+ final Set<String> tempUnknownTopics = new HashSet<>();
+
+ topicsNotReady = validateTopics(topicsNotReady, topics, tempUnknownTopics);
+ newlyCreatedTopics.addAll(topicsNotReady);
+
+ final Set<NewTopic> validatedTopicObjects = new HashSet<>();
+
+ for (final String topicName : topicsNotReady) {
+ if (tempUnknownTopics.contains(topicName)) {
+ // for the tempUnknownTopics, don't create topic for them
+ // we'll check again later if remaining retries > 0
+ continue;
+ }
+ final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
+ final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
- log.debug("Going to create topic {} with {} partitions and config {}.",
- internalTopicConfig.name(),
- internalTopicConfig.numberOfPartitions(),
- topicConfig);
+ log.debug("Going to create topic {} with {} partitions and config {}.",
+ internalTopicConfig.name(),
+ internalTopicConfig.numberOfPartitions(),
+ topicConfig);
- newTopics.add(
- new NewTopic(
+ validatedTopicObjects.add(
+ new NewTopic(
internalTopicConfig.name(),
internalTopicConfig.numberOfPartitions(),
Optional.of(replicationFactor))
.configs(topicConfig));
- }
+ }
+ return validatedTopicObjects;
+ }
- // it's possible that although some topics are not ready yet because they
- // are temporarily not available, not that they do not exist; in this case
- // the new topics to create may be empty and hence we can skip here
- if (!newTopics.isEmpty()) {
- final CreateTopicsResult createTopicsResult = adminClient.createTopics(newTopics);
+ private void setupValidatedTopics(final Set<NewTopic> validatedTopicObjects,
+ final Set<String> topicsNotReady) {
+ final CreateTopicsResult createTopicsResult = adminClient.createTopics(validatedTopicObjects);
- for (final Map.Entry<String, KafkaFuture<Void>> createTopicResult : createTopicsResult.values().entrySet()) {
- final String topicName = createTopicResult.getKey();
- try {
- createTopicResult.getValue().get();
- topicsNotReady.remove(topicName);
- } catch (final InterruptedException fatalException) {
- // this should not happen; if it ever happens it indicate a bug
- Thread.currentThread().interrupt();
- log.error(INTERRUPTED_ERROR_MESSAGE, fatalException);
- throw new IllegalStateException(INTERRUPTED_ERROR_MESSAGE, fatalException);
- } catch (final ExecutionException executionException) {
- final Throwable cause = executionException.getCause();
- if (cause instanceof TopicExistsException) {
- // This topic didn't exist earlier or its leader not known before; just retain it for next round of validation.
- log.info(
- "Could not create topic {}. Topic is probably marked for deletion (number of partitions is unknown).\n"
- +
- "Will retry to create this topic in {} ms (to let broker finish async delete operation first).\n"
- +
- "Error message was: {}", topicName, retryBackOffMs,
- cause.toString());
- } else {
- log.error("Unexpected error during topic creation for {}.\n" +
- "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
- );
- }
- }
+ for (final Map.Entry<String, KafkaFuture<Void>> createTopicResult : createTopicsResult.values().entrySet()) {
+ final String topicName = createTopicResult.getKey();
+ try {
+ createTopicResult.getValue().get();
+ topicsNotReady.remove(topicName);
+ } catch (final InterruptedException fatalException) {
+ // this should not happen; if it ever happens it indicate a bug
+ Thread.currentThread().interrupt();
+ log.error(INTERRUPTED_ERROR_MESSAGE, fatalException);
+ throw new IllegalStateException(INTERRUPTED_ERROR_MESSAGE, fatalException);
+ } catch (final ExecutionException executionException) {
+ final Throwable cause = executionException.getCause();
+ if (cause instanceof TopicExistsException) {
+ // This topic didn't exist earlier or its leader not known before; just retain it for next round of validation.
+ log.info(
+ "Could not create topic {}. Topic is probably marked for deletion (number of partitions is unknown).\n"
+ +
+ "Will retry to create this topic in {} ms (to let broker finish async delete operation first).\n"
+ +
+ "Error message was: {}", topicName, retryBackOffMs,
+ cause.toString());
+ } else {
+ log.error("Unexpected error during topic creation for {}.\n" +
+ "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()) {
- currentWallClockMs = time.milliseconds();
+ private void maybeThrowTimeoutExceptionDuringMakeReady(final long deadlineMs) {
+ final long 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);
- 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);
- throw new TimeoutException(timeoutError);
- }
- log.info(
Review Comment:
Why did we remove this log line? I mean, it does not belong into this method any longer, but we should not drop it.
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,120 +461,125 @@ 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> 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);
- newlyCreatedTopics.addAll(topicsNotReady);
-
+ final Set<NewTopic> validatedTopicObjects = createValidatedTopicObjects(topics, topicsNotReady, newlyCreatedTopics);
+ if (!validatedTopicObjects.isEmpty()) {
+ setupValidatedTopics(validatedTopicObjects, topicsNotReady);
+ }
if (!topicsNotReady.isEmpty()) {
- final Set<NewTopic> newTopics = new HashSet<>();
+ maybeThrowTimeoutExceptionDuringMakeReady(deadlineMs);
+ }
+ }
+ log.debug("Completed validating internal topics and created {}", newlyCreatedTopics);
- for (final String topicName : topicsNotReady) {
- if (tempUnknownTopics.contains(topicName)) {
- // for the tempUnknownTopics, don't create topic for them
- // we'll check again later if remaining retries > 0
- continue;
- }
- final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
- final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
+ return newlyCreatedTopics;
+ }
+
+ private Set<NewTopic> createValidatedTopicObjects(final Map<String, InternalTopicConfig> topics,
+ Set<String> topicsNotReady,
+ final Set<String> newlyCreatedTopics) {
+ final Set<String> tempUnknownTopics = new HashSet<>();
+
+ topicsNotReady = validateTopics(topicsNotReady, topics, tempUnknownTopics);
+ newlyCreatedTopics.addAll(topicsNotReady);
+
+ final Set<NewTopic> validatedTopicObjects = new HashSet<>();
+
+ for (final String topicName : topicsNotReady) {
+ if (tempUnknownTopics.contains(topicName)) {
+ // for the tempUnknownTopics, don't create topic for them
+ // we'll check again later if remaining retries > 0
+ continue;
+ }
+ final InternalTopicConfig internalTopicConfig = Objects.requireNonNull(topics.get(topicName));
+ final Map<String, String> topicConfig = internalTopicConfig.properties(defaultTopicConfigs, windowChangeLogAdditionalRetention);
- log.debug("Going to create topic {} with {} partitions and config {}.",
- internalTopicConfig.name(),
- internalTopicConfig.numberOfPartitions(),
- topicConfig);
+ log.debug("Going to create topic {} with {} partitions and config {}.",
+ internalTopicConfig.name(),
+ internalTopicConfig.numberOfPartitions(),
+ topicConfig);
- newTopics.add(
- new NewTopic(
+ validatedTopicObjects.add(
+ new NewTopic(
internalTopicConfig.name(),
internalTopicConfig.numberOfPartitions(),
Optional.of(replicationFactor))
.configs(topicConfig));
- }
+ }
+ return validatedTopicObjects;
+ }
- // it's possible that although some topics are not ready yet because they
- // are temporarily not available, not that they do not exist; in this case
- // the new topics to create may be empty and hence we can skip here
- if (!newTopics.isEmpty()) {
- final CreateTopicsResult createTopicsResult = adminClient.createTopics(newTopics);
+ private void setupValidatedTopics(final Set<NewTopic> validatedTopicObjects,
+ final Set<String> topicsNotReady) {
+ final CreateTopicsResult createTopicsResult = adminClient.createTopics(validatedTopicObjects);
- for (final Map.Entry<String, KafkaFuture<Void>> createTopicResult : createTopicsResult.values().entrySet()) {
- final String topicName = createTopicResult.getKey();
- try {
- createTopicResult.getValue().get();
- topicsNotReady.remove(topicName);
- } catch (final InterruptedException fatalException) {
- // this should not happen; if it ever happens it indicate a bug
- Thread.currentThread().interrupt();
- log.error(INTERRUPTED_ERROR_MESSAGE, fatalException);
- throw new IllegalStateException(INTERRUPTED_ERROR_MESSAGE, fatalException);
- } catch (final ExecutionException executionException) {
- final Throwable cause = executionException.getCause();
- if (cause instanceof TopicExistsException) {
- // This topic didn't exist earlier or its leader not known before; just retain it for next round of validation.
- log.info(
- "Could not create topic {}. Topic is probably marked for deletion (number of partitions is unknown).\n"
- +
- "Will retry to create this topic in {} ms (to let broker finish async delete operation first).\n"
- +
- "Error message was: {}", topicName, retryBackOffMs,
- cause.toString());
- } else {
- log.error("Unexpected error during topic creation for {}.\n" +
- "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
- );
- }
- }
+ for (final Map.Entry<String, KafkaFuture<Void>> createTopicResult : createTopicsResult.values().entrySet()) {
+ final String topicName = createTopicResult.getKey();
+ try {
+ createTopicResult.getValue().get();
+ topicsNotReady.remove(topicName);
+ } catch (final InterruptedException fatalException) {
+ // this should not happen; if it ever happens it indicate a bug
+ Thread.currentThread().interrupt();
+ log.error(INTERRUPTED_ERROR_MESSAGE, fatalException);
+ throw new IllegalStateException(INTERRUPTED_ERROR_MESSAGE, fatalException);
+ } catch (final ExecutionException executionException) {
+ final Throwable cause = executionException.getCause();
+ if (cause instanceof TopicExistsException) {
+ // This topic didn't exist earlier or its leader not known before; just retain it for next round of validation.
+ log.info(
+ "Could not create topic {}. Topic is probably marked for deletion (number of partitions is unknown).\n"
+ +
+ "Will retry to create this topic in {} ms (to let broker finish async delete operation first).\n"
+ +
+ "Error message was: {}", topicName, retryBackOffMs,
+ cause.toString());
+ } else {
+ log.error("Unexpected error during topic creation for {}.\n" +
+ "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()) {
- currentWallClockMs = time.milliseconds();
+ private void maybeThrowTimeoutExceptionDuringMakeReady(final long deadlineMs) {
Review Comment:
`maybeThrowTimeoutExceptionDuringMakeReady` -> the suffix `duringMakeReady` is a little weird I would drop it
Also: this method might actually block/sleep to implement a retry-backoff -- might be good if the name of the method reflects this?
Maybe `retryBackoffOrThrowTimeoutException` or similar?
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-10357 Refactor makeReady to reduce complexity and support upcoming KIP-698 changes [kafka]
Posted by "k-apol (via GitHub)" <gi...@apache.org>.
k-apol commented on code in PR #20063:
URL: https://github.com/apache/kafka/pull/20063#discussion_r2247716436
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopicManager.java:
##########
@@ -461,120 +461,125 @@ 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> 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);
- newlyCreatedTopics.addAll(topicsNotReady);
-
+ final Set<NewTopic> validatedTopicObjects = createValidatedTopicObjects(topics, topicsNotReady, newlyCreatedTopics);
Review Comment:
I thought it would be easier to reason about the current state of this method by writing it this way. The motivation for this was that I got a bit confused reading through the old code because they are not really 'topics' officially until we call the admin client. It was easier for me to reason about by being more explicit about their current state at this point in the method, I thought other maintainers would feel the same.
But, I see your point and also feel this is better - we are still communicating that they are not yet 'real' but with less and simpler wording. Making an update
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org