public Map<TaskId, RuntimeException> collectExceptionsAndFailedTasksFromStateUpdater() {
        final Map<TaskId, RuntimeException> taskExceptions = new LinkedHashMap<>();

        for (final StateUpdater.ExceptionAndTask exceptionAndTask : stateUpdater.drainExceptionsAndFailedTasks()) {
            final RuntimeException exception = exceptionAndTask.exception();
            final Task failedTask = exceptionAndTask.task();
            // need to add task back to the bookkeeping to be handled by the stream thread
            tasks.addTask(failedTask);
            taskExceptions.put(failedTask.id(), exception);
        }

        return taskExceptions;
    }
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