ableegoldman commented on a change in pull request #11686: URL: https://github.com/apache/kafka/pull/11686#discussion_r789243445
########## 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: We want to add a TaskId because we want to provide information about which topology is missing the source topics, and to a lesser extent which subtopology. If we create the StreamsException without a TaskId then it's considerably more difficult to figure out which topology (or which query) is experiencing the issue -- technically you could parse the error message in the MissingSourceTopicException but that's obviously very brittle and just not a good idea. Hence, we provide the `topologyName` through a "dummy" TaskId -- 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