@VisibleForTesting
    Predicate<Path> mkFileFilterForLogCleanup(long nowMillis) {
        //It seems safer not to follow symlinks, since we don't expect them here
        return file -> Files.isDirectory(file, LinkOption.NOFOLLOW_LINKS)
                && lastModifiedTimeWorkerLogdir(file) <= cleanupCutoffAgeMillis(nowMillis);
    }
@Test
    public void testMkFileFilterForLogCleanup() throws IOException {
        // this is to read default value for other configurations
        Map<String, Object> conf = Utils.readStormConfig();
        conf.put(LOGVIEWER_CLEANUP_AGE_MINS, 60);
        conf.put(LOGVIEWER_CLEANUP_INTERVAL_SECS, 300);

        StormMetricsRegistry metricRegistry = new StormMetricsRegistry();
        WorkerLogs workerLogs = new WorkerLogs(conf, Paths.get(""), metricRegistry);

        LogCleaner logCleaner = new LogCleaner(conf, workerLogs, new DirectoryCleaner(metricRegistry), null, metricRegistry);

        final long nowMillis = Time.currentTimeMillis();
        final long cutoffMillis = logCleaner.cleanupCutoffAgeMillis(nowMillis);
        final long oldMtimeMillis = cutoffMillis - 500;
        final long newMtimeMillis = cutoffMillis + 500;
       
        try (TmpPath testDir = new TmpPath()) {
            Files.createDirectories(testDir.getFile().toPath());

            List<Path> matchingFiles = Arrays.asList(
                createDir(testDir.getFile().toPath(), "3031", oldMtimeMillis),
                createDir(testDir.getFile().toPath(), "3032", oldMtimeMillis),
                createDir(testDir.getFile().toPath(), "7077", oldMtimeMillis)
            );
            List<Path> excludedFiles = Arrays.asList(
                createFile(testDir.getFile().toPath(), "oldlog-1-2-worker-.log", oldMtimeMillis),
                createFile(testDir.getFile().toPath(), "newlog-1-2-worker-.log", newMtimeMillis),
                createFile(testDir.getFile().toPath(), "some-old-file.txt", oldMtimeMillis),
                createFile(testDir.getFile().toPath(), "olddir-1-2-worker.log", newMtimeMillis),
                createFile(testDir.getFile().toPath(), "metadata", newMtimeMillis),
                createFile(testDir.getFile().toPath(), "newdir", newMtimeMillis)
            );

            Predicate<Path> fileFilter = logCleaner.mkFileFilterForLogCleanup(nowMillis);

            matchingFiles.forEach(p -> assertTrue(fileFilter.test(p), "Missing " + p.getFileName()));
            excludedFiles.forEach(p -> assertFalse(fileFilter.test(p), "Not excluded " + p.getFileName()));
        }
    }