public void cleanUpNamedTopology(final String name) {
        if (getTopologyByName(name).isPresent()) {
            throw new IllegalStateException("Can't clean up local state for an active NamedTopology: " + name);
        }
        stateDirectory.clearLocalStateForNamedTopology(name);
    }
Do a clean up of the local state directory for this NamedTopology by deleting all data with regard to the
@link StreamsConfig#APPLICATION_ID_CONFIG application ID} in the ({@link StreamsConfig#STATE_DIR_CONFIG})
<p>
May be called while the Streams is in any state, but only on a {@link NamedTopology} that has already been
removed via {@link #removeNamedTopology(String)}.
<p>
Calling this method triggers a restore of local {@link StateStore}s for this {@link NamedTopology} if it is
ever re-added via {@link #addNamedTopology(NamedTopology)}.

@throws IllegalStateException if this {@code NamedTopology} hasn't been removed
@throws StreamsException if cleanup failed
streams/src/main/java/org/apache/kafka/streams/processor/internals/namedtopology/KafkaStreamsNamedTopologyWrapper.java