CleanerTask implements Runnable { void processSingleResource(FileStatus resource) { Path path = resource.getPath(); ResourceStatus resourceStatus = ResourceStatus.INIT; if (path.toString().endsWith(RENAMED_SUFFIX)) { LOG.info("Found a renamed directory that was left undeleted at " + path.toString() + ". Deleting."); try { if (fs.delete(path, true)) { resourceStatus = ResourceStatus.DELETED; } } catch (IOException e) { LOG.error("Error while processing a shared cache resource: " + path, e); } } else { String key = path.getName(); try { store.cleanResourceReferences(key); } catch (YarnException e) { LOG.error("Exception thrown while removing dead appIds.", e); } if (store.isResourceEvictable(key, resource)) { try { if (store.removeResource(key)) { boolean deleted = removeResourceFromCacheFileSystem(path); if (deleted) { resourceStatus = ResourceStatus.DELETED; } else { LOG.error("Failed to remove path from the file system." + " Skipping this resource: " + path); resourceStatus = ResourceStatus.ERROR; } } else { resourceStatus = ResourceStatus.PROCESSED; } } catch (IOException e) { LOG.error( "Failed to remove path from the file system. Skipping this resource: " + path, e); resourceStatus = ResourceStatus.ERROR; } } else { resourceStatus = ResourceStatus.PROCESSED; } } switch (resourceStatus) { case DELETED: metrics.reportAFileDelete(); break; case PROCESSED: metrics.reportAFileProcess(); break; case ERROR: metrics.reportAFileError(); break; default: LOG.error("Cleaner encountered an invalid status (" + resourceStatus + ") while processing resource: " + path.getName()); } } CleanerTask(String location, long sleepTime, int nestedLevel, FileSystem fs,
      SCMStore store, CleanerMetrics metrics, Lock cleanerTaskLock); static CleanerTask create(Configuration conf, SCMStore store,
      CleanerMetrics metrics, Lock cleanerTaskLock); @Override void run();  }
@Test public void testProcessFreshResource() throws Exception { FileSystem fs = mock(FileSystem.class); CleanerMetrics metrics = mock(CleanerMetrics.class); SCMStore store = mock(SCMStore.class); CleanerTask task = createSpiedTask(fs, store, metrics, new ReentrantLock()); when(store.isResourceEvictable(isA(String.class), isA(FileStatus.class))) .thenReturn(false); FileStatus status = mock(FileStatus.class); when(status.getPath()).thenReturn(new Path(ROOT + "/a/b/c/abc")); task.processSingleResource(status); verify(fs, never()).rename(eq(status.getPath()), isA(Path.class)); verify(metrics).reportAFileProcess(); verify(metrics, never()).reportAFileDelete(); }
@Test public void testProcessEvictableResource() throws Exception { FileSystem fs = mock(FileSystem.class); CleanerMetrics metrics = mock(CleanerMetrics.class); SCMStore store = mock(SCMStore.class); CleanerTask task = createSpiedTask(fs, store, metrics, new ReentrantLock()); when(store.isResourceEvictable(isA(String.class), isA(FileStatus.class))) .thenReturn(true); FileStatus status = mock(FileStatus.class); when(status.getPath()).thenReturn(new Path(ROOT + "/a/b/c/abc")); when(store.removeResource(isA(String.class))).thenReturn(true); when(fs.rename(isA(Path.class), isA(Path.class))).thenReturn(true); when(fs.delete(isA(Path.class), anyBoolean())).thenReturn(true); task.processSingleResource(status); verify(fs).rename(eq(status.getPath()), isA(Path.class)); verify(metrics).reportAFileDelete(); verify(metrics, never()).reportAFileProcess(); }
@Test public void testResourceIsInUseHasAnActiveApp() throws Exception { FileSystem fs = mock(FileSystem.class); CleanerMetrics metrics = mock(CleanerMetrics.class); SCMStore store = mock(SCMStore.class); FileStatus resource = mock(FileStatus.class); when(resource.getPath()).thenReturn(new Path(ROOT + "/a/b/c/abc")); when(store.isResourceEvictable(isA(String.class), isA(FileStatus.class))) .thenReturn(true); when(store.removeResource(isA(String.class))).thenReturn(false); CleanerTask task = createSpiedTask(fs, store, metrics, new ReentrantLock()); task.processSingleResource(resource); verify(metrics).reportAFileProcess(); verify(metrics, never()).reportAFileDelete(); }