/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements.  See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership.  The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License.  You may obtain a copy of the License at
*
*     http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.apache.hadoop.yarn.util;

import static org.apache.hadoop.fs.CreateFlag.CREATE;
import static org.apache.hadoop.fs.CreateFlag.OVERWRITE;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertSame;
import static org.junit.Assert.assertTrue;
import static org.junit.Assume.assumeTrue;

import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.net.URISyntaxException;
import java.util.ArrayList;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Random;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.jar.JarEntry;
import java.util.jar.JarOutputStream;
import java.util.jar.Manifest;
import java.util.zip.GZIPOutputStream;
import java.util.zip.ZipEntry;
import java.util.zip.ZipOutputStream;

import org.junit.Assert;

import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
import org.apache.commons.compress.archivers.tar.TarArchiveOutputStream;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.CommonConfigurationKeys;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocalDirAllocator;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.hadoop.yarn.api.records.LocalResourceType;
import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
import org.apache.hadoop.yarn.factories.RecordFactory;
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
import org.junit.AfterClass;
import org.junit.Test;

import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;

public class TestFSDownload {

  private static final Log LOG = LogFactory.getLog(TestFSDownload.class);
  private static AtomicLong uniqueNumberGenerator =
    new AtomicLong(System.currentTimeMillis());
  private enum TEST_FILE_TYPE {
    TAR, JAR, ZIP, TGZ
  };
  
  @AfterClass
  public static void deleteTestDir() throws IOException {
    FileContext fs = FileContext.getLocalFSFileContext();
    fs.delete(new Path("target", TestFSDownload.class.getSimpleName()), true);
  }

  static final RecordFactory recordFactory =
    RecordFactoryProvider.getRecordFactory(null);

  static LocalResource createFile(FileContext files, Path p, int len,
      Random r, LocalResourceVisibility vis) throws IOException {
    createFile(files, p, len, r);
    LocalResource ret = recordFactory.newRecordInstance(LocalResource.class);
    ret.setResource(ConverterUtils.getYarnUrlFromPath(p));
    ret.setSize(len);
    ret.setType(LocalResourceType.FILE);
    ret.setVisibility(vis);
    ret.setTimestamp(files.getFileStatus(p).getModificationTime());
    return ret;
  }

  static void createFile(FileContext files, Path p, int len, Random r)
      throws IOException {
    FSDataOutputStream out = null;
    try {
      byte[] bytes = new byte[len];
      out = files.create(p, EnumSet.of(CREATE, OVERWRITE));
      r.nextBytes(bytes);
      out.write(bytes);
    } finally {
      if (out != null) out.close();
    }
  }

  static LocalResource createJar(FileContext files, Path p,
      LocalResourceVisibility vis) throws IOException {
    LOG.info("Create jar file " + p);
    File jarFile = new File((files.makeQualified(p)).toUri());
    FileOutputStream stream = new FileOutputStream(jarFile);
    LOG.info("Create jar out stream ");
    JarOutputStream out = new JarOutputStream(stream, new Manifest());
    LOG.info("Done writing jar stream ");
    out.close();
    LocalResource ret = recordFactory.newRecordInstance(LocalResource.class);
    ret.setResource(ConverterUtils.getYarnUrlFromPath(p));
    FileStatus status = files.getFileStatus(p);
    ret.setSize(status.getLen());
    ret.setTimestamp(status.getModificationTime());
    ret.setType(LocalResourceType.PATTERN);
    ret.setVisibility(vis);
    ret.setPattern("classes/.*");
    return ret;
  }

  static LocalResource createTarFile(FileContext files, Path p, int len,
      Random r, LocalResourceVisibility vis) throws IOException,
      URISyntaxException {
    byte[] bytes = new byte[len];
    r.nextBytes(bytes);

    File archiveFile = new File(p.toUri().getPath() + ".tar");
    archiveFile.createNewFile();
    TarArchiveOutputStream out = new TarArchiveOutputStream(
        new FileOutputStream(archiveFile));
    TarArchiveEntry entry = new TarArchiveEntry(p.getName());
    entry.setSize(bytes.length);
    out.putArchiveEntry(entry);
    out.write(bytes);
    out.closeArchiveEntry();
    out.close();

    LocalResource ret = recordFactory.newRecordInstance(LocalResource.class);
    ret.setResource(ConverterUtils.getYarnUrlFromPath(new Path(p.toString()
        + ".tar")));
    ret.setSize(len);
    ret.setType(LocalResourceType.ARCHIVE);
    ret.setVisibility(vis);
    ret.setTimestamp(files.getFileStatus(new Path(p.toString() + ".tar"))
        .getModificationTime());
    return ret;
  }

