@Override
    public void registerStore(final StateStore store,
                              final StateRestoreCallback stateRestoreCallback,
                              final CommitCallback ignored) {
        log.info("Restoring state for global store {}", store.name());

        // TODO (KAFKA-12887): we should not trigger user's exception handler for illegal-argument but always
        // fail-crash; in this case we would not need to immediately close the state store before throwing
        if (globalStores.containsKey(store.name())) {
            store.close();
            throw new IllegalArgumentException(String.format("Global Store %s has already been registered", store.name()));
        }

        if (!globalStoreNames.contains(store.name())) {
            store.close();
            throw new IllegalArgumentException(String.format("Trying to register store %s that is not a known global store", store.name()));
        }

        // register the store first, so that if later an exception is thrown then eventually while we call `close`
        // on the state manager this state store would be closed as well
        globalStores.put(store.name(), Optional.of(store));

        if (stateRestoreCallback == null) {
            throw new IllegalArgumentException(String.format("The stateRestoreCallback provided for store %s was null", store.name()));
        }

        final List<TopicPartition> topicPartitions = topicPartitionsForStore(store);
        final Map<TopicPartition, Long> highWatermarks = retryUntilSuccessOrThrowOnTaskTimeout(
            () -> globalConsumer.endOffsets(topicPartitions),
            String.format(
                "Failed to get offsets for partitions %s. The broker may be transiently unavailable at the moment.",
                topicPartitions
            )
        );

        try {
            final Optional<InternalTopologyBuilder.ReprocessFactory<?, ?, ?, ?>> reprocessFactory = topology
                .storeNameToReprocessOnRestore().getOrDefault(store.name(), Optional.empty());
            if (reprocessFactory.isPresent()) {
                reprocessState(
                    topicPartitions,
                    highWatermarks,
                    reprocessFactory.get(),
                    store.name());
            } else {
                restoreState(
                    stateRestoreCallback,
                    topicPartitions,
                    highWatermarks,
                    store.name(),
                    converterForStore(store)
                );
            }
        } finally {
            globalConsumer.unsubscribe();
        }
    }
This class is responsible for the initialization, restoration, closing, flushing etc
of Global State Stores. There is only ever 1 instance of this class per Application Instance.
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStateManagerImpl.java