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

Code before:
        final StreamTask activeTask1 = statefulTask(TASK_0_0, Set.of(TOPIC_PARTITION_B_0)).build();
        final StreamTask activeTask2 = statefulTask(TASK_0_1, Set.of(TOPIC_PARTITION_B_1)).build();
        final StandbyTask standbyTask1 = standbyTask(TASK_1_0, Set.of(TOPIC_PARTITION_A_0)).build();
        final StandbyTask standbyTask2 = standbyTask(TASK_1_1, Set.of(TOPIC_PARTITION_A_1)).build();
        tasks.addPendingTasksToInit(Set.of(activeTask1, activeTask2, standbyTask1, standbyTask2));

        final Set<Task> activeTasksToInit = tasks.drainPendingActiveTasksToInit();
        assertEquals(2, activeTasksToInit.size());
        assertTrue(activeTasksToInit.containsAll(Set.of(activeTask1, activeTask2)));
        assertFalse(activeTasksToInit.containsAll(Set.of(standbyTask1, standbyTask2)));
        assertEquals(2, tasks.pendingTasksToInit().size());
        assertTrue(tasks.hasPendingTasksToInit());
        assertTrue(tasks.pendingTasksToInit().containsAll(Set.of(standbyTask1, standbyTask2)));
    }

    @Test
    public void shouldVerifyIfPendingStandbyTaskToInitAreDrained() {
        final StreamTask activeTask1 = statefulTask(TASK_0_0, Set.of(TOPIC_PARTITION_B_0)).build();
        final StreamTask activeTask2 = statefulTask(TASK_0_1, Set.of(TOPIC_PARTITION_B_1)).build();
        final StandbyTask standbyTask1 = standbyTask(TASK_1_0, Set.of(TOPIC_PARTITION_A_0)).build();
        final StandbyTask standbyTask2 = standbyTask(TASK_1_1, Set.of(TOPIC_PARTITION_A_1)).build();
        tasks.addPendingTasksToInit(Set.of(activeTask1, activeTask2, standbyTask1, standbyTask2));

        final Set<Task> standbyTasksToInit = tasks.drainPendingStandbyTasksToInit();
        assertEquals(2, standbyTasksToInit.size());
        assertTrue(standbyTasksToInit.containsAll(Set.of(standbyTask1, standbyTask2)));
        assertFalse(standbyTasksToInit.containsAll(Set.of(activeTask1, activeTask2)));
        assertEquals(2, tasks.pendingTasksToInit().size());
        assertTrue(tasks.hasPendingTasksToInit());
        assertTrue(tasks.pendingTasksToInit().containsAll(Set.of(activeTask1, activeTask2)));
    }

    @Test
    public void shouldAddFailedTask() {
        final StreamTask activeTask1 = statefulTask(TASK_0_0, Set.of(TOPIC_PARTITION_B_0)).build();
        final StreamTask activeTask2 = statefulTask(TASK_0_1, Set.of(TOPIC_PARTITION_B_1)).build();
        tasks.addTask(activeTask2);

        tasks.addFailedTask(activeTask1);

        assertEquals(activeTask1, tasks.task(TASK_0_0));
        assertEquals(activeTask2, tasks.task(TASK_0_1));
        assertTrue(tasks.allTasks().contains(activeTask1));
        assertTrue(tasks.allTasks().contains(activeTask2));
        assertFalse(tasks.allNonFailedTasks().contains(activeTask1));
        assertTrue(tasks.allNonFailedTasks().contains(activeTask2));
    }

    @Test
    public void shouldRemoveFailedTask() {
        final StreamTask activeTask1 = statefulTask(TASK_0_0, Set.of(TOPIC_PARTITION_B_0))

Review comment (suggestion):
```suggestion

        assertEquals(2, standbyTasksToInit.size());
```

Code after:
        final StreamTask activeTask1 = statefulTask(TASK_0_0, Set.of(TOPIC_PARTITION_B_0)).build();
        final StreamTask activeTask2 = statefulTask(TASK_0_1, Set.of(TOPIC_PARTITION_B_1)).build();
        final StandbyTask standbyTask1 = standbyTask(TASK_1_0, Set.of(TOPIC_PARTITION_A_0)).build();
        final StandbyTask standbyTask2 = standbyTask(TASK_1_1, Set.of(TOPIC_PARTITION_A_1)).build();
        tasks.addPendingTasksToInit(Set.of(activeTask1, activeTask2, standbyTask1, standbyTask2));

        final Set<Task> activeTasksToInit = tasks.drainPendingActiveTasksToInit();
        assertEquals(2, activeTasksToInit.size());
        assertTrue(activeTasksToInit.containsAll(Set.of(activeTask1, activeTask2)));
        assertFalse(activeTasksToInit.containsAll(Set.of(standbyTask1, standbyTask2)));
        assertEquals(2, tasks.pendingTasksToInit().size());
        assertTrue(tasks.hasPendingTasksToInit());
        assertTrue(tasks.pendingTasksToInit().containsAll(Set.of(standbyTask1, standbyTask2)));
    }

    @Test
    public void shouldVerifyIfPendingStandbyTaskToInitAreDrained() {
        final StreamTask activeTask1 = statefulTask(TASK_0_0, Set.of(TOPIC_PARTITION_B_0)).build();
        final StreamTask activeTask2 = statefulTask(TASK_0_1, Set.of(TOPIC_PARTITION_B_1)).build();
        final StandbyTask standbyTask1 = standbyTask(TASK_1_0, Set.of(TOPIC_PARTITION_A_0)).build();
        final StandbyTask standbyTask2 = standbyTask(TASK_1_1, Set.of(TOPIC_PARTITION_A_1)).build();
        tasks.addPendingTasksToInit(Set.of(activeTask1, activeTask2, standbyTask1, standbyTask2));

        final Set<Task> standbyTasksToInit = tasks.drainPendingStandbyTasksToInit();

        assertEquals(2, standbyTasksToInit.size());
        assertTrue(standbyTasksToInit.containsAll(Set.of(standbyTask1, standbyTask2)));
        assertFalse(standbyTasksToInit.containsAll(Set.of(activeTask1, activeTask2)));
        assertEquals(2, tasks.pendingTasksToInit().size());
        assertTrue(tasks.hasPendingTasksToInit());
        assertTrue(tasks.pendingTasksToInit().containsAll(Set.of(activeTask1, activeTask2)));
    }

    @Test
    public void shouldAddFailedTask() {
        final StreamTask activeTask1 = statefulTask(TASK_0_0, Set.of(TOPIC_PARTITION_B_0)).build();
        final StreamTask activeTask2 = statefulTask(TASK_0_1, Set.of(TOPIC_PARTITION_B_1)).build();
        tasks.addTask(activeTask2);

        tasks.addFailedTask(activeTask1);

        assertEquals(activeTask1, tasks.task(TASK_0_0));
        assertEquals(activeTask2, tasks.task(TASK_0_1));
        assertTrue(tasks.allTasks().contains(activeTask1));
        assertTrue(tasks.allTasks().contains(activeTask2));
        assertFalse(tasks.allNonFailedTasks().contains(activeTask1));
        assertTrue(tasks.allNonFailedTasks().contains(activeTask2));
    }

    @Test
    public void shouldRemoveFailedTask() {