  static LocalResource createTgzFile(FileContext files, Path p, int len,
      Random r, LocalResourceVisibility vis) throws IOException,
      URISyntaxException {
    byte[] bytes = new byte[len];
    r.nextBytes(bytes);

    File gzipFile = new File(p.toUri().getPath() + ".tar.gz");
    gzipFile.createNewFile();
    TarArchiveOutputStream out = new TarArchiveOutputStream(
        new GZIPOutputStream(new FileOutputStream(gzipFile)));
    TarArchiveEntry entry = new TarArchiveEntry(p.getName());
    entry.setSize(bytes.length);
    out.putArchiveEntry(entry);
    out.write(bytes);
    out.closeArchiveEntry();
    out.close();

    LocalResource ret = recordFactory.newRecordInstance(LocalResource.class);
    ret.setResource(ConverterUtils.getYarnUrlFromPath(new Path(p.toString()
        + ".tar.gz")));
    ret.setSize(len);
    ret.setType(LocalResourceType.ARCHIVE);
    ret.setVisibility(vis);
    ret.setTimestamp(files.getFileStatus(new Path(p.toString() + ".tar.gz"))
        .getModificationTime());
    return ret;
  }

  static LocalResource createJarFile(FileContext files, Path p, int len,
      Random r, LocalResourceVisibility vis) throws IOException,
      URISyntaxException {
    byte[] bytes = new byte[len];
    r.nextBytes(bytes);

    File archiveFile = new File(p.toUri().getPath() + ".jar");
    archiveFile.createNewFile();
    JarOutputStream out = new JarOutputStream(
        new FileOutputStream(archiveFile));
    out.putNextEntry(new JarEntry(p.getName()));
    out.write(bytes);
    out.closeEntry();
    out.close();

    LocalResource ret = recordFactory.newRecordInstance(LocalResource.class);
    ret.setResource(ConverterUtils.getYarnUrlFromPath(new Path(p.toString()
        + ".jar")));
    ret.setSize(len);
    ret.setType(LocalResourceType.ARCHIVE);
    ret.setVisibility(vis);
    ret.setTimestamp(files.getFileStatus(new Path(p.toString() + ".jar"))
        .getModificationTime());
    return ret;
  }

  static LocalResource createZipFile(FileContext files, Path p, int len,
      Random r, LocalResourceVisibility vis) throws IOException,
      URISyntaxException {
    byte[] bytes = new byte[len];
    r.nextBytes(bytes);

    File archiveFile = new File(p.toUri().getPath() + ".ZIP");
    archiveFile.createNewFile();
    ZipOutputStream out = new ZipOutputStream(
        new FileOutputStream(archiveFile));
    out.putNextEntry(new ZipEntry(p.getName()));
    out.write(bytes);
    out.closeEntry();
    out.close();

    LocalResource ret = recordFactory.newRecordInstance(LocalResource.class);
    ret.setResource(ConverterUtils.getYarnUrlFromPath(new Path(p.toString()
        + ".ZIP")));
    ret.setSize(len);
    ret.setType(LocalResourceType.ARCHIVE);
    ret.setVisibility(vis);
    ret.setTimestamp(files.getFileStatus(new Path(p.toString() + ".ZIP"))
        .getModificationTime());
    return ret;
  }

  @Test (timeout=10000)
  public void testDownloadBadPublic() throws IOException, URISyntaxException,
      InterruptedException {
    Configuration conf = new Configuration();
    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 = Executors.newSingleThreadExecutor();
    LocalDirAllocator dirs =
      new LocalDirAllocator(TestFSDownload.class.getName());
    int size = 512;
    LocalResourceVisibility vis = LocalResourceVisibility.PUBLIC;
    Path path = new Path(basedir, "test-file");
    LocalResource rsrc = createFile(files, path, size, rand, vis);
    rsrcVis.put(rsrc, vis);
    Path destPath = dirs.getLocalPathForWrite(
        basedir.toString(), size, 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));
    Assert.assertTrue(pending.get(rsrc).isDone());

