@VisibleForTesting
    List<DeletionMeta> perWorkerDirCleanup(long size) {
        return workerLogs.getAllWorkerDirs().stream()
                .map(dir -> {
                    try {
                        return directoryCleaner.deleteOldestWhileTooLarge(Collections.singletonList(dir), size, true, null);
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                })
                .collect(toList());
    }
@Test
    public void testPerWorkerDirectoryCleanup() throws IOException {
        long nowMillis = Time.currentTimeMillis();

        try (TmpPath testDir = new TmpPath()) {
            Files.createDirectories(testDir.getFile().toPath());
            Path rootDir = createDir(testDir.getFile().toPath(), "workers-artifacts");
            Path topo1Dir = createDir(rootDir, "topo1");
            Path topo2Dir = createDir(rootDir, "topo2");
            Path port1Dir = createDir(topo1Dir, "port1");
            Path port2Dir = createDir(topo1Dir, "port2");
            Path port3Dir = createDir(topo2Dir, "port3");

            IntStream.range(0, 10)
                .forEach(idx -> createFile(port1Dir, "A" + idx, nowMillis + 100L * idx, 200));
            IntStream.range(0, 10)
                .forEach(idx -> createFile(port2Dir, "B" + idx, nowMillis + 100L * idx, 200));
            IntStream.range(0, 10)
                .forEach(idx -> createFile(port3Dir, "C" + idx, nowMillis + 100L * idx, 200));

            Map<String, Object> conf = Utils.readStormConfig();
            StormMetricsRegistry metricRegistry = new StormMetricsRegistry();
            WorkerLogs workerLogs = new WorkerLogs(conf, rootDir, metricRegistry);
            LogCleaner logCleaner = new LogCleaner(conf, workerLogs, new DirectoryCleaner(metricRegistry), rootDir, metricRegistry);

            List<Integer> deletedFiles = logCleaner.perWorkerDirCleanup(1200)
                .stream()
                .map(deletionMeta -> deletionMeta.deletedFiles)
                .collect(toList());
            assertEquals(Integer.valueOf(4), deletedFiles.get(0));
            assertEquals(Integer.valueOf(4), deletedFiles.get(1));
            assertEquals(Integer.valueOf(4), deletedFiles.get(deletedFiles.size() - 1));
        }
    }