private void initializeSubtopologyIdToStateStoreNamesMap() {
        final Map<Integer, Set<String>> storeNames = new HashMap<>();

        for (final Map.Entry<Integer, Set<String>> nodeGroup : makeNodeGroups().entrySet()) {
            final Set<String> subtopologyNodes = nodeGroup.getValue();
            final boolean isNodeGroupOfGlobalStores = nodeGroupContainsGlobalSourceNode(subtopologyNodes);

            if (!isNodeGroupOfGlobalStores) {
                final int subtopologyId = nodeGroup.getKey();
                final Set<String> subtopologyStoreNames = new HashSet<>();

                for (final String nodeName : subtopologyNodes) {
                    final AbstractNode node = nodeFactories.get(nodeName).describe();
                    if (node instanceof Processor) {
                        subtopologyStoreNames.addAll(((Processor) node).stores());
                    }
                }

                storeNames.put(subtopologyId, subtopologyStoreNames);
            }
        }
        subtopologyIdToStateStoreNames = storeNames;
    }
@return names of all source topics, including the application id/named topology prefix for repartition sources
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java