Move path resolution logic and fix a bug

dora/core/client/fs/src/main/java/alluxio/client/file/DoraCacheFileSystem.java

@@ -61,10 +61,10 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.function.Consumer;
-import java.util.stream.Collectors;
 
 /**
  * Dora Cache file system implementation.
@@ -253,12 +253,12 @@ public List<URIStatus> listStatus(AlluxioURI path, ListStatusPOptions options)
 
       final List<URIStatus> uriStatuses = mDoraClient.listStatus(ufsFullPath.toString(),
           mergedOptions);
-      List<URIStatus> statusesWithRelativePath = uriStatuses.stream()
-          .map(status -> new URIStatus(
-              GrpcUtils.fromProto(GrpcUtils.toProto(status.getFileInfo()))
-                  .setPath(convertToAlluxioPath(new AlluxioURI(status.getUfsPath()))
-                      .getPath())))
-          .collect(Collectors.toList());
+      List<URIStatus> statusesWithRelativePath = new ArrayList<>(uriStatuses.size());
+      for (URIStatus s : uriStatuses) {
+        statusesWithRelativePath.add(
+            new URIStatus(GrpcUtils.fromProto(GrpcUtils.toProto(s.getFileInfo())).setPath(
+                convertToAlluxioPath(new AlluxioURI(s.getUfsPath())).getPath())));
+      }
       return statusesWithRelativePath;
     } catch (RuntimeException ex) {
       if (ex instanceof StatusRuntimeException) {
@@ -443,7 +443,7 @@ public AlluxioURI convertToUfsPath(AlluxioURI alluxioPath) {
         "FileSystem is not UfsBaseFileSystem");
     UfsBaseFileSystem under = (UfsBaseFileSystem) mDelegatedFileSystem;
     AlluxioURI rootUFS = under.getRootUFS();
-    return convertAlluxioPathToUfsPath(alluxioPath, rootUFS);
+    return PathUtils.convertAlluxioPathToUfsPath(alluxioPath, rootUFS);
   }
 
   /**
@@ -453,12 +453,13 @@ public AlluxioURI convertToUfsPath(AlluxioURI alluxioPath) {
    *
    * @param ufsPath UfsBaseFileSystem based full path
    * @return an Alluxio path
+   * @throws InvalidPathException if ufsPath is not a child of the UFS mounted at Alluxio root
    */
-  public AlluxioURI convertToAlluxioPath(AlluxioURI ufsPath) {
+  public AlluxioURI convertToAlluxioPath(AlluxioURI ufsPath) throws InvalidPathException {
     Preconditions.checkArgument(mDelegatedFileSystem instanceof UfsBaseFileSystem,
         "FileSystem is not UfsBaseFileSystem");
     AlluxioURI rootUfs = ((UfsBaseFileSystem) mDelegatedFileSystem).getRootUFS();
-    return convertUfsPathToAlluxioPath(ufsPath, rootUfs);
+    return PathUtils.convertUfsPathToAlluxioPath(ufsPath, rootUfs);
   }
 
   @Override
@@ -502,63 +503,4 @@ public List<BlockLocationInfo> getBlockLocations(URIStatus status)
     }
     return listBuilder.build();
   }
