HDFS-16791 fix pr comments

Author: Tom McCormick

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

@@ -4877,11 +4877,16 @@ public CompletableFuture<FSDataInputStream> build() throws IOException {
   }
 
   /**
-   * Return root path
-   * @param path
-   * @return
+   * 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
+   * @param path file path to find the enclosing root path for
+   * @return a path to the enclosing rot
    * @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 {
     return this.makeQualified(new Path("/"));
   }

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

@@ -1369,13 +1369,13 @@ public boolean hasPathCapability(Path path, String capability)
       throw new NotInMountpointException(p, "hasPathCapability");
     }
   }
-
+  
   @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 enclosingPath.depth() > fullPath.depth() ?  enclosingPath : fullPath;
+    return fixRelativePart(enclosingPath.depth() > fullPath.depth() ?  enclosingPath : fullPath);
   }
 
   /**
@@ -1933,7 +1933,7 @@ public Path getEnclosingRoot(Path path) throws IOException {
       InodeTree.ResolveResult<FileSystem> res = fsState.resolve(path.toString(), true);
       Path fullPath = new Path(res.resolvedPath);
       Path enclosingPath = res.targetFileSystem.getEnclosingRoot(path);
-      return enclosingPath.depth() > fullPath.depth() ?  enclosingPath : fullPath;
+      return (enclosingPath.depth() > fullPath.depth() ?  enclosingPath : fullPath).makeQualified(myUri, null);
     }
   }
 

hadoop-common-project/hadoop-common/src/site/markdown/filesystem/filesystem.md

@@ -599,6 +599,25 @@ on the filesystem.
     if len(FS, P) > 0:  getFileStatus(P).getBlockSize() > 0
     result == getFileStatus(P).getBlockSize()
 
+1. The outcome of this operation MUST be identical to the value of
+   `getFileStatus(P).getBlockSize()`.
+2. By inference, it MUST be > 0 for any file of length > 0.
+###  `Path getEnclosingRoot(Path p)`
+
+This method is used to find a root directory for a path given. This is useful for creating 
+staging and temp directories in the same root directory. There are constraints around how 
+renames are allowed to atomically occur (ex. across hdfs volumes or across encryption zones).
+
+#### Preconditions
+
+if no root path is found:  raise IOException
+No root path would be found only if there is no mount point for the given path
+
+
+#### Postconditions
+
+The path return will not be null
+
 1. The outcome of this operation MUST be identical to the value of
    `getFileStatus(P).getBlockSize()`.
 1. By inference, it MUST be > 0 for any file of length > 0.

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/TestHarFileSystem.java

@@ -250,6 +250,8 @@ CompletableFuture<FSDataInputStream> openFileWithOptions(
 
     MultipartUploaderBuilder createMultipartUploader(Path basePath)
         throws IOException;
+
+    Path getEnclosingRoot(Path path) throws IOException;
   }
 
   @Test

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DFSClient.java

@@ -3531,7 +3531,7 @@ public DatanodeInfo[] slowDatanodeReport() throws IOException {
   public Path getEnclosingRoot(String src) throws IOException {
     checkOpen();
     try (TraceScope ignored = newPathTraceScope("getEnclosingRoot", src)) {
-      return new Path(namenode.getEnclosingRoot(src));
+      return namenode.getEnclosingRoot(src);
     } catch (RemoteException re) {
       throw re.unwrapRemoteException(AccessControlException.class,
           UnresolvedPathException.class);

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/DistributedFileSystem.java

@@ -3913,15 +3913,14 @@ public Path doCall(final Path p) throws IOException {
       @Override
       public Path next(final FileSystem fs, final Path p)
           throws IOException {
-        if (fs instanceof DistributedFileSystem) {
-          DistributedFileSystem myDfs = (DistributedFileSystem) fs;
-          return myDfs.getEnclosingRoot(p);
-        } else {
+        if (!(fs instanceof DistributedFileSystem)) {
           throw new UnsupportedOperationException(
-              "Cannot call getEZForPath"
+              "Cannot call getEnclosingRoot"
                   + " on a symlink to a non-DistributedFileSystem: " + path
                   + " -> " + p);
         }
+        DistributedFileSystem myDfs = (DistributedFileSystem) fs;
+        return myDfs.getEnclosingRoot(p);
       }
     }.resolve(this, absF);
   }

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java

@@ -1894,6 +1894,6 @@ BatchedEntries<OpenFileEntry> listOpenFiles(long prevId,
    */
   @Idempotent
   @ReadOnly(isCoordinated = true)
