DeletionService extends AbstractService { public void delete(DeletionTask deletionTask) { if (debugDelay != -1) { if (LOG.isDebugEnabled()) { String msg = String.format("Scheduling DeletionTask (delay %d) : %s", debugDelay, deletionTask.toString()); LOG.debug(msg); } recordDeletionTaskInStateStore(deletionTask); sched.schedule(deletionTask, debugDelay, TimeUnit.SECONDS); } } DeletionService(ContainerExecutor exec);  DeletionService(ContainerExecutor containerExecutor,
      NMStateStoreService stateStore); int getDebugDelay(); ContainerExecutor getContainerExecutor(); NMStateStoreService getStateStore(); void delete(DeletionTask deletionTask); @Override void serviceStop(); @Private boolean isTerminated();  }
@Test public void testAbsDelete() throws Exception { Random r = new Random(); long seed = r.nextLong(); r.setSeed(seed); System.out.println("SEED: " + seed); List<Path> dirs = buildDirs(r, base, 20); createDirs(new Path("."), dirs); FakeDefaultContainerExecutor exec = new FakeDefaultContainerExecutor(); Configuration conf = new Configuration(); exec.setConf(conf); DeletionService del = new DeletionService(exec); del.init(conf); del.start(); try { for (Path p : dirs) { FileDeletionTask deletionTask = new FileDeletionTask(del, (Long.parseLong(p.getName()) % 2) == 0 ? null : "dingo", p, null); del.delete(deletionTask); } int msecToWait = 20 * 1000; for (Path p : dirs) { while (msecToWait > 0 && lfs.util().exists(p)) { Thread.sleep(100); msecToWait -= 100; } assertFalse(lfs.util().exists(p)); } } finally { del.stop(); } }
@Test public void testRelativeDelete() throws Exception { Random r = new Random(); long seed = r.nextLong(); r.setSeed(seed); System.out.println("SEED: " + seed); List<Path> baseDirs = buildDirs(r, base, 4); createDirs(new Path("."), baseDirs); List<Path> content = buildDirs(r, new Path("."), 10); for (Path b : baseDirs) { createDirs(b, content); } DeletionService del = new DeletionService(new FakeDefaultContainerExecutor()); try { del.init(new Configuration()); del.start(); for (Path p : content) { assertTrue(lfs.util().exists(new Path(baseDirs.get(0), p))); FileDeletionTask deletionTask = new FileDeletionTask(del, (Long.parseLong(p.getName()) % 2) == 0 ? null : "dingo", p, baseDirs); del.delete(deletionTask); } int msecToWait = 20 * 1000; for (Path p : baseDirs) { for (Path q : content) { Path fp = new Path(p, q); while (msecToWait > 0 && lfs.util().exists(fp)) { Thread.sleep(100); msecToWait -= 100; } assertFalse(lfs.util().exists(fp)); } } } finally { del.stop(); } }
@Test public void testNoDelete() throws Exception { Random r = new Random(); long seed = r.nextLong(); r.setSeed(seed); System.out.println("SEED: " + seed); List<Path> dirs = buildDirs(r, base, 20); createDirs(new Path("."), dirs); FakeDefaultContainerExecutor exec = new FakeDefaultContainerExecutor(); Configuration conf = new Configuration(); conf.setInt(YarnConfiguration.DEBUG_NM_DELETE_DELAY_SEC, -1); exec.setConf(conf); DeletionService del = new DeletionService(exec); try { del.init(conf); del.start(); for (Path p : dirs) { FileDeletionTask deletionTask = new FileDeletionTask(del, (Long.parseLong(p.getName()) % 2) == 0 ? null : "dingo", p, null); del.delete(deletionTask); } int msecToWait = 20 * 1000; for (Path p : dirs) { while (msecToWait > 0 && lfs.util().exists(p)) { Thread.sleep(100); msecToWait -= 100; } assertTrue(lfs.util().exists(p)); } } finally { del.stop(); } }
@Test (timeout=60000) public void testFileDeletionTaskDependency() throws Exception { FakeDefaultContainerExecutor exec = new FakeDefaultContainerExecutor(); Configuration conf = new Configuration(); exec.setConf(conf); DeletionService del = new DeletionService(exec); del.init(conf); del.start(); try { Random r = new Random(); long seed = r.nextLong(); r.setSeed(seed); System.out.println("SEED: " + seed); List<Path> dirs = buildDirs(r, base, 2); createDirs(new Path("."), dirs); List<Path> subDirs = buildDirs(r, dirs.get(0), 2); FileDeletionTask dependentDeletionTask = new FileDeletionTask(del, null, dirs.get(0), new ArrayList<Path>()); List<FileDeletionTask> deletionTasks = new ArrayList<FileDeletionTask>(); for (Path subDir : subDirs) { List<Path> subDirList = new ArrayList<>(); subDirList.add(subDir); FileDeletionTask deletionTask = new FileDeletionTask(del, null, dirs.get(0), subDirList); deletionTask.addDeletionTaskDependency(dependentDeletionTask); deletionTasks.add(deletionTask); } for (FileDeletionTask task : deletionTasks) { del.delete(task); } int msecToWait = 20 * 1000; while (msecToWait > 0 && (lfs.util().exists(dirs.get(0)))) { Thread.sleep(100); msecToWait -= 100; } assertFalse(lfs.util().exists(dirs.get(0))); subDirs = buildDirs(r, dirs.get(1), 2); subDirs.add(new Path(dirs.get(1), "absentFile")); dependentDeletionTask = new FileDeletionTask(del, null, dirs.get(1), new ArrayList<Path>()); deletionTasks = new ArrayList<FileDeletionTask>(); for (Path subDir : subDirs) { List<Path> subDirList = new ArrayList<>(); subDirList.add(subDir); FileDeletionTask deletionTask = new FileDeletionTask(del, null, null, subDirList); deletionTask.addDeletionTaskDependency(dependentDeletionTask); deletionTasks.add(deletionTask); } deletionTasks.get(2).setSuccess(false); for (FileDeletionTask task : deletionTasks) { del.delete(task); } msecToWait = 20 * 1000; while (msecToWait > 0 && (lfs.util().exists(subDirs.get(0)) || lfs.util().exists( subDirs.get(1)))) { Thread.sleep(100); msecToWait -= 100; } assertTrue(lfs.util().exists(dirs.get(1))); } finally { del.stop(); } }
@Test public void testRecovery() throws Exception { Random r = new Random(); long seed = r.nextLong(); r.setSeed(seed); System.out.println("SEED: " + seed); List<Path> baseDirs = buildDirs(r, base, 4); createDirs(new Path("."), baseDirs); List<Path> content = buildDirs(r, new Path("."), 10); for (Path b : baseDirs) { createDirs(b, content); } Configuration conf = new YarnConfiguration(); conf.setBoolean(YarnConfiguration.NM_RECOVERY_ENABLED, true); conf.setInt(YarnConfiguration.DEBUG_NM_DELETE_DELAY_SEC, 1); NMMemoryStateStoreService stateStore = new NMMemoryStateStoreService(); stateStore.init(conf); stateStore.start(); DeletionService del = new DeletionService(new FakeDefaultContainerExecutor(), stateStore); try { del.init(conf); del.start(); for (Path p : content) { assertTrue(lfs.util().exists(new Path(baseDirs.get(0), p))); FileDeletionTask deletionTask = new FileDeletionTask(del, (Long.parseLong(p.getName()) % 2) == 0 ? null : "dingo", p, baseDirs); del.delete(deletionTask); } del.stop(); del = new DeletionService(new FakeDefaultContainerExecutor(), stateStore); del.init(conf); del.start(); int msecToWait = 10 * 1000; for (Path p : baseDirs) { System.out.println("TEST Basedir: " + p.getName()); for (Path q : content) { Path fp = new Path(p, q); System.out.println("TEST Path: " + fp.toString()); while (msecToWait > 0 && lfs.util().exists(fp)) { Thread.sleep(100); msecToWait -= 100; } assertFalse(lfs.util().exists(fp)); } } } finally { del.close(); stateStore.close(); } }