-
-  /**
-   * Converts the Alluxio based path to UfsBaseFileSystem based path if needed.
-   * This is a static utility intended for reuse.
-   * <p>
-   * UfsBaseFileSystem expects absolute/full file path. The Dora Worker
-   * expects absolute/full file path, too. So we need to convert the input path from Alluxio
-   * relative path to full UFS path if it is an Alluxio relative path.
-   * We do this by checking if the path is leading with the UFS root. If the input path
-   * is already considered to be UFS path, it should be leading a UFS path with appropriate scheme.
-   * If local file system is used, please add "file://" scheme before the path.
-   *
-   * @param alluxioPath Alluxio based path
-   * @param ufsRootPath the UFS root path to resolve against
-   * @return UfsBaseFileSystem based full path
-   */
-  public static AlluxioURI convertAlluxioPathToUfsPath(
-      AlluxioURI alluxioPath, AlluxioURI ufsRootPath) {
-    try {
-      if (ufsRootPath.isAncestorOf(alluxioPath)) {
-        // Treat this path as a full UFS path.
-        return alluxioPath;
-      }
-    } catch (InvalidPathException e) {
-      LOG.error("Invalid path {}", alluxioPath);
-      throw new RuntimeException(e);
-    }
-
-    // Treat this path as Alluxio relative, and add the UFS root before it.
-    String ufsFullPath = PathUtils.concatPath(ufsRootPath, alluxioPath.toString());
-    if (alluxioPath.isRoot()) {
-      ufsFullPath = ufsFullPath + AlluxioURI.SEPARATOR;
-    }
-
-    return new AlluxioURI(ufsFullPath);
-  }
-
-  /**
-   * Converts the UFS path back to Alluxio path.
-   * This is a static utility intended for reuse.
-   * <p>
-   * This is the opposite operation to {@link #convertAlluxioPathToUfsPath(AlluxioURI, AlluxioURI)}.
-   *
-   * @param ufsPath UfsBaseFileSystem based full path
-   * @param ufsRootPath the UFS root path to resolve against
-   * @return an Alluxio path
-   */
-  public static AlluxioURI convertUfsPathToAlluxioPath(AlluxioURI ufsPath, AlluxioURI ufsRootPath) {
-    try {
-      if (ufsRootPath.isAncestorOf(ufsPath)) {
-        return new AlluxioURI(PathUtils.concatPath(AlluxioURI.SEPARATOR,
-            PathUtils.subtractPaths(ufsPath.getPath(), ufsRootPath.getPath())));
-      }
-    } catch (InvalidPathException e) {
-      throw new RuntimeException(e);
-    }
-
-    return ufsPath;
-  }
 }

dora/core/common/src/main/java/alluxio/util/io/PathUtils.java

@@ -468,4 +468,62 @@ public static List<String> getPossibleMountPoints(String path) throws InvalidPat
     }
     return paths;
   }
+
+  /**
+   * Converts the Alluxio based path to UfsBaseFileSystem based path if needed.
+   * This is a static utility intended for reuse.
+   * <p>
+   * UfsBaseFileSystem expects absolute/full file path. The Dora Worker
+   * expects absolute/full file path, too. So we need to convert the input path from Alluxio
+   * relative path to full UFS path if it is an Alluxio relative path.
+   * We do this by checking if the path is leading with the UFS root. If the input path
+   * is already considered to be UFS path, it should be leading a UFS path with appropriate scheme.
+   * If local file system is used, please add "file://" scheme before the path.
+   *
+   * @param alluxioPath Alluxio based path
+   * @param ufsRootPath the UFS root path to resolve against
+   * @return UfsBaseFileSystem based full path
+   */
+  public static AlluxioURI convertAlluxioPathToUfsPath(
+          AlluxioURI alluxioPath, AlluxioURI ufsRootPath) {
+    try {
+      if (ufsRootPath.isAncestorOf(alluxioPath)) {
+        // Treat this path as a full UFS path.
+        return alluxioPath;
+      }
+    } catch (InvalidPathException e) {
+      throw new RuntimeException(e);
+    }
+
+    // Treat this path as Alluxio relative, and add the UFS root before it.
+    String ufsFullPath = PathUtils.concatPath(ufsRootPath, alluxioPath.toString());
+    if (alluxioPath.isRoot()) {
+      ufsFullPath = ufsFullPath + AlluxioURI.SEPARATOR;
+    }
+
+    return new AlluxioURI(ufsFullPath);
+  }
+
+  /**
+   * Converts the UFS path back to Alluxio path.
+   * This is a static utility intended for reuse.
+   * <p>
+   * This is the opposite operation to
+   * {@link #convertAlluxioPathToUfsPath(AlluxioURI, AlluxioURI)}.
+   *
+   * @param ufsPath     UfsBaseFileSystem based full path
+   * @param ufsRootPath the UFS root path to resolve against
+   * @return an Alluxio path
+   * @throws InvalidPathException if ufsPath is not a child of the UFS mounted at Alluxio root
+   */
+  public static AlluxioURI convertUfsPathToAlluxioPath(
+      AlluxioURI ufsPath, AlluxioURI ufsRootPath) throws InvalidPathException {
+    if (ufsRootPath.isAncestorOf(ufsPath)) {
+      return new AlluxioURI(PathUtils.concatPath(AlluxioURI.SEPARATOR,
+          PathUtils.subtractPaths(ufsPath.getPath(), ufsRootPath.getPath())));
+    } else {
+      throw new InvalidPathException(String.format(
+          "UFS root %s is not a prefix of %s", ufsRootPath, ufsPath));
+    }
+  }
 }