public synchronized Collection<Set<String>> copartitionGroups() {
        // compute transitive closures of copartitionGroups to relieve registering code to know all members
        // of a copartitionGroup at the same time
        final List<Set<String>> copartitionSourceTopics =
            copartitionSourceGroups
                .stream()
                .map(sourceGroup ->
                         sourceGroup
                             .stream()
                             .flatMap(node -> maybeDecorateInternalSourceTopics(nodeToSourceTopics.get(node)).stream())
                             .collect(Collectors.toSet())
                ).collect(Collectors.toList());

        final Map<String, Set<String>> topicsToCopartitionGroup = new LinkedHashMap<>();
        for (final Set<String> topics : copartitionSourceTopics) {
            if (topics != null) {
                Set<String> coparititonGroup = null;
                for (final String topic : topics) {
                    coparititonGroup = topicsToCopartitionGroup.get(topic);
                    if (coparititonGroup != null) {
                        break;
                    }
                }
                if (coparititonGroup == null) {
                    coparititonGroup = new HashSet<>();
                }
                coparititonGroup.addAll(maybeDecorateInternalSourceTopics(topics));
                for (final String topic : topics) {
                    topicsToCopartitionGroup.put(topic, coparititonGroup);
                }
            }
        }
        final Set<Set<String>> uniqueCopartitionGroups = new HashSet<>(topicsToCopartitionGroup.values());
        return Collections.unmodifiableList(new ArrayList<>(uniqueCopartitionGroups));
    }
@return  the full names (including application id/topology name prefix) of all source topics whose
         processors are connected to the given state store
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java