class TopologyBuilder {

    public synchronized Map<Integer, TopicsInfo> topicGroups(String applicationId) {
        Map<Integer, TopicsInfo> topicGroups = new HashMap<>();

        if (nodeGroups == null)
            nodeGroups = makeNodeGroups();

        for (Map.Entry<Integer, Set<String>> entry : nodeGroups.entrySet()) {
            Set<String> sinkTopics = new HashSet<>();
            Set<String> sourceTopics = new HashSet<>();
            Set<String> internalSourceTopics = new HashSet<>();
            Set<String> stateChangelogTopics = new HashSet<>();
            for (String node : entry.getValue()) {
                // if the node is a source node, add to the source topics
                String[] topics = nodeToSourceTopics.get(node);
                if (topics != null) {
                    // if some of the topics are internal, add them to the internal topics
                    for (String topic : topics) {
                        if (this.internalTopicNames.contains(topic)) {
                            // prefix the internal topic name with the application id
                            String internalTopic = applicationId + "-" + topic;
                            internalSourceTopics.add(internalTopic);
                            sourceTopics.add(internalTopic);
                        } else {
                            sourceTopics.add(topic);
                        }
                    }
                }

                // if the node is a sink node, add to the sink topics
                String topic = nodeToSinkTopic.get(node);
                if (topic != null) {
                    if (internalTopicNames.contains(topic)) {
                        // prefix the change log topic name with the application id
                        sinkTopics.add(applicationId + "-" + topic);
                    } else {
                        sinkTopics.add(topic);
                    }
                }

                // if the node is connected to a state, add to the state topics
                for (StateStoreFactory stateFactory : stateFactories.values()) {
                    if (stateFactory.isInternal && stateFactory.users.contains(node)) {
                        // prefix the change log topic name with the application id
                        stateChangelogTopics.add(applicationId + "-" + stateFactory.supplier.name() + ProcessorStateManager.STATE_CHANGELOG_TOPIC_SUFFIX);
                    }
                }
            }
            topicGroups.put(entry.getKey(), new TopicsInfo(
                    Collections.unmodifiableSet(sinkTopics),
                    Collections.unmodifiableSet(sourceTopics),
                    Collections.unmodifiableSet(internalSourceTopics),
                    Collections.unmodifiableSet(stateChangelogTopics)));
        }

        return Collections.unmodifiableMap(topicGroups);
    }

    public  TopologyBuilder();

}