public KafkaFuture<Void> unregisterTopology(final KafkaFutureImpl<Void> removeTopologyFuture,
                                                final String topologyName) {
        try {
            lock();
            log.info("Beginning removal of NamedTopology {}, old topology version is {}", topologyName, version.topologyVersion.get());
            version.topologyVersion.incrementAndGet();
            version.activeTopologyUpdateListeners.add(new TopologyVersionListener(topologyVersion(), removeTopologyFuture));
            final InternalTopologyBuilder removedBuilder = builders.remove(topologyName);
            removedBuilder.fullSourceTopicNames().forEach(allInputTopics::remove);
            removedBuilder.allSourcePatternStrings().forEach(allInputTopics::remove);
            log.info("Finished removing NamedTopology {}, topology version was updated to {}", topologyName, version.topologyVersion.get());
        } catch (final Throwable throwable) {
            log.error("Failed to remove NamedTopology {}, please retry.", topologyName);
            removeTopologyFuture.completeExceptionally(throwable);
        } finally {
            unlock();
        }
        return removeTopologyFuture;
    }
Removes the topology and registers a future that listens for all threads on the older version to see the update
streams/src/main/java/org/apache/kafka/streams/processor/internals/TopologyMetadata.java