public final <KIn, VIn> void addGlobalStore(final StoreFactory storeFactory,
                                                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,
                                                final boolean reprocessOnRestore) {
        Objects.requireNonNull(storeFactory, "store builder must not be null");
        ApiUtils.checkSupplier(stateUpdateSupplier);
        validateGlobalStoreArguments(sourceName,
                                     topic,
                                     processorName,
                                     stateUpdateSupplier,
                                     storeFactory.name(),
                                     storeFactory.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)
        );
        storeNameToReprocessOnRestore.put(storeFactory.name(),
            reprocessOnRestore ?
                Optional.of(new ReprocessFactory<>(stateUpdateSupplier, keyDeserializer, valueDeserializer))
                : Optional.empty());
        nodeToSourceTopics.put(sourceName, Arrays.asList(topics));
        nodeGrouper.add(sourceName);
        nodeFactory.addStateStore(storeFactory.name());
        nodeFactories.put(processorName, nodeFactory);
        nodeGrouper.add(processorName);
        nodeGrouper.unite(processorName, predecessors);
        globalStateBuilders.put(storeFactory.name(), storeFactory);
        connectSourceStoreAndTopic(storeFactory.name(), topic);
        nodeGroups = null;
    }
@Test
    public void testAddGlobalStoreWithBadSupplier() {
        final org.apache.kafka.streams.processor.api.Processor<?, ?, Void, Void> processor = new MockApiProcessorSupplier<Object, Object, Void, Void>().get();
        final IllegalArgumentException exception = assertThrows(
                IllegalArgumentException.class,
            () -> builder.addGlobalStore(
                        new MockKeyValueStoreBuilder("global-store", false).asFactory().withLoggingDisabled(),
                        "globalSource",
                        null,
                        null,
                        null,
                        "globalTopic",
                        "global-processor",
                () -> processor,
                false
            )
        );
        assertThat(exception.getMessage(), containsString("#get() must return a new object each time it is called."));
    }