@Override
    public void run() {
        final StateConsumer stateConsumer = initialize();

        if (stateConsumer == null) {
            // during initialization, the caller thread would wait for the state consumer
            // to restore the global state store before transiting to RUNNING state and return;
            // if an error happens during the restoration process, the stateConsumer will be null
            // and in this case we will transit the state to PENDING_SHUTDOWN and DEAD immediately.
            // the exception will be thrown in the caller thread during start() function.
            setState(State.PENDING_SHUTDOWN);
            setState(State.DEAD);

            log.error("Error happened during initialization of the global state store; this thread has shutdown.");
            streamsMetrics.removeAllThreadLevelSensors(getName());
            streamsMetrics.removeAllThreadLevelMetrics(getName());

            return;
        }
        setState(RUNNING);

        boolean wipeStateStore = false;
        try {
            while (stillRunning()) {
                final long size = cacheSize.getAndSet(-1L);
                if (size != -1L) {
                    cache.resize(size);
                }
                stateConsumer.pollAndUpdate();

                if (fetchDeadlineClientInstanceId != -1) {
                    if (fetchDeadlineClientInstanceId >= time.milliseconds()) {
                        try {
                            // we pass in a timeout of zero, to just trigger the "get instance id" background RPC,
                            // we don't want to block the global thread that can do useful work in the meantime
                            clientInstanceIdFuture.complete(globalConsumer.clientInstanceId(Duration.ZERO));
                            fetchDeadlineClientInstanceId = -1;
                        } catch (final IllegalStateException disabledError) {
                            // if telemetry is disabled on a client, we swallow the error,
                            // to allow returning a partial result for all other clients
                            clientInstanceIdFuture.complete(null);
                            fetchDeadlineClientInstanceId = -1;
                        } catch (final TimeoutException swallow) {
                            // swallow
                        } catch (final Exception error) {
                            clientInstanceIdFuture.completeExceptionally(error);
                            fetchDeadlineClientInstanceId = -1;
                        }
                    } else {
                        clientInstanceIdFuture.completeExceptionally(
                            new TimeoutException("Could not retrieve global consumer client instance id.")
                        );
                        fetchDeadlineClientInstanceId = -1;
                    }
                }
            }
        } catch (final InvalidOffsetException recoverableException) {
            wipeStateStore = true;
            log.error(
                "Updating 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
            );
            final StreamsException e = new StreamsException(
                "Updating global state failed. You can restart KafkaStreams to launch a new GlobalStreamThread to recover from this error.",
                recoverableException
            );
            this.streamsUncaughtExceptionHandler.accept(e);
        } catch (final Exception e) {
            log.error("Error happened while maintaining global state store. The streams application or client will now close to ERROR.", e);
            this.streamsUncaughtExceptionHandler.accept(e);
        } finally {
            // set the state to pending shutdown first as it may be called due to error;
            // its state may already be PENDING_SHUTDOWN so it will return false but we
            // intentionally do not check the returned flag
            setState(State.PENDING_SHUTDOWN);

            log.info("Shutting down");

            try {
                stateConsumer.close(wipeStateStore);
            } catch (final IOException e) {
                log.error("Failed to close state maintainer due to the following error:", e);
            }

            streamsMetrics.removeAllThreadLevelSensors(getName());
            streamsMetrics.removeAllThreadLevelMetrics(getName());

            setState(DEAD);

            log.info("Shutdown complete");
        }
    }
@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