class 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;
    }

}