public boolean checkStateUpdater(final long now,
                                     final java.util.function.Consumer<Set<TopicPartition>> offsetResetter) {
        addTasksToStateUpdater();
        if (stateUpdater.hasExceptionsAndFailedTasks()) {
            handleExceptionsFromStateUpdater();
        }
        if (stateUpdater.restoresActiveTasks()) {
            handleRestoredTasksFromStateUpdater(now, offsetResetter);
        }
        return !stateUpdater.restoresActiveTasks()
            && !tasks.hasPendingTasksToInit();
    }
@Test
    public void shouldRethrowStreamsExceptionFromStateUpdater() {
        final StreamTask statefulTask = statefulTask(taskId00, taskId00ChangelogPartitions)
            .inState(State.RESTORING)
            .withInputPartitions(taskId00Partitions).build();
        final StreamsException exception = new StreamsException("boom!");
        final ExceptionAndTask exceptionAndTasks = new ExceptionAndTask(exception, statefulTask);
        when(stateUpdater.hasExceptionsAndFailedTasks()).thenReturn(true);
        when(stateUpdater.drainExceptionsAndFailedTasks()).thenReturn(Collections.singletonList(exceptionAndTasks));

        final TasksRegistry tasks = mock(TasksRegistry.class);
        final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);

        final StreamsException thrown = assertThrows(
            StreamsException.class,
            () -> taskManager.checkStateUpdater(time.milliseconds(), noOpResetter)
        );

        assertEquals(exception, thrown);
        assertEquals(statefulTask.id(), thrown.taskId().get());
    }