    try {
      for (Map.Entry<LocalResource,Future<Path>> p : pending.entrySet()) {
        p.getValue().get();
        Assert.fail("We localized a file that is not public.");
      }
    } catch (ExecutionException e) {
      Assert.assertTrue(e.getCause() instanceof IOException);
    }
  }

  @Test (timeout=60000)
  public void testDownloadPublicWithStatCache() throws IOException,
      URISyntaxException, InterruptedException, ExecutionException {
    final Configuration conf = new Configuration();
    FileContext files = FileContext.getLocalFSFileContext(conf);
    Path basedir = files.makeQualified(new Path("target",
      TestFSDownload.class.getSimpleName()));

    // if test directory doesn't have ancestor permission, skip this test
    FileSystem f = basedir.getFileSystem(conf);
    assumeTrue(FSDownload.ancestorsHaveExecutePermissions(f, basedir, null));

    files.mkdir(basedir, null, true);
    conf.setStrings(TestFSDownload.class.getName(), basedir.toString());

    int size = 512;

    final ConcurrentMap<Path,AtomicInteger> counts =
        new ConcurrentHashMap<Path,AtomicInteger>();
    final CacheLoader<Path,Future<FileStatus>> loader =
        FSDownload.createStatusCacheLoader(conf);
    final LoadingCache<Path,Future<FileStatus>> statCache =
        CacheBuilder.newBuilder().build(new CacheLoader<Path,Future<FileStatus>>() {
      public Future<FileStatus> load(Path path) throws Exception {
        // increment the count
        AtomicInteger count = counts.get(path);
        if (count == null) {
          count = new AtomicInteger(0);
          AtomicInteger existing = counts.putIfAbsent(path, count);
          if (existing != null) {
            count = existing;
          }
        }
        count.incrementAndGet();

        // use the default loader
        return loader.load(path);
      }
    });

    // test FSDownload.isPublic() concurrently
    final int fileCount = 3;
    List<Callable<Boolean>> tasks = new ArrayList<Callable<Boolean>>();
    for (int i = 0; i < fileCount; i++) {
      Random rand = new Random();
      long sharedSeed = rand.nextLong();
      rand.setSeed(sharedSeed);
      System.out.println("SEED: " + sharedSeed);
      final Path path = new Path(basedir, "test-file-" + i);
      createFile(files, path, size, rand);
      final FileSystem fs = path.getFileSystem(conf);
      final FileStatus sStat = fs.getFileStatus(path);
      tasks.add(new Callable<Boolean>() {
        public Boolean call() throws IOException {
          return FSDownload.isPublic(fs, path, sStat, statCache);
        }
      });
    }

    ExecutorService exec = Executors.newFixedThreadPool(fileCount);
    try {
      List<Future<Boolean>> futures = exec.invokeAll(tasks);
      // files should be public
      for (Future<Boolean> future: futures) {
        assertTrue(future.get());
      }
      // for each path exactly one file status call should be made
      for (AtomicInteger count: counts.values()) {
        assertSame(count.get(), 1);
      }
    } finally {
      exec.shutdown();
    }
  }

  @Test (timeout=10000)
  public void testDownload() throws IOException, URISyntaxException,
      InterruptedException {
    Configuration conf = new Configuration();
    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 = Executors.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.valueOf(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);
    }
  }

  private void downloadWithFileType(TEST_FILE_TYPE fileType) throws IOException, 
      URISyntaxException, InterruptedException{
    Configuration conf = new Configuration();
    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());

    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 = Executors.newSingleThreadExecutor();
    LocalDirAllocator dirs = new LocalDirAllocator(
        TestFSDownload.class.getName());

    int size = rand.nextInt(512) + 512;
    LocalResourceVisibility vis = LocalResourceVisibility.PRIVATE;
    Path p = new Path(basedir, "" + 1);
    String strFileName = "";
    LocalResource rsrc = null;
    switch (fileType) {
    case TAR:
      rsrc = createTarFile(files, p, size, rand, vis);
      break;
    case JAR:
      rsrc = createJarFile(files, p, size, rand, vis);
      rsrc.setType(LocalResourceType.PATTERN);
      break;
    case ZIP:
      rsrc = createZipFile(files, p, size, rand, vis);
      strFileName = p.getName() + ".ZIP";
      break;
    case TGZ:
      rsrc = createTgzFile(files, p, size, rand, vis);
      break;
    }
    Path destPath = dirs.getLocalPathForWrite(basedir.toString(), size, 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));
    try {
      pending.get(rsrc).get(); // see if there was an Exception during download
      FileStatus[] filesstatus = files.getDefaultFileSystem().listStatus(
          basedir);
      for (FileStatus filestatus : filesstatus) {
        if (filestatus.isDirectory()) {
          FileStatus[] childFiles = files.getDefaultFileSystem().listStatus(
              filestatus.getPath());
          for (FileStatus childfile : childFiles) {
            if(strFileName.endsWith(".ZIP") &&
               childfile.getPath().getName().equals(strFileName) &&
               !childfile.isDirectory()) {
               Assert.fail("Failure...After unzip, there should have been a" +
                 " directory formed with zip file name but found a file. "
                 + childfile.getPath());
            }
            if (childfile.getPath().getName().startsWith("tmp")) {
              Assert.fail("Tmp File should not have been there "
                  + childfile.getPath());
            }
          }
        }
      }
    }catch (Exception e) {
      throw new IOException("Failed exec", e);
    }
  }

  @Test (timeout=10000) 
  public void testDownloadArchive() throws IOException, URISyntaxException,
      InterruptedException {
    downloadWithFileType(TEST_FILE_TYPE.TAR);
  }

  @Test (timeout=10000)
  public void testDownloadPatternJar() throws IOException, URISyntaxException,
      InterruptedException {
    downloadWithFileType(TEST_FILE_TYPE.JAR);
  }

  @Test (timeout=10000) 
  public void testDownloadArchiveZip() throws IOException, URISyntaxException,
      InterruptedException {
    downloadWithFileType(TEST_FILE_TYPE.ZIP);
  }

  /*
   * To test fix for YARN-3029
   */
  @Test (timeout=10000)
  public void testDownloadArchiveZipWithTurkishLocale() throws IOException,
      URISyntaxException, InterruptedException {
    Locale defaultLocale = Locale.getDefault();
    // Set to Turkish
    Locale turkishLocale = new Locale("tr", "TR");
    Locale.setDefault(turkishLocale);
    downloadWithFileType(TEST_FILE_TYPE.ZIP);
    // Set the locale back to original default locale
    Locale.setDefault(defaultLocale);
  }

  @Test (timeout=10000)
  public void testDownloadArchiveTgz() throws IOException, URISyntaxException,
      InterruptedException {
    downloadWithFileType(TEST_FILE_TYPE.TGZ);
  }

  private void verifyPermsRecursively(FileSystem fs,
      FileContext files, Path p,
      LocalResourceVisibility vis) throws IOException {
    FileStatus status = files.getFileStatus(p);
    if (status.isDirectory()) {
      if (vis == LocalResourceVisibility.PUBLIC) {
        Assert.assertTrue(status.getPermission().toShort() ==
          FSDownload.PUBLIC_DIR_PERMS.toShort());
      }
      else {
        Assert.assertTrue(status.getPermission().toShort() ==
          FSDownload.PRIVATE_DIR_PERMS.toShort());
      }
      if (!status.isSymlink()) {
        FileStatus[] statuses = fs.listStatus(p);
        for (FileStatus stat : statuses) {
          verifyPermsRecursively(fs, files, stat.getPath(), vis);
        }
      }
    }
    else {
      if (vis == LocalResourceVisibility.PUBLIC) {
        Assert.assertTrue(status.getPermission().toShort() ==
          FSDownload.PUBLIC_FILE_PERMS.toShort());
      }
      else {
        Assert.assertTrue(status.getPermission().toShort() ==
          FSDownload.PRIVATE_FILE_PERMS.toShort());
      }
    }      
  }

  @Test (timeout=10000)
  public void testDirDownload() throws IOException, InterruptedException {
    Configuration conf = new Configuration();
    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 = Executors.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);
    }
  }

  @Test (timeout=10000)
  public void testUniqueDestinationPath() throws Exception {
    Configuration conf = new Configuration();
    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());

    ExecutorService singleThreadedExec = Executors.newSingleThreadExecutor();

    LocalDirAllocator dirs =
        new LocalDirAllocator(TestFSDownload.class.getName());
    Path destPath = dirs.getLocalPathForWrite(basedir.toString(), conf);
    destPath =
        new Path(destPath, Long.toString(uniqueNumberGenerator
            .incrementAndGet()));

    Path p = new Path(basedir, "dir" + 0 + ".jar");
    LocalResourceVisibility vis = LocalResourceVisibility.PRIVATE;
    LocalResource rsrc = createJar(files, p, vis);
    FSDownload fsd =
        new FSDownload(files, UserGroupInformation.getCurrentUser(), conf,
            destPath, rsrc);
    Future<Path> rPath = singleThreadedExec.submit(fsd);
    singleThreadedExec.shutdown();
    while (!singleThreadedExec.awaitTermination(1000, TimeUnit.MILLISECONDS));
    Assert.assertTrue(rPath.isDone());
    // Now FSDownload will not create a random directory to localize the
    // resource. Therefore the final localizedPath for the resource should be
    // destination directory (passed as an argument) + file name.
    Assert.assertEquals(destPath, rPath.get().getParent());
  }
}
