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

        closeRunningTasksDirty();
        removeLostActiveTasksFromStateUpdaterAndPendingTasksToInit();

        if (processingMode == EXACTLY_ONCE_V2) {
            activeTaskCreator.reInitializeThreadProducer();
        }
    }
@Test
    public void shouldCloseDirtyWhenRemoveFailedActiveTasksFromStateUpdaterOnPartitionLost() {
        final StreamTask task1 = statefulTask(taskId00, taskId00ChangelogPartitions)
            .inState(State.RESTORING)
            .withInputPartitions(taskId00Partitions).build();
        final StreamTask task2 = statefulTask(taskId02, taskId02ChangelogPartitions)
            .inState(State.RESTORING)
            .withInputPartitions(taskId02Partitions).build();
        final TasksRegistry tasks = mock(TasksRegistry.class);
        final TaskManager taskManager = setupForRevocationAndLost(mkSet(task1, task2), tasks);
        final CompletableFuture<StateUpdater.RemovedTaskResult> future1 = new CompletableFuture<>();
        when(stateUpdater.remove(task1.id())).thenReturn(future1);
        future1.complete(new StateUpdater.RemovedTaskResult(task1, new StreamsException("Something happened")));
        final CompletableFuture<StateUpdater.RemovedTaskResult> future3 = new CompletableFuture<>();
        when(stateUpdater.remove(task2.id())).thenReturn(future3);
        future3.complete(new StateUpdater.RemovedTaskResult(task2, new StreamsException("Something else happened")));

        taskManager.handleLostAll();

        verify(task1).prepareCommit();
        verify(task1).suspend();
        verify(task1).closeDirty();
        verify(task2).prepareCommit();
        verify(task2).suspend();
        verify(task2).closeDirty();
    }