private void cleanStateAndTaskDirectoriesCalledByUser() throws Exception {
        if (!lockedTasksToOwner.isEmpty()) {
            log.warn("Found some still-locked task directories when user requested to cleaning up the state, "
                + "since Streams is not running any more these will be ignored to complete the cleanup");
        }
        final AtomicReference<Exception> firstException = new AtomicReference<>();
        for (final TaskDirectory taskDir : listAllTaskDirectories()) {
            final String dirName = taskDir.file().getName();
            final TaskId id = parseTaskDirectoryName(dirName, taskDir.namedTopology());
            try {
                log.info("{} Deleting task directory {} for {} as user calling cleanup.",
                    logPrefix(), dirName, id);

                if (lockedTasksToOwner.containsKey(id)) {
                    log.warn("{} Task {} in state directory {} was still locked by {}",
                        logPrefix(), dirName, id, lockedTasksToOwner.get(id));
                }
                Utils.delete(taskDir.file());
            } catch (final IOException exception) {
                log.error(
                    String.format("%s Failed to delete task directory %s for %s with exception:",
                        logPrefix(), dirName, id),
                    exception
                );
                firstException.compareAndSet(null, exception);
            }
        }

        firstException.compareAndSet(null, maybeCleanEmptyNamedTopologyDirs(false));

        final Exception exception = firstException.get();
        if (exception != null) {
            throw exception;
        }
    }
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