public void clearLocalStateForNamedTopology(final String topologyName) {
        final File namedTopologyDir = new File(stateDir, getNamedTopologyDirName(topologyName));
        if (!namedTopologyDir.exists() || !namedTopologyDir.isDirectory()) {
            log.debug("Tried to clear out the local state for NamedTopology {} but none was found", topologyName);
        }
        try {
            Utils.delete(namedTopologyDir);
        } catch (final IOException e) {
            log.error("Hit an unexpected error while clearing local state for topology " + topologyName, e);
            throw new StreamsException("Unable to delete state for the named topology " + topologyName,
                                       e, new TaskId(-1, -1, topologyName)); // use dummy taskid to report source topology for this error
        }
    }
Clears out any local state found for the given NamedTopology after it was removed

@throws StreamsException if cleanup failed
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java