@Test (timeout=10000)
public void testDirDownload() throws IOException, InterruptedException {

  FileContext files = FileContext.getLocalFSFileContext(conf);
  final Path basedir = files.makeQualified(new Path("target",
    TestFSDownload.class.getSimpleName()));
  files.mkdir(basedir, null, true);
  conf.setStrings(TestFSDownload.class.getName(), basedir.toString());
  
  Map<LocalResource, LocalResourceVisibility> rsrcVis =
      new HashMap<LocalResource, LocalResourceVisibility>();

  Random rand = new Random();
  long sharedSeed = rand.nextLong();
  rand.setSeed(sharedSeed);
  System.out.println("SEED: " + sharedSeed);
  Map<LocalResource,Future<Path>> pending =
    new HashMap<LocalResource,Future<Path>>();
  ExecutorService exec = HadoopExecutors.newSingleThreadExecutor();
  LocalDirAllocator dirs =
    new LocalDirAllocator(TestFSDownload.class.getName());
  for (int i = 0; i < 5; ++i) {
    LocalResourceVisibility vis = LocalResourceVisibility.PRIVATE;
    if (i%2 == 1) {
      vis = LocalResourceVisibility.APPLICATION;
    }
    Path p = new Path(basedir, "dir" + i + ".jar");
    LocalResource rsrc = createJar(files, p, vis);
    rsrcVis.put(rsrc, vis);
    Path destPath = dirs.getLocalPathForWrite(
        basedir.toString(), conf);
    destPath = new Path (destPath,
        Long.toString(uniqueNumberGenerator.incrementAndGet()));
    FSDownload fsd =
        new FSDownload(files, UserGroupInformation.getCurrentUser(), conf,
            destPath, rsrc);
    pending.put(rsrc, exec.submit(fsd));
  }
  exec.shutdown();
  while (!exec.awaitTermination(1000, TimeUnit.MILLISECONDS));
  for (Future<Path> path: pending.values()) {
    Assert.assertTrue(path.isDone());
  }
  try {
    
    for (Map.Entry<LocalResource,Future<Path>> p : pending.entrySet()) {
      Path localized = p.getValue().get();
      FileStatus status = files.getFileStatus(localized);
      System.out.println("Testing path " + localized);
      assert(status.isDirectory());
      assert(rsrcVis.containsKey(p.getKey()));
      
      verifyPermsRecursively(localized.getFileSystem(conf),
          files, localized, rsrcVis.get(p.getKey()));
    }
  } catch (ExecutionException e) {
    throw new IOException("Failed exec", e);
  }
}