public void handleAssignment(final Map<TaskId, Set<TopicPartition>> activeTasks,
                                 final Map<TaskId, Set<TopicPartition>> standbyTasks) {
        log.info("Handle new assignment with:\n" +
                     "\tNew active tasks: {}\n" +
                     "\tNew standby tasks: {}\n" +
                     "\tExisting active tasks: {}\n" +
                     "\tExisting standby tasks: {}",
                 activeTasks.keySet(), standbyTasks.keySet(), activeTaskIds(), standbyTaskIds());

        topologyMetadata.addSubscribedTopicsFromAssignment(
            activeTasks.values().stream().flatMap(Collection::stream).collect(Collectors.toSet()),
            logPrefix
        );

        final Map<TaskId, Set<TopicPartition>> activeTasksToCreate = new HashMap<>(activeTasks);
        final Map<TaskId, Set<TopicPartition>> standbyTasksToCreate = new HashMap<>(standbyTasks);
        final Map<Task, Set<TopicPartition>> tasksToRecycle = new HashMap<>();
        final Set<Task> tasksToCloseClean = new TreeSet<>(Comparator.comparing(Task::id));

        final Set<TaskId> tasksToLock =
            tasks.allTaskIds().stream()
                .filter(x -> activeTasksToCreate.containsKey(x) || standbyTasksToCreate.containsKey(x))
                .collect(Collectors.toSet());

        maybeLockTasks(tasksToLock);

        // first put aside those unrecognized tasks because of unknown named-topologies
        tasks.clearPendingTasksToCreate();
        tasks.addPendingActiveTasksToCreate(pendingTasksToCreate(activeTasksToCreate));
        tasks.addPendingStandbyTasksToCreate(pendingTasksToCreate(standbyTasksToCreate));
        
        // first rectify all existing tasks:
        // 1. for tasks that are already owned, just update input partitions / resume and skip re-creating them
        // 2. for tasks that have changed active/standby status, just recycle and skip re-creating them
        // 3. otherwise, close them since they are no longer owned
        final Map<TaskId, RuntimeException> failedTasks = new LinkedHashMap<>();
        if (stateUpdater == null) {
            handleTasksWithoutStateUpdater(activeTasksToCreate, standbyTasksToCreate, tasksToRecycle, tasksToCloseClean);
        } else {
            handleTasksWithStateUpdater(
                activeTasksToCreate,
                standbyTasksToCreate,
                tasksToRecycle,
                tasksToCloseClean,
                failedTasks
            );
            failedTasks.putAll(collectExceptionsAndFailedTasksFromStateUpdater());
        }

        final Map<TaskId, RuntimeException> taskCloseExceptions = closeAndRecycleTasks(tasksToRecycle, tasksToCloseClean);

        maybeUnlockTasks(tasksToLock);

        failedTasks.putAll(taskCloseExceptions);
        maybeThrowTaskExceptions(failedTasks);

        createNewTasks(activeTasksToCreate, standbyTasksToCreate);
    }
@Test
    public void shouldHandleExceptionThrownDuringRecyclingStandbyTask() {
        final StandbyTask standbyTaskToRecycle = standbyTask(taskId03, taskId03ChangelogPartitions)
            .inState(State.RUNNING)
            .withInputPartitions(taskId03Partitions).build();
        final TasksRegistry tasks = mock(TasksRegistry.class);
        final TaskManager taskManager = setUpTaskManager(ProcessingMode.AT_LEAST_ONCE, tasks, true);
        when(stateUpdater.getTasks()).thenReturn(mkSet(standbyTaskToRecycle));
        when(activeTaskCreator.createActiveTaskFromStandby(
            standbyTaskToRecycle,
            standbyTaskToRecycle.inputPartitions(),
            consumer))
            .thenThrow(new RuntimeException());
        final CompletableFuture<StateUpdater.RemovedTaskResult> future = new CompletableFuture<>();
        when(stateUpdater.remove(standbyTaskToRecycle.id())).thenReturn(future);
        future.complete(new StateUpdater.RemovedTaskResult(standbyTaskToRecycle));

        assertThrows(
            StreamsException.class,
            () -> taskManager.handleAssignment(
                mkMap(mkEntry(standbyTaskToRecycle.id(), standbyTaskToRecycle.inputPartitions())),
                Collections.emptyMap()
            )
        );

        verify(stateUpdater, never()).add(any());
        verify(tasks, never()).addPendingTasksToInit(Collections.singleton(any()));
        verify(standbyTaskToRecycle).closeDirty();
    }