private void cleanRemovedTasksCalledByCleanerThread(final long cleanupDelayMs) {
        for (final TaskDirectory taskDir : listAllTaskDirectories()) {
            final String dirName = taskDir.file().getName();
            final TaskId id = parseTaskDirectoryName(dirName, taskDir.namedTopology());
            if (!lockedTasksToOwner.containsKey(id)) {
                try {
                    if (lock(id)) {
                        final long now = time.milliseconds();
                        final long lastModifiedMs = taskDir.file().lastModified();
                        if (now - cleanupDelayMs > lastModifiedMs) {
                            log.info("{} Deleting obsolete state directory {} for task {} as {}ms has elapsed (cleanup delay is {}ms).",
                                logPrefix(), dirName, id, now - lastModifiedMs, cleanupDelayMs);
                            Utils.delete(taskDir.file());
                        }
                    }
                } catch (final IOException exception) {
                    log.warn(
                        String.format("%s Swallowed the following exception during deletion of obsolete state directory %s for task %s:",
                            logPrefix(), dirName, id),
                        exception
                    );
                } finally {
                    unlock(id);
                }
            }
        }
        // Ok to ignore returned exception as it should be swallowed
        maybeCleanEmptyNamedTopologyDirs(true);
    }
Remove the directories for any {@link TaskId}s that are no-longer
owned by this {@link StreamThread} and aren't locked by either
another process or another {@link StreamThread}
@param cleanupDelayMs only remove directories if they haven't been modified for at least
                      this amount of time (milliseconds)
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java