private void recycleTaskFromStateUpdater(final Task task,
                                             final Set<TopicPartition> inputPartitions,
                                             final Set<Task> tasksToCloseDirty,
                                             final Map<TaskId, RuntimeException> taskExceptions) {
        Task newTask = null;
        try {
            task.suspend();
            newTask = task.isActive() ?
                convertActiveToStandby((StreamTask) task, inputPartitions) :
                convertStandbyToActive((StandbyTask) task, inputPartitions);
            tasks.addPendingTasksToInit(Collections.singleton(newTask));
        } catch (final RuntimeException e) {
            final TaskId taskId = task.id();
            final String uncleanMessage = String.format("Failed to recycle task %s cleanly. " +
                "Attempting to close remaining tasks before re-throwing:", taskId);
            log.error(uncleanMessage, e);

            if (task.state() != State.CLOSED) {
                tasksToCloseDirty.add(task);
            }
            if (newTask != null && newTask.state() != State.CLOSED) {
                tasksToCloseDirty.add(newTask);
            }

            taskExceptions.putIfAbsent(taskId, e);
        }
    }
Tries to initialize any new or still-uninitialized tasks, then checks if they can/have completed restoration.

@throws IllegalStateException If store gets registered after initialized is already finished
@throws StreamsException if the store's change log does not contain the partition
@return {@code true} if all tasks are fully restored
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java