public SortedSet<Path> getLogDirs(Set<Path> logDirs, Predicate<String> predicate) {
        // we could also make this static, but not to do it due to mock
        TreeSet<Path> ret = new TreeSet<>();
        for (Path logDir: logDirs) {
            String workerId = "";
            try {
                Optional<Path> metaFile = getMetadataFileForWorkerLogDir(logDir);
                if (metaFile.isPresent()) {
                    workerId = getWorkerIdFromMetadataFile(metaFile.get().toAbsolutePath().normalize());
                    if (workerId == null) {
                        workerId = "";
                    }
                }
            } catch (IOException e) {
                LOG.warn("Error trying to find worker.yaml in {}", logDir, e);
            }
            if (predicate.test(workerId)) {
                ret.add(logDir);
            }
        }
        return ret;
    }
@Test
    public void testIdentifyWorkerLogDirs() throws Exception {
        try (TmpPath testDir = new TmpPath()) {
            Path port1Dir = Files.createDirectories(testDir.getFile().toPath().resolve("workers-artifacts/topo1/port1"));
            Path metaFile = Files.createFile(testDir.getFile().toPath().resolve("worker.yaml"));

            String expId = "id12345";
            SortedSet<Path> expected = new TreeSet<>();
            expected.add(port1Dir);
            SupervisorUtils mockedSupervisorUtils = mock(SupervisorUtils.class);
            SupervisorUtils.setInstance(mockedSupervisorUtils);

            Map<String, Object> stormConf = Utils.readStormConfig();
            WorkerLogs workerLogs = new WorkerLogs(stormConf, port1Dir, new StormMetricsRegistry()) {
                @Override
                public Optional<Path> getMetadataFileForWorkerLogDir(Path logDir) {
                    return Optional.of(metaFile);
                }

                @Override
                public String getWorkerIdFromMetadataFile(Path metaFile) {
                    return expId;
                }
            };

            when(mockedSupervisorUtils.readWorkerHeartbeatsImpl(anyMap())).thenReturn(null);
            assertEquals(expected, workerLogs.getLogDirs(Collections.singleton(port1Dir), (wid) -> true));
        } finally {
            SupervisorUtils.resetInstance();
        }
    }