cadonna commented on a change in pull request #11686:
URL: https://github.com/apache/kafka/pull/11686#discussion_r788604115
##########
File path:
streams/src/main/java/org/apache/kafka/streams/processor/internals/RepartitionTopics.java
##########
@@ -97,49 +101,74 @@ public boolean setup() {
}
}
}
+ }
- return missingUserInputTopicsPerTopology.isEmpty();
+ public Set<String> topologiesWithMissingInputTopics() {
+ return missingInputTopicsBySubtopology.keySet()
+ .stream()
+ .map(s -> getTopologyNameOrElseUnnamed(s.namedTopology))
+ .collect(Collectors.toSet());
}
- public Map<String, Set<String>> missingUserInputTopicsPerTopology() {
- return Collections.unmodifiableMap(missingUserInputTopicsPerTopology);
+ public Queue<StreamsException> missingSourceTopicExceptions() {
+ return missingInputTopicsBySubtopology.entrySet().stream().map(entry
-> {
+ final Set<String> missingSourceTopics = entry.getValue();
+ final int subtopologyId = entry.getKey().nodeGroupId;
+ final String topologyName = entry.getKey().namedTopology;
+
+ return new StreamsException(
+ new MissingSourceTopicException(String.format(
+ "Missing source topics %s for subtopology %s of topology
%s",
+ missingSourceTopics, subtopologyId, topologyName)),
+ new TaskId(subtopologyId, 0, topologyName));
Review comment:
My point was that I could not see where you use the task ID in this
specific case when you handle the exception and a `StreamsException` can also
be created without a task ID. So, I did not see the reason to introduce a dummy
task ID if it is not strictly needed by the exception. I am also fine with your
current solution.
--
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
