boolean createNonExistentDirs(FileContext localFs,
      FsPermission perm) {
    boolean failed = false;
    List<String> localDirectories = null;
    this.readLock.lock();
    try {
      localDirectories = new ArrayList<>(localDirs);
    } finally {
      this.readLock.unlock();
    }
    for (final String dir : localDirectories) {
      try {
        createDir(localFs, new Path(dir), perm);
      } catch (IOException e) {
        LOG.warn("Unable to create directory " + dir + " error " +
            e.getMessage() + ", removing from the list of valid directories.");
        this.writeLock.lock();
        try {
          localDirs.remove(dir);
          errorDirs.add(dir);
          directoryErrorInfo.put(dir,
              new DiskErrorInformation(DiskErrorCause.OTHER,
                  "Cannot create directory : " + dir + ", error " + e.getMessage()));
          numFailures++;
        } finally {
          this.writeLock.unlock();
        }
        failed = true;
      }
    }
    return !failed;
  }
@Test
  public void testCreateDirectories() throws IOException {
    
    conf.set(CommonConfigurationKeys.FS_PERMISSIONS_UMASK_KEY, "077");

    String dirA = new File(testDir, "dirA").getPath();
    String dirB = new File(dirA, "dirB").getPath();
    String dirC = new File(testDir, "dirC").getPath();
    Path pathC = new Path(dirC);
    FsPermission permDirC = new FsPermission((short)0710);

    localFs.mkdir(pathC, null, true);
    localFs.setPermission(pathC, permDirC);

    String[] dirs = { dirA, dirB, dirC };
    DirectoryCollection dc =
        new DirectoryCollection(dirs, conf.getFloat(
          YarnConfiguration.NM_MAX_PER_DISK_UTILIZATION_PERCENTAGE,
          YarnConfiguration.DEFAULT_NM_MAX_PER_DISK_UTILIZATION_PERCENTAGE));
    FsPermission defaultPerm = FsPermission.getDefault()
        .applyUMask(new FsPermission((short)FsPermission.DEFAULT_UMASK));
    boolean createResult = dc.createNonExistentDirs(localFs, defaultPerm);
    Assert.assertTrue(createResult);

    FileStatus status = localFs.getFileStatus(new Path(dirA));
    Assert.assertEquals("local dir parent not created with proper permissions",
        defaultPerm, status.getPermission());
    status = localFs.getFileStatus(new Path(dirB));
    Assert.assertEquals("local dir not created with proper permissions",
        defaultPerm, status.getPermission());
    status = localFs.getFileStatus(pathC);
    Assert.assertEquals("existing local directory permissions modified",
        permDirC, status.getPermission());
  }