private void transitRestoredTaskToRunning(final Task task,
                                              final long now,
                                              final java.util.function.Consumer<Set<TopicPartition>> offsetResetter) {
        try {
            task.completeRestoration(offsetResetter);
            tasks.addTask(task);
            mainConsumer.resume(task.inputPartitions());
            task.clearTaskTimeout();
        } catch (final TimeoutException timeoutException) {
            task.maybeInitTaskTimeoutOrThrow(now, timeoutException);
            stateUpdater.add(task);
            log.debug(
                String.format(
                    "Could not complete restoration for %s due to the following exception; adding the task " +
                        "back to the state updater and will retry",
                    task.id()),
                timeoutException
            );
        }
    }
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