/*
 * The Alluxio Open Foundation licenses this work under the Apache License, version 2.0
 * (the "License"). You may not use this work except in compliance with the License, which is
 * available at www.apache.org/licenses/LICENSE-2.0
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND,
 * either express or implied, as more fully set forth in the License.
 *
 * See the NOTICE file distributed with this work for information regarding copyright ownership.
 */

package alluxio.client.file;

import alluxio.AlluxioURI;
import alluxio.collections.Pair;
import alluxio.exception.AlluxioException;
import alluxio.security.authorization.Permission;
import alluxio.underfs.UnderFileSystem;
import alluxio.underfs.options.MkdirsOptions;

import java.io.IOException;
import java.util.Stack;

import javax.annotation.concurrent.ThreadSafe;

/**
 * Utility functions for working with {@link alluxio.underfs.UnderFileSystem}.
 *
 */
@ThreadSafe
public final class UnderFileSystemUtils {

  /**
   * Creates parent directories for path with correct permissions if required.
   *
   * @param alluxioPath Alluxio path
   * @param ufsPath path in the under file system
   * @param fs file system master client
   * @param ufs the under file system
   * @throws AlluxioException if an unexpected Alluxio exception is thrown
   * @throws IOException if folder creation fails
   */
  public static void prepareFilePath(AlluxioURI alluxioPath, String ufsPath,
      FileSystem fs, UnderFileSystem ufs) throws AlluxioException, IOException {
    AlluxioURI dstPath = new AlluxioURI(ufsPath);
    String parentPath = dstPath.getParent().getPath();
    // creates the parent folder if it does not exist
    if (!ufs.isDirectory(parentPath)) {
      // Create ancestor directories from top to the bottom. We cannot use recursive create parents
      // here because the permission for the ancestors can be different.
      Stack<Pair<String, MkdirsOptions>> ufsDirsToMakeWithOptions = new Stack<>();
      AlluxioURI curAlluxioPath = alluxioPath.getParent();
      AlluxioURI curUfsPath = dstPath.getParent();
      // Stop at the Alluxio root because the mapped directory of Alluxio root in UFS may not exist.
      while (curUfsPath != null && !ufs.isDirectory(curUfsPath.toString())
          && curAlluxioPath != null) {
        URIStatus curDirStatus;
        curDirStatus = fs.getStatus(curAlluxioPath);
        Permission perm = new Permission(curDirStatus.getOwner(), curDirStatus.getGroup(),
            (short) curDirStatus.getMode());
        ufsDirsToMakeWithOptions.push(new Pair<>(curUfsPath.toString(),
            MkdirsOptions.defaults().setCreateParent(false).setPermission(perm)));
        curAlluxioPath = curAlluxioPath.getParent();
        curUfsPath = curUfsPath.getParent();
      }
      while (!ufsDirsToMakeWithOptions.empty()) {
        Pair<String, MkdirsOptions> ufsDirAndPerm = ufsDirsToMakeWithOptions.pop();
        // UFS mkdirs might fail if the directory is already created. If so, skip the mkdirs
        // and assume the directory is already prepared, regardless of permission matching.
        if (!ufs.mkdirs(ufsDirAndPerm.getFirst(), ufsDirAndPerm.getSecond())
            && !ufs.isDirectory(ufsDirAndPerm.getFirst())) {
          throw new IOException("Failed to create dir: " + ufsDirAndPerm.getFirst());
        }
      }
    }
  }

  private UnderFileSystemUtils() {} // prevent instantiation
}

@@ -636,24 +636,67 @@
 oException;%0A
+import alluxio.exception.ExceptionMessage;%0A
 import allux
@@ -2197,20 +2197,16 @@
  at 
-the 
 Alluxio 
 root
@@ -2201,19 +2201,26 @@
 Alluxio 
-roo
+mount poin
 t becaus
@@ -2246,24 +2246,8 @@
 ory 
-of Alluxio root 
 in U
@@ -2408,16 +2408,47 @@
 irStatus
+ = fs.getStatus(curAlluxioPath)
 ;%0A      
@@ -2449,16 +2449,20 @@
         
+if (
 curDirSt
@@ -2469,40 +2469,134 @@
 atus
- = fs.getStatus(curAlluxioPath);
+.isMountPoint()) %7B%0A          throw new IOException(ExceptionMessage.UFS_PATH_DOES_NOT_EXIST.getMessage(curUfsPath));%0A        %7D
 %0A   

Stop parent creation at mount point