public synchronized void cleanRemovedTasks(final long cleanupDelayMs) {
        try {
            cleanRemovedTasksCalledByCleanerThread(cleanupDelayMs);
        } catch (final Exception cannotHappen) {
            throw new IllegalStateException("Should have swallowed exception.", cannotHappen);
        }
    }
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