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

Code before:

    @Test
    public void shouldCommitActiveAndStandbyTasks() {
        final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
            .withInputPartitions(taskId00Partitions)
            .inState(State.RUNNING)
            .build();
        final Map<TopicPartition, OffsetAndMetadata> offsets = singletonMap(t1p0, new OffsetAndMetadata(0L, null));

        final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
            .withInputPartitions(taskId01Partitions)
            .inState(State.RUNNING)
            .build();

        when(task00.commitNeeded()).thenReturn(true);
        when(task00.prepareCommit(true)).thenReturn(offsets);
        when(task01.commitNeeded()).thenReturn(true);
        when(task01.prepareCommit(true)).thenReturn(emptyMap());

        final TasksRegistry tasks = mock(TasksRegistry.class);
        when(tasks.allTasks()).thenReturn(Set.of(task00, task01));

        final TaskManager taskManager = setUpTaskManagerWithStateUpdater(ProcessingMode.AT_LEAST_ONCE, tasks);

        assertThat(taskManager.commitAll(), equalTo(2));

        verify(task00, times(2)).commitNeeded();
        verify(task00).prepareCommit(true);
        verify(task00).postCommit(false);
        verify(task01, times(2)).commitNeeded();
        verify(task01).prepareCommit(true);
        verify(task01).postCommit(false);
        verify(consumer).commitSync(offsets);
    }

    @Test
    public void shouldCommitProvidedTasksIfNeeded() {
        final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
            .withInputPartitions(taskId00Partitions)
            .inState(State.RUNNING)
            .build();
        final Map<TopicPartition, OffsetAndMetadata> offsetsTask00 = singletonMap(t1p0, new OffsetAndMetadata(0L, null));

        final StreamTask task01 = statefulTask(taskId01, taskId01ChangelogPartitions)
            .withInputPartitions(taskId01Partitions)
            .inState(State.RUNNING)
            .build();
        final Map<TopicPartition, OffsetAndMetadata> offsetsTask01 = singletonMap(t1p1, new OffsetAndMetadata(1L, null));

        final StreamTask task02 = statefulTask(taskId02, taskId02ChangelogPartitions)
            .withInputPartitions(taskId02Partitions)

Review comment (suggestion):
This test seems to be covered already in `shouldAddNewActiveTasks`, hence removed it.

Code after:

        final StandbyTask task01 = standbyTask(taskId01, taskId01ChangelogPartitions)
            .withInputPartitions(taskId01Partitions)
            .inState(State.RUNNING)
            .build();

        when(task00.commitNeeded()).thenReturn(true);
        when(task00.prepareCommit(true)).thenReturn(offsets);
        when(task01.commitNeeded()).thenReturn(true);
        when(task01.prepareCommit(true)).thenReturn(emptyMap());

        final TasksRegistry tasks = mock(TasksRegistry.class);
        when(tasks.allTasks()).thenReturn(Set.of(task00, task01));

        final TaskManager taskManager = setUpTaskManagerWithStateUpdater(ProcessingMode.AT_LEAST_ONCE, tasks);

        assertThat(taskManager.commitAll(), equalTo(2));

        verify(task00, times(2)).commitNeeded();
        verify(task00).prepareCommit(true);
        verify(task00).postCommit(false);
        verify(task01, times(2)).commitNeeded();
        verify(task01).prepareCommit(true);
        verify(task01).postCommit(false);
        verify(consumer).commitSync(offsets);
    }

    @Test
    public void shouldCommitProvidedTasksIfNeeded() {
        final StreamTask task00 = statefulTask(taskId00, taskId00ChangelogPartitions)
            .withInputPartitions(taskId00Partitions)
            .inState(State.RUNNING)
            .build();
        final Map<TopicPartition, OffsetAndMetadata> offsetsTask00 = singletonMap(t1p0, new OffsetAndMetadata(0L, null));

        final StreamTask task01 = statefulTask(taskId01, taskId01ChangelogPartitions)
            .withInputPartitions(taskId01Partitions)
            .inState(State.RUNNING)
            .build();
        final Map<TopicPartition, OffsetAndMetadata> offsetsTask01 = singletonMap(t1p1, new OffsetAndMetadata(1L, null));

        final StreamTask task02 = statefulTask(taskId02, taskId02ChangelogPartitions)
            .withInputPartitions(taskId02Partitions)
            .inState(State.RUNNING)
            .build();

        final StandbyTask task03 = standbyTask(taskId03, taskId03ChangelogPartitions)
            .withInputPartitions(taskId03Partitions)
            .inState(State.RUNNING)
            .build();