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
        }
    }
@Test
    public void shouldRemoveEmptyNamedTopologyDirsWhenCallingClearLocalStateForNamedTopology() throws IOException {
        initializeStateDirectory(true, true);
        final String topologyName = "topology1";
        final File namedTopologyDir = new File(appDir, "__" + topologyName + "__");
        assertThat(namedTopologyDir.mkdir(), is(true));
        assertThat(namedTopologyDir.exists(), is(true));
        directory.clearLocalStateForNamedTopology(topologyName);
        assertThat(namedTopologyDir.exists(), is(false));
    }