private void buildAndVerifyTopology(final InternalTopologyBuilder builder) {
        builder.rewriteTopology(config);
        builder.buildTopology();

        final Set<String> allInputTopicsCopy = new HashSet<>(allInputTopics);

        // As we go, check each topology for overlap in the set of input topics/patterns
        final int numInputTopics = allInputTopicsCopy.size();
        final List<String> inputTopics = builder.fullSourceTopicNames();
        final Collection<String> inputPatterns = builder.allSourcePatternStrings();

        final Set<String> newInputTopics = new HashSet<>(inputTopics);
        newInputTopics.addAll(inputPatterns);

        final int numNewInputTopics = newInputTopics.size();
        allInputTopicsCopy.addAll(newInputTopics);

        if (allInputTopicsCopy.size() != numInputTopics + numNewInputTopics) {
            inputTopics.retainAll(allInputTopicsCopy);
            inputPatterns.retainAll(allInputTopicsCopy);
            log.error("Tried to add the NamedTopology {} but it had overlap with other input topics {} or patterns {}",
                      builder.topologyName(), inputTopics, inputPatterns);
            throw new TopologyException("Named Topologies may not subscribe to the same input topics or patterns");
        }

        final ProcessorTopology globalTopology = builder.buildGlobalStateTopology();
        if (globalTopology != null) {
            if (builder.topologyName() != null) {
                throw new IllegalStateException("Global state stores are not supported with Named Topologies");
            } else if (this.globalTopology != null) {
                throw new TopologyException("Topology builder had global state, but global topology has already been set");
            } else {
                this.globalTopology = globalTopology;
                globalStateStores.putAll(builder.globalStateStores());
            }
        }
        allInputTopics.addAll(newInputTopics);
    }
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