private void buildProcessorNode(final Map<String, ProcessorNode<?, ?, ?, ?>> processorMap,
                                    final Map<String, StateStore> stateStoreMap,
                                    final ProcessorNodeFactory<?, ?, ?, ?> factory,
                                    final ProcessorNode<Object, Object, Object, Object> node) {

        for (final String predecessor : factory.predecessors) {
            final ProcessorNode<Object, Object, Object, Object> predecessorNode = getProcessor(processorMap, predecessor);
            predecessorNode.addChild(node);
        }
        for (final String stateStoreName : factory.stateStoreNames) {
            if (!stateStoreMap.containsKey(stateStoreName)) {
                final StateStore store;
                if (stateFactories.containsKey(stateStoreName)) {
                    final StoreFactory storeFactory = stateFactories.get(stateStoreName);

                    // remember the changelog topic if this state store is change-logging enabled
                    if (storeFactory.loggingEnabled() && !storeToChangelogTopic.containsKey(stateStoreName)) {
                        final String prefix = topologyConfigs == null ?
                                applicationId :
                                ProcessorContextUtils.getPrefix(topologyConfigs.applicationConfigs.originals(), applicationId);
                        final String changelogTopic =
                            ProcessorStateManager.storeChangelogTopic(prefix, stateStoreName, topologyName);
                        storeToChangelogTopic.put(stateStoreName, changelogTopic);
                        changelogTopicToStore.put(changelogTopic, stateStoreName);
                    }
                    if (topologyConfigs != null) {
                        storeFactory.configure(topologyConfigs.applicationConfigs);
                    }
                    store = storeFactory.build();
                    stateStoreMap.put(stateStoreName, store);
                } else {
                    store = globalStateStores.get(stateStoreName);
                    stateStoreMap.put(stateStoreName, store);
                }

                if (store.persistent()) {
                    hasPersistentStores = true;
                }
            }
        }
    }
Builds the topology for any global state stores
@return ProcessorTopology of global state
streams/src/main/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilder.java