private static DefaultTaskManager maybeCreateSchedulingTaskManager(final boolean processingThreadsEnabled,
                                                                       final boolean stateUpdaterEnabled,
                                                                       final TopologyMetadata topologyMetadata,
                                                                       final Time time,
                                                                       final String threadId,
                                                                       final Tasks tasks) {
        if (processingThreadsEnabled) {
            if (!stateUpdaterEnabled) {
                throw new IllegalStateException("Processing threads require the state updater to be enabled");
            }

            final DefaultTaskManager defaultTaskManager = new DefaultTaskManager(
                time,
                threadId,
                tasks,
                new DefaultTaskExecutorCreator(),
                topologyMetadata.taskExecutionMetadata(),
                1
            );
            defaultTaskManager.startTaskExecutors();
            return defaultTaskManager;
        }
        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