Repository: apache/kafka
File: streams/src/test/java/org/apache/kafka/streams/processor/internals/DefaultStateUpdaterTest.java (Java)

Code before:
        verifyRemovedTasks(task);
        verifyUpdatingTasks();
        verifyPausedTasks();
    }

    @Test
    public void shouldResumeActiveStatefulTask() throws Exception {
        final StreamTask task = statefulTask(TASK_0_0, mkSet(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
        shouldResumeStatefulTask(task);
        verify(changelogReader, times(2)).enforceRestoreActive();
    }

    @Test
    public void shouldResumeStandbyTask() throws Exception {
        final StandbyTask task = standbyTask(TASK_0_0, mkSet(TOPIC_PARTITION_A_0)).inState(State.RUNNING).build();
        shouldResumeStatefulTask(task);
        verify(changelogReader, times(2)).transitToUpdateStandby();
    }

    private void shouldResumeStatefulTask(final Task task) throws Exception {
        stateUpdater.start();
        stateUpdater.add(task);
        verifyUpdatingTasks(task);

        when(topologyMetadata.isPaused(null)).thenReturn(true);

        verifyPausedTasks(task);
        verifyUpdatingTasks();

        when(topologyMetadata.isPaused(null)).thenReturn(false);
        stateUpdater.signalResume();

        verifyPausedTasks();
        verifyUpdatingTasks(task);
    }

    @Test
    public void shouldNotResumeNonExistingTasks() throws Exception {
        stateUpdater.start();

        verifyPausedTasks();
        verifyRestoredActiveTasks();
        verifyRemovedTasks();
        verifyUpdatingTasks();
        verifyExceptionsAndFailedTasks();
    }

    @Test
    public void shouldNotResumeActiveStatefulTaskInRestoredActiveTasks() throws Exception {
        final StreamTask task = statefulTask(TASK_0_0, mkSet(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
        final StreamTask controlTask = statefulTask(TASK_1_0, mkSet(TOPIC_PARTITION_B_0)).inState(State.RESTORING).build();

Review comment (suggestion):
I think, you do not need to reset the mock since we guarantee that paused tasks that are added to the state updater are not restored. We miss a test for that case, though.



Code after:
        verifyRemovedTasks(task);
        verifyUpdatingTasks();
        verifyPausedTasks();
    }

    @Test
    public void shouldResumeActiveStatefulTask() throws Exception {
        final StreamTask task = statefulTask(TASK_0_0, mkSet(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
        shouldResumeStatefulTask(task);
        verify(changelogReader, times(2)).enforceRestoreActive();
    }

    @Test
    public void shouldIdleWhenAllTasksPaused() throws Exception {
        final StreamTask task = statefulTask(TASK_0_0, mkSet(TOPIC_PARTITION_A_0)).inState(State.RESTORING).build();
        stateUpdater.start();
        stateUpdater.add(task);

        when(topologyMetadata.isPaused(null)).thenReturn(true);

        verifyPausedTasks(task);
        verifyIdle();

        when(topologyMetadata.isPaused(null)).thenReturn(false);
        stateUpdater.signalResume();

        verifyPausedTasks();
        verifyUpdatingTasks(task);
    }

    @Test
    public void shouldResumeStandbyTask() throws Exception {
        final StandbyTask task = standbyTask(TASK_0_0, mkSet(TOPIC_PARTITION_A_0)).inState(State.RUNNING).build();
        shouldResumeStatefulTask(task);
        verify(changelogReader, times(2)).transitToUpdateStandby();
    }

    private void shouldResumeStatefulTask(final Task task) throws Exception {
        stateUpdater.start();
        stateUpdater.add(task);
        verifyUpdatingTasks(task);

        when(topologyMetadata.isPaused(null)).thenReturn(true);

        verifyPausedTasks(task);
        verifyUpdatingTasks();

        when(topologyMetadata.isPaused(null)).thenReturn(false);
        stateUpdater.signalResume();

        verifyPausedTasks();