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);
        }
    }
Unlock the state directory for the given {@link TaskId}.
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java