HDFS-16791 More cleanup, fixes and tests

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileSystem.java

@@ -4879,15 +4879,19 @@ public CompletableFuture<FSDataInputStream> build() throws IOException {
   /**
    * Return path of the enclosing root for a given path
    * The enclosing root path is a common ancestor that should be used for temp and staging dirs
-   * as well as within encryption zones and other restricted directories
+   * as well as within encryption zones and other restricted directories.
+   * 
+   * Call makeQualified on the param path to ensure the param path to ensure its part of the correct filesystem
+   * 
    * @param path file path to find the enclosing root path for
-   * @return a path to the enclosing rot
+   * @return a path to the enclosing root
    * @throws IOException
    */
   @InterfaceAudience.Public
   @InterfaceStability.Unstable
   // Should this throw RuntimeException (instead of IO), so we can throw NotInMountpointException from viewfs/rbf?
   public Path getEnclosingRoot(Path path) throws IOException {
+    this.makeQualified(path);
     return this.makeQualified(new Path("/"));
   }
 

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java

@@ -1372,10 +1372,15 @@ public boolean hasPathCapability(Path path, String capability)
 
   @Override
   public Path getEnclosingRoot(Path path) throws IOException {
-    InodeTree.ResolveResult<FileSystem> res = fsState.resolve(getUriPath(path), true);
-    Path fullPath = new Path(res.resolvedPath);
-    Path enclosingPath = res.targetFileSystem.getEnclosingRoot(path);
-    return fixRelativePart(enclosingPath.depth() > fullPath.depth() ?  enclosingPath : fullPath);
+    InodeTree.ResolveResult<FileSystem> res;
+    try {
+      res = fsState.resolve(getUriPath(path), true);
+    } catch (FileNotFoundException ex) {
+      throw new NotInMountpointException(path, "getEnclosingRoot");
+    }
+    Path mountPath = new Path(res.resolvedPath);
+    Path enclosingPath = res.targetFileSystem.getEnclosingRoot(new Path(getUriPath(path)));
+    return fixRelativePart(this.makeQualified(enclosingPath.depth() > mountPath.depth() ?  enclosingPath : mountPath));
   }
 
   /**
@@ -1930,10 +1935,15 @@ public Collection<? extends BlockStoragePolicySpi> getAllStoragePolicies()
 
     @Override
     public Path getEnclosingRoot(Path path) throws IOException {
-      InodeTree.ResolveResult<FileSystem> res = fsState.resolve(path.toString(), true);
+      InodeTree.ResolveResult<FileSystem> res;
+      try {
+        res = fsState.resolve((path.toString()), true);
+      } catch (FileNotFoundException ex) {
+        throw new NotInMountpointException(path, "getEnclosingRoot");
+      }
       Path fullPath = new Path(res.resolvedPath);
       Path enclosingPath = res.targetFileSystem.getEnclosingRoot(path);
-      return (enclosingPath.depth() > fullPath.depth() ?  enclosingPath : fullPath).makeQualified(myUri, null);
+      return enclosingPath.depth() > fullPath.depth() ?  enclosingPath : fullPath;
     }
   }
 

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/router/RouterClientProtocol.java

@@ -1960,10 +1960,10 @@ public Path getEnclosingRoot(String src) throws IOException {
     }
 
     EncryptionZone zone = getEZForPath(src);
-    Path zonePath = new Path((zone != null ? zone.getPath() : null));
-    if (zonePath == null) {
+    if (zone == null) {
       return mountPath;
     } else {
+      Path zonePath = new Path(zone.getPath());
       return zonePath.depth() > mountPath.depth() ? zonePath : mountPath;
     }
   }

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/router/TestRouterMountTable.java

@@ -95,6 +95,7 @@ public static void globalSetUp() throws Exception {
     conf.setInt(RBFConfigKeys.DFS_ROUTER_ADMIN_MAX_COMPONENT_LENGTH_KEY, 20);
     cluster.addRouterOverrides(conf);
     cluster.startCluster();
+    cluster.startCluster();
     cluster.startRouters();
     cluster.waitClusterUp();
 

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/fs/viewfs/TestViewFileSystemHdfs.java

@@ -506,6 +506,14 @@ public void testInternalDirectoryPermissions() throws IOException {
     assertEquals(fs.getFileStatus(subDirOfInternalDir).getPermission(),
         fs.getFileStatus(subDirOfRealDir).getPermission());
   }
+
+  private Path getViewFsPath(Path path, FileSystem fs) {
+    return fs.makeQualified(path);
+  }
+
+  private Path getViewFsPath(String path, FileSystem fs) {
+    return getViewFsPath(new Path(path), fs);
+  }
 
   @Test
   public void testEnclosingRootsBase() throws Exception {
@@ -519,16 +527,46 @@ public void testEnclosingRootsBase() throws Exception {
     final EnumSet<CreateEncryptionZoneFlag> provisionTrash =
         EnumSet.of(CreateEncryptionZoneFlag.PROVISION_TRASH);
     hdfsAdmin.createEncryptionZone(zone1, "test_key", provisionTrash);
-    assertEquals(fsView.getEnclosingRoot(zone), new Path("/data"));
-    assertEquals(fsView.getEnclosingRoot(zone1), zone1);
+    assertEquals(fsView.getEnclosingRoot(zone), getViewFsPath("/data", fsView));
+    assertEquals(fsView.getEnclosingRoot(zone1), getViewFsPath(zone1, fsView));
 
     Path nn02Ez = new Path("/mountOnNn2/EZ");
     fsTarget2.mkdirs(nn02Ez);
-    assertEquals(fsView.getEnclosingRoot((nn02Ez)), new Path("/mountOnNn2"));
+    assertEquals(fsView.getEnclosingRoot((nn02Ez)), getViewFsPath("/mountOnNn2", fsView));
     HdfsAdmin hdfsAdmin2 = new HdfsAdmin(cluster.getURI(1), CONF);
     DFSTestUtil.createKey("test_key", cluster, 1, CONF);
     hdfsAdmin2.createEncryptionZone(nn02Ez, "test_key", provisionTrash);
-    assertEquals(fsView.getEnclosingRoot((nn02Ez)), nn02Ez);
-    assertEquals(fsView.getEnclosingRoot(new Path(nn02Ez, "dir/dir2/file")), nn02Ez);
+    assertEquals(fsView.getEnclosingRoot((nn02Ez)), getViewFsPath(nn02Ez, fsView));
+    assertEquals(fsView.getEnclosingRoot(new Path(nn02Ez, "dir/dir2/file")), getViewFsPath(nn02Ez, fsView));
+
+    // With viewfs:// scheme
+    assertEquals(fsView.getEnclosingRoot(fsView.getWorkingDirectory()), getViewFsPath("/user", fsView));
+  }
+
+  @Test
+  public void testEnclosingRootFailure() throws IOException {
+    try {
+      fsView.getEnclosingRoot(new Path("/does/not/exist"));
+      fail("Not a mount point");
+    } catch (NotInMountpointException ex) {
+      // expected
+    }
+
+    final Path zone = new Path("/data/EZ");
+    Path fs1 = fsTarget.makeQualified(zone);
+    try {
+      fsTarget2.getEnclosingRoot(fs1);
+      fail("Wrong filesystem");
+    } catch (IllegalArgumentException ex) {
+      // expected
+    }
+
+    try {
+      fsTarget2.getEnclosingRoot(new Path("/"));
+      fail("Wrong filesystem");
+    } catch (IllegalArgumentException ex) {
+      // expected
+    }
+
   }
 }