class InternalTopologyBuilder {

    public final <KIn, VIn> void addGlobalStore(final StoreBuilder<?> storeBuilder,
                                                final String sourceName,
                                                final TimestampExtractor timestampExtractor,
                                                final Deserializer<KIn> keyDeserializer,
                                                final Deserializer<VIn> valueDeserializer,
                                                final String topic,
                                                final String processorName,
                                                final ProcessorSupplier<KIn, VIn, Void, Void> stateUpdateSupplier) {
        Objects.requireNonNull(storeBuilder, "store builder must not be null");
        validateGlobalStoreArguments(sourceName,
                                     topic,
                                     processorName,
                                     stateUpdateSupplier,
                                     storeBuilder.name(),
                                     storeBuilder.loggingEnabled());
        validateTopicNotAlreadyRegistered(topic);

        final String[] topics = {topic};
        final String[] predecessors = {sourceName};

        final ProcessorNodeFactory<KIn, VIn, Void, Void> nodeFactory = new ProcessorNodeFactory<>(
            processorName,
            predecessors,
            stateUpdateSupplier
        );

        globalTopics.add(topic);
        nodeFactories.put(sourceName, new SourceNodeFactory<>(
            sourceName,
            topics,
            null,
            timestampExtractor,
            keyDeserializer,
            valueDeserializer)
        );
        nodeToSourceTopics.put(sourceName, Arrays.asList(topics));
        nodeGrouper.add(sourceName);
        nodeFactory.addStateStore(storeBuilder.name());
        nodeFactories.put(processorName, nodeFactory);
        nodeGrouper.add(processorName);
        nodeGrouper.unite(processorName, predecessors);
        globalStateBuilders.put(storeBuilder.name(), storeBuilder);
        connectSourceStoreAndTopic(storeBuilder.name(), topic);
        nodeGroups = null;
    }

}