@Test public void shouldNotRemoveNonTaskDirectoriesAndFiles() throws Exception { final File otherDir = TestUtils.tempDirectory(stateDir.toPath(), "foo"); directory.cleanRemovedTasks(0); assertTrue(otherDir.exists()); }
public void cleanRemovedTasks(final long cleanupDelayMs) { final File[] taskDirs = listTaskDirectories(); if (taskDirs == null || taskDirs.length == 0) { return; } for (File taskDir : taskDirs) { final String dirName = taskDir.getName(); TaskId id = TaskId.parse(dirName); if (!locks.containsKey(id)) { try { if (lock(id, 0)) { if (time.milliseconds() > taskDir.lastModified() + cleanupDelayMs) { log.info("{} Deleting obsolete state directory {} for task {} as cleanup delay of {} ms has passed", logPrefix, dirName, id, cleanupDelayMs); Utils.delete(taskDir); } } } catch (OverlappingFileLockException e) { } catch (IOException e) { log.error("{} Failed to lock the state directory due to an unexpected exception", logPrefix, e); } finally { try { unlock(id); } catch (IOException e) { log.error("{} Failed to release the state directory lock", logPrefix); } } } } }
StateDirectory { public void cleanRemovedTasks(final long cleanupDelayMs) { final File[] taskDirs = listTaskDirectories(); if (taskDirs == null || taskDirs.length == 0) { return; } for (File taskDir : taskDirs) { final String dirName = taskDir.getName(); TaskId id = TaskId.parse(dirName); if (!locks.containsKey(id)) { try { if (lock(id, 0)) { if (time.milliseconds() > taskDir.lastModified() + cleanupDelayMs) { log.info("{} Deleting obsolete state directory {} for task {} as cleanup delay of {} ms has passed", logPrefix, dirName, id, cleanupDelayMs); Utils.delete(taskDir); } } } catch (OverlappingFileLockException e) { } catch (IOException e) { log.error("{} Failed to lock the state directory due to an unexpected exception", logPrefix, e); } finally { try { unlock(id); } catch (IOException e) { log.error("{} Failed to release the state directory lock", logPrefix); } } } } } }
StateDirectory { public void cleanRemovedTasks(final long cleanupDelayMs) { final File[] taskDirs = listTaskDirectories(); if (taskDirs == null || taskDirs.length == 0) { return; } for (File taskDir : taskDirs) { final String dirName = taskDir.getName(); TaskId id = TaskId.parse(dirName); if (!locks.containsKey(id)) { try { if (lock(id, 0)) { if (time.milliseconds() > taskDir.lastModified() + cleanupDelayMs) { log.info("{} Deleting obsolete state directory {} for task {} as cleanup delay of {} ms has passed", logPrefix, dirName, id, cleanupDelayMs); Utils.delete(taskDir); } } } catch (OverlappingFileLockException e) { } catch (IOException e) { log.error("{} Failed to lock the state directory due to an unexpected exception", logPrefix, e); } finally { try { unlock(id); } catch (IOException e) { log.error("{} Failed to release the state directory lock", logPrefix); } } } } } StateDirectory(final String applicationId, final String stateDirConfig, final Time time);  StateDirectory(final String applicationId, final String threadId, final String stateDirConfig, final Time time); }
StateDirectory { public void cleanRemovedTasks(final long cleanupDelayMs) { final File[] taskDirs = listTaskDirectories(); if (taskDirs == null || taskDirs.length == 0) { return; } for (File taskDir : taskDirs) { final String dirName = taskDir.getName(); TaskId id = TaskId.parse(dirName); if (!locks.containsKey(id)) { try { if (lock(id, 0)) { if (time.milliseconds() > taskDir.lastModified() + cleanupDelayMs) { log.info("{} Deleting obsolete state directory {} for task {} as cleanup delay of {} ms has passed", logPrefix, dirName, id, cleanupDelayMs); Utils.delete(taskDir); } } } catch (OverlappingFileLockException e) { } catch (IOException e) { log.error("{} Failed to lock the state directory due to an unexpected exception", logPrefix, e); } finally { try { unlock(id); } catch (IOException e) { log.error("{} Failed to release the state directory lock", logPrefix); } } } } } StateDirectory(final String applicationId, final String stateDirConfig, final Time time);  StateDirectory(final String applicationId, final String threadId, final String stateDirConfig, final Time time); void cleanRemovedTasks(final long cleanupDelayMs); }
StateDirectory { public void cleanRemovedTasks(final long cleanupDelayMs) { final File[] taskDirs = listTaskDirectories(); if (taskDirs == null || taskDirs.length == 0) { return; } for (File taskDir : taskDirs) { final String dirName = taskDir.getName(); TaskId id = TaskId.parse(dirName); if (!locks.containsKey(id)) { try { if (lock(id, 0)) { if (time.milliseconds() > taskDir.lastModified() + cleanupDelayMs) { log.info("{} Deleting obsolete state directory {} for task {} as cleanup delay of {} ms has passed", logPrefix, dirName, id, cleanupDelayMs); Utils.delete(taskDir); } } } catch (OverlappingFileLockException e) { } catch (IOException e) { log.error("{} Failed to lock the state directory due to an unexpected exception", logPrefix, e); } finally { try { unlock(id); } catch (IOException e) { log.error("{} Failed to release the state directory lock", logPrefix); } } } } } StateDirectory(final String applicationId, final String stateDirConfig, final Time time);  StateDirectory(final String applicationId, final String threadId, final String stateDirConfig, final Time time); void cleanRemovedTasks(final long cleanupDelayMs);  }