public void registerAndBuildNewTopology(final KafkaFutureImpl<Void> future, final InternalTopologyBuilder newTopologyBuilder) {
        try {
            lock();
            buildAndVerifyTopology(newTopologyBuilder);
            log.info("New NamedTopology {} passed validation and will be added, old topology version is {}", newTopologyBuilder.topologyName(), version.topologyVersion.get());
            version.topologyVersion.incrementAndGet();
            version.activeTopologyUpdateListeners.add(new TopologyVersionListener(topologyVersion(), future));
            builders.put(newTopologyBuilder.topologyName(), newTopologyBuilder);
            wakeupThreads();
            log.info("Added NamedTopology {} and updated topology version to {}", newTopologyBuilder.topologyName(), version.topologyVersion.get());
        } catch (final Throwable throwable) {
            log.error("Failed to add NamedTopology {}, please retry the operation.", newTopologyBuilder.topologyName());
            future.completeExceptionally(throwable);
        } finally {
            unlock();
        }
    }
Adds 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