public synchronized void clean() {
        try {
            cleanStateAndTaskDirectoriesCalledByUser();
        } catch (final Exception e) {
            throw new StreamsException(e);
        }

        try {
            if (stateDir.exists()) {
                Utils.delete(globalStateDir().getAbsoluteFile());
            }
        } catch (final IOException exception) {
            log.error(
                String.format("%s Failed to delete global state directory of %s due to an unexpected exception",
                    logPrefix(), appId),
                exception
            );
            throw new StreamsException(exception);
        }

        try {
            if (hasPersistentStores && stateDir.exists() && !stateDir.delete()) {
                log.warn(
                    String.format("%s Failed to delete state store directory of %s for it is not empty",
                        logPrefix(), stateDir.getAbsolutePath())
                );
            }
        } catch (final SecurityException exception) {
            log.error(
                String.format("%s Failed to delete state store directory of %s due to an unexpected exception",
                    logPrefix(), stateDir.getAbsolutePath()),
                exception
            );
            throw new StreamsException(exception);
        }
    }
@Test
    public void shouldNotRemoveDirsThatDoNotMatchNamedTopologyDirsWhenCallingClean() throws IOException {
        initializeStateDirectory(true, true);
        final File someDir = new File(appDir, "_not-a-valid-named-topology_dir_name_");
        assertThat(someDir.mkdir(), is(true));
        assertThat(someDir.exists(), is(true));
        directory.clean();
        assertThat(someDir.exists(), is(true));
    }