private StateConsumer initialize() {
        StateConsumer stateConsumer = null;
        try {
            final GlobalStateManager stateMgr = new GlobalStateManagerImpl(
                logContext,
                time,
                topology,
                globalConsumer,
                stateDirectory,
                stateRestoreListener,
                config
            );

            final GlobalProcessorContextImpl globalProcessorContext = new GlobalProcessorContextImpl(
                config,
                stateMgr,
                streamsMetrics,
                cache,
                time
            );
            stateMgr.setGlobalProcessorContext(globalProcessorContext);

            stateConsumer = new StateConsumer(
                logContext,
                globalConsumer,
                new GlobalStateUpdateTask(
                    logContext,
                    topology,
                    globalProcessorContext,
                    stateMgr,
                    config.defaultDeserializationExceptionHandler(),
                    time,
                    config.getLong(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG)
                ),
                Duration.ofMillis(config.getLong(StreamsConfig.POLL_MS_CONFIG))
            );

            try {
                stateConsumer.initialize();
            } catch (final InvalidOffsetException recoverableException) {
                log.error(
                    "Bootstrapping global state failed due to inconsistent local state. Will attempt to clean up the local state. You can restart KafkaStreams to recover from this error.",
                    recoverableException
                );

                closeStateConsumer(stateConsumer, true);

                throw new StreamsException(
                    "Bootstrapping global state failed. You can restart KafkaStreams to recover from this error.",
                    recoverableException
                );
            }

            return stateConsumer;
        } catch (final StreamsException fatalException) {
            closeStateConsumer(stateConsumer, false);
            startupException = fatalException;
        } catch (final Exception fatalException) {
            closeStateConsumer(stateConsumer, false);
            startupException = new StreamsException("Exception caught during initialization of GlobalStreamThread", fatalException);
        }
        return null;
    }
@throws IllegalStateException If store gets registered after initialized is already finished
@throws StreamsException      if the store's change log does not contain the partition
streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java