Callable<Path> download(Path destDirPath, LocalResource rsrc,
      UserGroupInformation ugi) throws IOException {
    // For private localization FsDownload creates folder in destDirPath. Parent
    // directories till user filecache folder is created here.
    if (rsrc.getVisibility() == LocalResourceVisibility.PRIVATE) {
      createParentDirs(destDirPath);
    }
    diskValidator
        .checkStatus(new File(destDirPath.getParent().toUri().getRawPath()));
    return new FSDownloadWrapper(lfs, ugi, conf, destDirPath, rsrc);
  }
@Test(timeout = 10000)
  public void testUserCacheDirPermission() throws Exception {
    Configuration conf = new Configuration();
    conf.set(CommonConfigurationKeys.FS_PERMISSIONS_UMASK_KEY, "077");
    FileContext lfs = FileContext.getLocalFSFileContext(conf);
    Path fileCacheDir = lfs.makeQualified(new Path(basedir, "filecache"));
    lfs.mkdir(fileCacheDir, FsPermission.getDefault(), true);
    RecordFactory recordFactory = mock(RecordFactory.class);
    ContainerLocalizer localizer = new ContainerLocalizer(lfs,
        UserGroupInformation.getCurrentUser().getUserName(), "application_01",
        "container_01", String.format(ContainerExecutor.TOKEN_FILE_NAME_FMT,
        "container_01"), new ArrayList<>(), recordFactory);
    LocalResource rsrc = mock(LocalResource.class);
    when(rsrc.getVisibility()).thenReturn(LocalResourceVisibility.PRIVATE);
    Path destDirPath = new Path(fileCacheDir, "0/0/85");
    //create one of the parent directories with the wrong permissions first
    FsPermission wrongPerm = new FsPermission((short) 0700);
    lfs.mkdir(destDirPath.getParent().getParent(), wrongPerm, false);
    lfs.mkdir(destDirPath.getParent(), wrongPerm, false);
    //Localize and check the directory permission are correct.
    localizer
        .download(destDirPath, rsrc, UserGroupInformation.getCurrentUser());
    Assert
        .assertEquals("Cache directory permissions filecache/0/0 is incorrect",
            USERCACHE_DIR_PERM,
            lfs.getFileStatus(destDirPath.getParent()).getPermission());
    Assert.assertEquals("Cache directory permissions filecache/0 is incorrect",
        USERCACHE_DIR_PERM,
        lfs.getFileStatus(destDirPath.getParent().getParent()).getPermission());
  }