-  String getEnclosingRoot(String src) throws IOException;
+  Path getEnclosingRoot(String src) throws IOException;
 
 }

hadoop-hdfs-project/hadoop-hdfs-client/src/main/java/org/apache/hadoop/hdfs/protocolPB/ClientNamenodeProtocolTranslatorPB.java

@@ -2110,15 +2110,15 @@ public HAServiceProtocol.HAServiceState getHAServiceState()
   }
 
   @Override
-  public String getEnclosingRoot(String filename) throws IOException {
+  public Path getEnclosingRoot(String filename) throws IOException {
     final GetEnclosingRootRequestProto.Builder builder =
         GetEnclosingRootRequestProto.newBuilder();
     builder.setFilename(filename);
     final GetEnclosingRootRequestProto req = builder.build();
     try {
       final GetEnclosingRootResponseProto response =
           rpcProxy.getEnclosingRoot(null, req);
-      return response.getEnclosingRootPath();
+      return new Path(response.getEnclosingRootPath());
     } catch (ServiceException e) {
       throw ProtobufHelper.getRemoteException(e);
     }

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

@@ -144,6 +144,9 @@ public class RouterClientProtocol implements ClientProtocol {
   private final boolean allowPartialList;
   /** Time out when getting the mount statistics. */
   private long mountStatusTimeOut;
+  
+  /** Default nameservice enabled */
+  private final boolean defaultNameserviceEnabled;
 
   /** Identifier for the super user. */
   private String superUser;
@@ -194,6 +197,8 @@ public class RouterClientProtocol implements ClientProtocol {
     this.routerCacheAdmin = new RouterCacheAdmin(rpcServer);
     this.securityManager = rpcServer.getRouterSecurityManager();
     this.rbfRename = new RouterFederationRename(rpcServer, conf);
+    this.defaultNameserviceEnabled = conf.getBoolean(RBFConfigKeys.DFS_ROUTER_DEFAULT_NAMESERVICE_ENABLE, 
+        RBFConfigKeys.DFS_ROUTER_DEFAULT_NAMESERVICE_ENABLE_DEFAULT);
   }
 
   @Override
@@ -1936,18 +1941,31 @@ public DatanodeInfo[] getSlowDatanodeReport() throws IOException {
   }
 
   @Override
-  public String getEnclosingRoot(String src) throws IOException {
-    Path mountPath = new Path("/");
+  public Path getEnclosingRoot(String src) throws IOException {
+    Path mountPath = null;
+    if (defaultNameserviceEnabled) {
+      mountPath = new Path("/");
+    }
+    
     if (subclusterResolver instanceof MountTableResolver) {
       MountTableResolver mountTable = (MountTableResolver) subclusterResolver;
       if (mountTable.getMountPoint(src) != null) {
         // unclear if this is the correct thing to do, probably depends on default mount point / link fallback
         mountPath = new Path(mountTable.getMountPoint(src).getSourcePath());
       }
     }
+    
+    if (mountPath == null) {
+      throw new IOException(String.format("No mount point for %s", src));
+    }
+    
     EncryptionZone zone = getEZForPath(src);
-    Path zonePath = new Path((zone != null ? zone.getPath() : "/"));
-    return (zonePath.depth() > mountPath.depth() ? zonePath : mountPath).toString();
+    Path zonePath = new Path((zone != null ? zone.getPath() : null));
+    if (zonePath == null) {
+      return mountPath;
+    } else {
+      return zonePath.depth() > mountPath.depth() ? zonePath : mountPath;
+    }
   }
 
   @Override

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

@@ -2070,7 +2070,7 @@ public ListenableFuture<DatanodeInfo[]> reload(
   }
 
   @Override // ClientProtocol
-  public String getEnclosingRoot(String src) throws IOException {
+  public Path getEnclosingRoot(String src) throws IOException {
     // need to resolve this src to a mount point in RBF and return the max this and the enclosing root from nn
     return clientProto.getEnclosingRoot(src);
   }