@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);
    
    // first we will try to delete sub directories which are present. This
    // should then trigger parent directory to be deleted.
    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)));
    
    // Now we will try to delete sub directories; one of the deletion task we
    // will mark as failure and then parent directory should not be deleted.
    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);
    }
    // marking one of the tasks as a failure.
    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();
  }
}