void handleLostAll() {
        log.debug("Closing lost active tasks as zombies.");

        closeRunningTasksDirty();
        removeLostActiveTasksFromStateUpdaterAndPendingTasksToInit();

        if (processingMode == EXACTLY_ONCE_V2) {
            activeTaskCreator.reInitializeThreadProducer();
        }
    }
@Test
    public void shouldCloseCleanTasksPendingInitOnPartitionLost() {
        final StreamTask task1 = statefulTask(taskId00, taskId00ChangelogPartitions)
            .inState(State.CREATED)
            .withInputPartitions(taskId00Partitions).build();
        final StreamTask task2 = statefulTask(taskId02, taskId02ChangelogPartitions)
            .inState(State.CREATED)
            .withInputPartitions(taskId02Partitions).build();
        final TasksRegistry tasks = mock(TasksRegistry.class);
        when(tasks.drainPendingActiveTasksToInit()).thenReturn(mkSet(task1, task2));
        final TaskManager taskManager = setupForRevocationAndLost(emptySet(), tasks);

        taskManager.handleLostAll();

        verify(task1).suspend();
        verify(task1).closeClean();
        verify(task2).suspend();
        verify(task2).closeClean();
    }