public AddNamedTopologyResult addNamedTopology(final NamedTopology newTopology) {
        log.info("Adding new NamedTopology: {}", newTopology.name());
        final KafkaFutureImpl<Void> future = new KafkaFutureImpl<>();

        if (hasStartedOrFinishedShuttingDown()) {
            future.completeExceptionally(
                new IllegalStateException("Cannot add a NamedTopology while the state is " + super.state)
            );
        } else if (getTopologyByName(newTopology.name()).isPresent()) {
            future.completeExceptionally(
                new IllegalArgumentException("Unable to add the new NamedTopology " + newTopology.name() +
                                                   " as another of the same name already exists")
            );
        } else {
            topologyMetadata.registerAndBuildNewTopology(future, newTopology.internalTopologyBuilder());
            maybeCompleteFutureIfStillInCREATED(future, "adding topology " + newTopology.name());
        }

        return new AddNamedTopologyResult(future);
    }
Add a new NamedTopology to a running Kafka Streams app. If multiple instances of the application are running,
you should inform all of them by calling {@code #addNamedTopology(NamedTopology)} on each client in order for
it to begin processing the new topology.
This method is not purely Async.

May complete exceptionally (does not actually directly throw) with:

@throws IllegalArgumentException if this topology name is already in use
@throws IllegalStateException    if streams has not been started or has already shut down
@throws TopologyException        if this topology subscribes to any input topics or pattern already in use
streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java