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

Code before:
        futureForRemovedFailedStatefulTaskDuringRemoval
            .completeExceptionally(new StreamsException("KABOOM!"));
        futureForRemovedFailedStandbyTaskDuringRemoval
            .completeExceptionally(new StreamsException("KABOOM!"));

        taskManager.shutdown(true);

        verify(stateUpdater).shutdown(Duration.ofMinutes(1L));
        verify(tasks).addTask(removedStatefulTask);
        verify(tasks).addTask(removedStandbyTask);
        verify(removedFailedStatefulTask).prepareCommit(false);
        verify(removedFailedStatefulTask).suspend();
        verify(removedFailedStatefulTask).closeDirty();
        verify(removedFailedStandbyTask).prepareCommit(false);
        verify(removedFailedStandbyTask).suspend();
        verify(removedFailedStandbyTask).closeDirty();
        verify(removedFailedStatefulTaskDuringRemoval).prepareCommit(false);
        verify(removedFailedStatefulTaskDuringRemoval).suspend();
        verify(removedFailedStatefulTaskDuringRemoval).closeDirty();
        verify(removedFailedStandbyTaskDuringRemoval).prepareCommit(false);
        verify(removedFailedStandbyTaskDuringRemoval).suspend();
        verify(removedFailedStandbyTaskDuringRemoval).closeDirty();
    }

    @Test
    public void shouldShutDownPendingTasksToInit() {
        final TasksRegistry tasks = mock(TasksRegistry.class);
        final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks);

        final StandbyTask standbyTask00 = standbyTask(taskId00, taskId00ChangelogPartitions)
                .inState(State.RUNNING)
                .withInputPartitions(taskId00Partitions)
                .build();

        final StreamTask activeTask01 = statefulTask(taskId01, taskId00ChangelogPartitions)
                .inState(State.RUNNING)
                .withInputPartitions(taskId00Partitions).build();

        when(tasks.drainPendingStandbyTasksToInit()).thenReturn(Set.of(standbyTask00));
        when(tasks.drainPendingActiveTasksToInit()).thenReturn(Set.of(activeTask01));

        taskManager.shutdown(true);

        verify(standbyTask00).prepareCommit(true);
        verify(standbyTask00).postCommit(true);
        verify(standbyTask00).suspend();
        verify(standbyTask00).closeClean();

        verify(activeTask01).prepareCommit(true);
        verify(activeTask01).suspend();
        verify(activeTask01).closeClean();

Review comment (suggestion):
```suggestion
    public void shouldClosePendingTasksToInitDuringShutdown() {
```

Code after:
        futureForRemovedFailedStatefulTaskDuringRemoval
            .completeExceptionally(new StreamsException("KABOOM!"));
        futureForRemovedFailedStandbyTaskDuringRemoval
            .completeExceptionally(new StreamsException("KABOOM!"));

        taskManager.shutdown(true);

        verify(stateUpdater).shutdown(Duration.ofMinutes(1L));
        verify(tasks).addTask(removedStatefulTask);
        verify(tasks).addTask(removedStandbyTask);
        verify(removedFailedStatefulTask).prepareCommit(false);
        verify(removedFailedStatefulTask).suspend();
        verify(removedFailedStatefulTask).closeDirty();
        verify(removedFailedStandbyTask).prepareCommit(false);
        verify(removedFailedStandbyTask).suspend();
        verify(removedFailedStandbyTask).closeDirty();
        verify(removedFailedStatefulTaskDuringRemoval).prepareCommit(false);
        verify(removedFailedStatefulTaskDuringRemoval).suspend();
        verify(removedFailedStatefulTaskDuringRemoval).closeDirty();
        verify(removedFailedStandbyTaskDuringRemoval).prepareCommit(false);
        verify(removedFailedStandbyTaskDuringRemoval).suspend();
        verify(removedFailedStandbyTaskDuringRemoval).closeDirty();
    }

    @Test
    public void shouldClosePendingTasksToInitDuringShutdown() {
        final TasksRegistry tasks = mock(TasksRegistry.class);
        final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks);

        final StandbyTask standbyTask00 = standbyTask(taskId00, taskId00ChangelogPartitions)
                .inState(State.CREATED)
                .withInputPartitions(taskId00Partitions)
                .build();

        final StreamTask activeTask01 = statefulTask(taskId01, taskId00ChangelogPartitions)
                .inState(State.CREATED)
                .withInputPartitions(taskId00Partitions).build();

        when(tasks.drainPendingStandbyTasksToInit()).thenReturn(Set.of(standbyTask00));
        when(tasks.drainPendingActiveTasksToInit()).thenReturn(Set.of(activeTask01));

        taskManager.shutdown(true);

        verify(standbyTask00).prepareCommit(true);
        verify(standbyTask00).postCommit(true);
        verify(standbyTask00).suspend();
        verify(standbyTask00).closeClean();

        verify(activeTask01).prepareCommit(true);
        verify(activeTask01).postCommit(true);
        verify(activeTask01).suspend();