class InternalTopologyBuilder {

    @SuppressWarnings("unchecked")
    private ProcessorTopology build(final Set<String> nodeGroup) {
        Objects.requireNonNull(applicationId, "topology has not completed optimization");

        final Map<String, ProcessorNode<?, ?, ?, ?>> processorMap = new LinkedHashMap<>();
        final Map<String, SourceNode<?, ?, ?, ?>> topicSourceMap = new HashMap<>();
        final Map<String, SinkNode<?, ?, ?, ?>> topicSinkMap = new HashMap<>();
        final Map<String, StateStore> stateStoreMap = new LinkedHashMap<>();
        final Set<String> repartitionTopics = new HashSet<>();

        // create processor nodes in a topological order ("nodeFactories" is already topologically sorted)
        // also make sure the state store map values following the insertion ordering
        for (final NodeFactory<?, ?, ?, ?> factory : nodeFactories.values()) {
            if (nodeGroup == null || nodeGroup.contains(factory.name)) {
                final ProcessorNode<?, ?, ?, ?> node = factory.build();
                processorMap.put(node.name(), node);

                if (factory instanceof ProcessorNodeFactory) {
                    buildProcessorNode(processorMap,
                                       stateStoreMap,
                                       (ProcessorNodeFactory<?, ?, ?, ?>) factory,
                                       (ProcessorNode<Object, Object, Object, Object>) node);

                } else if (factory instanceof SourceNodeFactory) {
                    buildSourceNode(topicSourceMap,
                                    repartitionTopics,
                                    (SourceNodeFactory<?, ?, ?, ?>) factory,
                                    (SourceNode<?, ?, ?, ?>) node);

                } else if (factory instanceof SinkNodeFactory) {
                    buildSinkNode(processorMap,
                                  topicSinkMap,
                                  repartitionTopics,
                                  (SinkNodeFactory<?, ?, ?, ?>) factory,
                                  (SinkNode<Object, Object, ?, ?>) node);
                } else {
                    throw new TopologyException("Unknown definition class: " + factory.getClass().getName());
                }
            }
        }

        return new ProcessorTopology(new ArrayList<>(processorMap.values()),
                                     topicSourceMap,
                                     topicSinkMap,
                                     new ArrayList<>(stateStoreMap.values()),
                                     new ArrayList<>(globalStateStores.values()),
                                     storeToChangelogTopic,
                                     repartitionTopics);
    }

}