### Question:
TopologyUtils { public static Set<String> getSourceTopics(final TopologyDescription topology) { final Set<String> topics = new HashSet<>(); topology.globalStores().forEach(s -> { topics.addAll(s.source().topicSet()); }); topology.subtopologies().forEach(sub -> sub.nodes().forEach(n -> { if (n instanceof TopologyDescription.Source) { topics.addAll(((TopologyDescription.Source) n).topicSet()); } }) ); return topics; }  static Set<String> getSourceTopics(final TopologyDescription topology); static Set<String> getSinkTopics(final TopologyDescription topology); static Set<String> getUserDeclaredTopics(final TopologyDescription topology); static boolean isInternalTopic(final String applicationId,
                                          final String topic); static boolean isInternalTopic(final String topic);  }### Answer:
@Test public void shouldReturnOnlySourceTopics() { Set<String> userTopics = getSourceTopics(topology.describe()); assertEquals(2, userTopics.size()); assertTrue(Arrays.asList("input-topic", REPARTITION_TOPIC).containsAll(userTopics)); }