FSDownload implements Callable<Path> { private static FileStatus getFileStatus(final FileSystem fs, final Path path, LoadingCache<Path,Future<FileStatus>> statCache) throws IOException { if (statCache == null) { return fs.getFileStatus(path); } try { return statCache.get(path).get(); } catch (ExecutionException e) { Throwable cause = e.getCause(); if (cause instanceof IOException) { throw (IOException)cause; } else { throw new IOException(cause); } } catch (InterruptedException e) { Thread.currentThread().interrupt(); throw new IOException(e); } } FSDownload(FileContext files, UserGroupInformation ugi, Configuration conf,
      Path destDirPath, LocalResource resource);  FSDownload(FileContext files, UserGroupInformation ugi, Configuration conf,
      Path destDirPath, LocalResource resource,
      LoadingCache<Path,Future<FileStatus>> statCache); static CacheLoader<Path,Future<FileStatus>> createStatusCacheLoader(final Configuration conf); @Private static boolean isPublic(FileSystem fs, Path current, FileStatus sStat,
      LoadingCache<Path,Future<FileStatus>> statCache); @Override Path call();  }
@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); } }
@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); } }