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

  conf.set(CommonConfigurationKeys.FS_PERMISSIONS_UMASK_KEY, "077");
  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());
  int[] sizes = new int[10];
  for (int i = 0; i < 10; ++i) {
    sizes[i] = rand.nextInt(512) + 512;
    LocalResourceVisibility vis = LocalResourceVisibility.PRIVATE;
    if (i%2 == 1) {
      vis = LocalResourceVisibility.APPLICATION;
    }
    Path p = new Path(basedir, "" + i);
    LocalResource rsrc = createFile(files, p, sizes[i], rand, vis);
    rsrcVis.put(rsrc, vis);
    Path destPath = dirs.getLocalPathForWrite(
        basedir.toString(), sizes[i], 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();
      assertEquals(sizes[Integer.parseInt(localized.getName())], p.getKey()
          .getSize());
      FileStatus status = files.getFileStatus(localized.getParent());
      FsPermission perm = status.getPermission();
      assertEquals("Cache directory permissions are incorrect",
          new FsPermission((short)0755), perm);
      status = files.getFileStatus(localized);
      perm = status.getPermission();
      System.out.println("File permission " + perm + 
          " for rsrc vis " + p.getKey().getVisibility().name());
      assert(rsrcVis.containsKey(p.getKey()));
      Assert.assertTrue("Private file should be 500",
          perm.toShort() == FSDownload.PRIVATE_FILE_PERMS.toShort());
    }
  } catch (ExecutionException e) {
    throw new IOException("Failed exec", e);
  }
}