private void closeTaskClean(final Task task,
                                final Set<Task> tasksToCloseDirty,
                                final Map<TaskId, RuntimeException> taskExceptions) {
        try {
            task.suspend();
            task.closeClean();
            if (task.isActive()) {
                activeTaskCreator.closeAndRemoveTaskProducerIfNeeded(task.id());
            }
        } catch (final RuntimeException e) {
            final String uncleanMessage = String.format("Failed to close task %s cleanly. " +
                "Attempting to close remaining tasks before re-throwing:", task.id());
            log.error(uncleanMessage, e);

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

            taskExceptions.putIfAbsent(task.id(), 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