@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))); del.delete((Long.parseLong(p.getName()) % 2) == 0 ? null : "dingo", p, baseDirs.toArray(new Path[4])); } del.stop(); del = new DeletionService(new FakeDefaultContainerExecutor(), stateStore); del.init(conf); del.start(); int msecToWait = 10 * 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.close(); stateStore.close(); } }
public void delete(String user, Path subDir, Path... baseDirs) { if (debugDelay != -1) { List<Path> baseDirList = null; if (baseDirs != null && baseDirs.length != 0) { baseDirList = Arrays.asList(baseDirs); } FileDeletionTask task = new FileDeletionTask(this, user, subDir, baseDirList); recordDeletionTaskInStateStore(task); sched.schedule(task, debugDelay, TimeUnit.SECONDS); } }
DeletionService extends AbstractService { public void delete(String user, Path subDir, Path... baseDirs) { if (debugDelay != -1) { List<Path> baseDirList = null; if (baseDirs != null && baseDirs.length != 0) { baseDirList = Arrays.asList(baseDirs); } FileDeletionTask task = new FileDeletionTask(this, user, subDir, baseDirList); recordDeletionTaskInStateStore(task); sched.schedule(task, debugDelay, TimeUnit.SECONDS); } } }
DeletionService extends AbstractService { public void delete(String user, Path subDir, Path... baseDirs) { if (debugDelay != -1) { List<Path> baseDirList = null; if (baseDirs != null && baseDirs.length != 0) { baseDirList = Arrays.asList(baseDirs); } FileDeletionTask task = new FileDeletionTask(this, user, subDir, baseDirList); recordDeletionTaskInStateStore(task); sched.schedule(task, debugDelay, TimeUnit.SECONDS); } } DeletionService(ContainerExecutor exec);  DeletionService(ContainerExecutor exec,
      NMStateStoreService stateStore); }
DeletionService extends AbstractService { public void delete(String user, Path subDir, Path... baseDirs) { if (debugDelay != -1) { List<Path> baseDirList = null; if (baseDirs != null && baseDirs.length != 0) { baseDirList = Arrays.asList(baseDirs); } FileDeletionTask task = new FileDeletionTask(this, user, subDir, baseDirList); recordDeletionTaskInStateStore(task); sched.schedule(task, debugDelay, TimeUnit.SECONDS); } } DeletionService(ContainerExecutor exec);  DeletionService(ContainerExecutor exec,
      NMStateStoreService stateStore); void delete(String user, Path subDir, Path... baseDirs); void scheduleFileDeletionTask(FileDeletionTask fileDeletionTask); @Private boolean isTerminated(); FileDeletionTask createFileDeletionTask(String user, Path subDir,
      Path[] baseDirs); }
DeletionService extends AbstractService { public void delete(String user, Path subDir, Path... baseDirs) { if (debugDelay != -1) { List<Path> baseDirList = null; if (baseDirs != null && baseDirs.length != 0) { baseDirList = Arrays.asList(baseDirs); } FileDeletionTask task = new FileDeletionTask(this, user, subDir, baseDirList); recordDeletionTaskInStateStore(task); sched.schedule(task, debugDelay, TimeUnit.SECONDS); } } DeletionService(ContainerExecutor exec);  DeletionService(ContainerExecutor exec,
      NMStateStoreService stateStore); void delete(String user, Path subDir, Path... baseDirs); void scheduleFileDeletionTask(FileDeletionTask fileDeletionTask); @Private boolean isTerminated(); FileDeletionTask createFileDeletionTask(String user, Path subDir,
      Path[] baseDirs);  }