private IOException maybeCleanEmptyNamedTopologyDirs(final boolean logExceptionAsWarn) {
        if (!hasNamedTopologies) {
            return null;
        }

        final AtomicReference<IOException> firstException = new AtomicReference<>(null);
        final File[] namedTopologyDirs = stateDir.listFiles(pathname ->
                pathname.isDirectory() && NAMED_TOPOLOGY_DIR_PATH_NAME.matcher(pathname.getName()).matches()
        );
        if (namedTopologyDirs != null) {
            for (final File namedTopologyDir : namedTopologyDirs) {
                final File[] contents = namedTopologyDir.listFiles();
                if (contents != null && contents.length == 0) {
                    try {
                        Utils.delete(namedTopologyDir);
                    } catch (final IOException exception) {
                        if (logExceptionAsWarn) {
                            log.warn(
                                String.format("%sSwallowed the following exception during deletion of named topology directory %s",
                                    logPrefix(), namedTopologyDir.getName()),
                                exception
                            );
                        } else {
                            log.error(
                                String.format("%s Failed to delete named topology directory %s with exception:",
                                    logPrefix(), namedTopologyDir.getName()),
                                exception
                            );
                        }
                        firstException.compareAndSet(null, exception);
                    }
                }
            }
        }
        return firstException.get();
    }
Cleans up any leftover named topology directories that are empty, if any exist
@param logExceptionAsWarn if true, an exception will be logged as a warning
                      if false, an exception will be logged as error
@return the first IOException to be encountered
streams/src/main/java/org/apache/kafka/streams/processor/internals/StateDirectory.java