@Override
    public void run() {
        int numFilesCleaned = 0;
        long diskSpaceCleaned = 0L;
        try (Timer.Context t = cleanupRoutineDuration.time()) {
            final long nowMills = Time.currentTimeMillis();
            Set<Path> oldLogDirs = selectDirsForCleanup(nowMills);

            final long nowSecs = TimeUnit.MILLISECONDS.toSeconds(nowMills);
            SortedSet<Path> deadWorkerDirs = getDeadWorkerDirs((int) nowSecs, oldLogDirs);

            LOG.debug("log cleanup: now={} old log dirs {} dead worker dirs {}", nowSecs,
                    oldLogDirs.stream().map(p -> p.getFileName().toString()).collect(joining(",")),
                    deadWorkerDirs.stream().map(p -> p.getFileName().toString()).collect(joining(",")));

            for (Path dir : deadWorkerDirs) {
                Path path = dir.toAbsolutePath().normalize();
                long sizeInBytes = sizeOfDir(dir);
                LOG.info("Cleaning up: Removing {}, {} KB", path, sizeInBytes * 1e-3);

                try {
                    Utils.forceDelete(path.toString());
                    cleanupEmptyTopoDirectory(dir);
                    numFilesCleaned++;
                    diskSpaceCleaned += sizeInBytes;
                } catch (Exception ex) {
                    numFileRemovalExceptions.mark();
                    LOG.error(ex.getMessage(), ex);
                }
            }

            final List<DeletionMeta> perWorkerDirCleanupMeta = perWorkerDirCleanup(maxPerWorkerLogsSizeMb * 1024 * 1024);
            numFilesCleaned += perWorkerDirCleanupMeta.stream().mapToInt(meta -> meta.deletedFiles).sum();
            diskSpaceCleaned += perWorkerDirCleanupMeta.stream().mapToLong(meta -> meta.deletedSize).sum();
            final DeletionMeta globalLogCleanupMeta = globalLogCleanup(maxSumWorkerLogsSizeMb * 1024 * 1024);
            numFilesCleaned += globalLogCleanupMeta.deletedFiles;
            diskSpaceCleaned += globalLogCleanupMeta.deletedSize;
        } catch (Exception ex) {
            numCleanupExceptions.mark();
            LOG.error("Exception while cleaning up old log.", ex);
        }
        numFilesCleanedUp.update(numFilesCleaned);
        diskSpaceFreed.update(diskSpaceCleaned);
    }
@Test
    public void testCleanupFn() throws IOException {
        try (TmpPath dir1 = new TmpPath(); TmpPath dir2 = new TmpPath()) {
            Files.createDirectory(dir1.getFile().toPath());
            Files.createDirectory(dir2.getFile().toPath());

            Map<String, Object> conf = Utils.readStormConfig();
            StormMetricsRegistry metricRegistry = new StormMetricsRegistry();
            WorkerLogs stubbedWorkerLogs = new WorkerLogs(conf, Paths.get(""), metricRegistry);

            LogCleaner logCleaner = new LogCleaner(conf, stubbedWorkerLogs, new DirectoryCleaner(metricRegistry), null, metricRegistry) {
                @Override
                Set<Path> selectDirsForCleanup(long nowMillis) {
                    return Collections.emptySet();
                }

                @Override
                SortedSet<Path> getDeadWorkerDirs(int nowSecs, Set<Path> logDirs) {
                    SortedSet<Path> dirs = new TreeSet<>();
                    dirs.add(dir1.getFile().toPath());
                    dirs.add(dir2.getFile().toPath());
                    return dirs;
                }

                @Override
                void cleanupEmptyTopoDirectory(Path dir) {
                }
            };

            logCleaner.run();

            assertThat(Files.exists(dir1.getFile().toPath()), is(false));
            assertThat(Files.exists(dir2.getFile().toPath()), is(false));
        }
    }