private static StateUpdater maybeCreateAndStartStateUpdater(final boolean stateUpdaterEnabled,
                                                                final StreamsMetricsImpl streamsMetrics,
                                                                final StreamsConfig streamsConfig,
                                                                final Consumer<byte[], byte[]> restoreConsumer,
                                                                final ChangelogReader changelogReader,
                                                                final TopologyMetadata topologyMetadata,
                                                                final Time time,
                                                                final String clientId,
                                                                final int threadIdx) {
        if (stateUpdaterEnabled) {
            final String name = clientId + "-StateUpdater-" + threadIdx;
            final StateUpdater stateUpdater = new DefaultStateUpdater(
                name,
                streamsMetrics.metricsRegistry(),
                streamsConfig,
                restoreConsumer,
                changelogReader,
                topologyMetadata,
                time
            );
            stateUpdater.start();
            return stateUpdater;
        } else {
            return null;
        }
    }
Sets the state

@param newState New state
@return The state prior to the call to setState, or null if the transition is invalid
streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java