void maybeCloseTasksFromRemovedTopologies(final Set<String> currentNamedTopologies) {
        try {
            final Set<Task> activeTasksToRemove = new HashSet<>();
            final Set<Task> standbyTasksToRemove = new HashSet<>();
            for (final Task task : tasks.allTasks()) {
                if (!currentNamedTopologies.contains(task.id().topologyName())) {
                    if (task.isActive()) {
                        activeTasksToRemove.add(task);
                    } else {
                        standbyTasksToRemove.add(task);
                    }
                }
            }

            final Set<Task> allTasksToRemove = union(HashSet::new, activeTasksToRemove, standbyTasksToRemove);
            closeAndCleanUpTasks(activeTasksToRemove, standbyTasksToRemove, true);
            releaseLockedDirectoriesForTasks(allTasksToRemove.stream().map(Task::id).collect(Collectors.toSet()));
        } catch (final Exception e) {
            // TODO KAFKA-12648: for now just swallow the exception to avoid interfering with the other topologies
            //  that are running alongside, but eventually we should be able to rethrow up to the handler to inform
            //  the user of an error in this named topology without killing the thread and delaying the others
            log.error("Caught the following exception while closing tasks from a removed topology:", e);
        }
    }
Handle any added or removed NamedTopologies. Check if any uncreated assigned tasks belong to a newly
added NamedTopology and create them if so, then close any tasks whose named topology no longer exists
streams/src/main/java/org/apache/kafka/streams/processor/internals/TaskManager.java