void handleTopologyUpdates() {
        topologyMetadata.executeTopologyUpdatesAndBumpThreadVersion(
            this::createPendingTasks,
            this::maybeCloseTasksFromRemovedTopologies
        );

        if (topologyMetadata.isEmpty()) {
            log.info("Proactively unsubscribing from all topics due to empty topology");
            mainConsumer.unsubscribe();
        }

        topologyMetadata.maybeNotifyTopologyVersionListeners();
    }
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