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);
    }
@throws TaskMigratedException if the task producer got fenced (EOS only)
@throws StreamsException fatal error while creating / initializing the task

public for upgrade testing only
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java