Merging change r1100026 and r1091618 from trunk

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/yahoo-merge@1102011 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -10,6 +10,8 @@ Trunk (unreleased changes)
 
     HADOOP-7171. Support UGI in FileContext API. (jitendra)
 
+    HADOOP-7257 Client side mount tables (sanjay)
+
   IMPROVEMENTS
 
     HADOOP-7133. Batch the calls in DataStorage to FileUtil.createHardLink().

src/java/org/apache/hadoop/fs/AbstractFileSystem.java

@@ -23,6 +23,7 @@ import java.io.IOException;
 import java.lang.reflect.Constructor;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.ArrayList;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.List;
@@ -432,6 +433,21 @@ public abstract class AbstractFileSystem {
    */
   public abstract FsServerDefaults getServerDefaults() throws IOException; 
 
+  /**
+   * Return the fully-qualified path of path f resolving the path
+   * through any internal symlinks or mount point
+   * @param p path to be resolved
+   * @return fully qualified path 
+   * @throws FileNotFoundException, AccessControlException, IOException
+   *         UnresolvedLinkException if symbolic link on path cannot be resolved
+   *          internally
+   */
+   public Path resolvePath(final Path p) throws FileNotFoundException,
+           UnresolvedLinkException, AccessControlException, IOException {
+     checkPath(p);
+     return getFileStatus(p).getPath(); // default impl is to return the path
+   }
+  
   /**
    * The specification of this method matches that of
    * {@link FileContext#create(Path, EnumSet, Options.CreateOpts...)} except

src/java/org/apache/hadoop/fs/DelegateToFileSystem.java

@@ -222,4 +222,4 @@ public abstract class DelegateToFileSystem extends AbstractFileSystem {
   public List<Token<?>> getDelegationTokens(String renewer) throws IOException {
     return fsImpl.getDelegationTokens(renewer);
   }
-}
+}

src/java/org/apache/hadoop/fs/FileContext.java

@@ -43,7 +43,6 @@ import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem.Statistics;
 import org.apache.hadoop.fs.Options.CreateOpts;
-import org.apache.hadoop.fs.Options.Rename;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.ipc.RpcClientException;
@@ -535,6 +534,7 @@ public final class FileContext {
   }
   
   /**
+   * 
    * @return the umask of this FileContext
    */
   public FsPermission getUMask() {

src/java/org/apache/hadoop/fs/FileStatus.java

@@ -255,18 +255,18 @@ public class FileStatus implements Writable, Comparable {
   //////////////////////////////////////////////////
   public void write(DataOutput out) throws IOException {
     Text.writeString(out, getPath().toString());
-    out.writeLong(length);
-    out.writeBoolean(isdir);
-    out.writeShort(block_replication);
-    out.writeLong(blocksize);
-    out.writeLong(modification_time);
-    out.writeLong(access_time);
-    permission.write(out);
-    Text.writeString(out, owner);
-    Text.writeString(out, group);
+    out.writeLong(getLen());
+    out.writeBoolean(isDirectory());
+    out.writeShort(getReplication());
+    out.writeLong(getBlockSize());
+    out.writeLong(getModificationTime());
+    out.writeLong(getAccessTime());
+    getPermission().write(out);
+    Text.writeString(out, getOwner());
+    Text.writeString(out, getGroup());
     out.writeBoolean(isSymlink());
     if (isSymlink()) {
-      Text.writeString(out, symlink.toString());
+      Text.writeString(out, getSymlink().toString());
     }
   }
 

src/java/org/apache/hadoop/fs/FileSystem.java

@@ -360,6 +360,7 @@ public abstract class FileSystem extends Configured implements Closeable {
   }
     
   /**
+   * Deprecated  - use @link {@link #getDelegationTokens(String)}
    * Get a new delegation token for this file system.
    * @param renewer the account name that is allowed to renew the token.
    * @return a new delegation token
@@ -535,7 +536,7 @@ public abstract class FileSystem extends Configured implements Closeable {
         getDefaultReplication(), 
         conf.getInt("io.file.buffer.size", 4096));
   }
-
+  
   /**
    * Return the fully-qualified path of path f resolving the path
    * through any symlinks or mount point

src/java/org/apache/hadoop/fs/FilterFileSystem.java

@@ -94,11 +94,12 @@ public class FilterFileSystem extends FileSystem {
     long len) throws IOException {
       return fs.getFileBlockLocations(file, start, len);
   }
-  
+
   @Override
   public Path resolvePath(final Path p) throws FileNotFoundException {
     return fs.resolvePath(p);
   }
+
   /**
    * Opens an FSDataInputStream at the indicated Path.
    * @param f the file name to open
@@ -358,6 +359,7 @@ public class FilterFileSystem extends FileSystem {
   }
 
   @Override
+  @SuppressWarnings("deprecation")
   protected boolean primitiveMkdir(Path f, FsPermission abdolutePermission)
       throws IOException {
     return fs.primitiveMkdir(f, abdolutePermission);

src/java/org/apache/hadoop/fs/FilterFs.java

@@ -138,6 +138,13 @@ public abstract class FilterFs extends AbstractFileSystem {
   public FsServerDefaults getServerDefaults() throws IOException {
     return myFs.getServerDefaults();
   }
+  
+
+  @Override
+  public Path resolvePath(final Path p) throws FileNotFoundException,
+        UnresolvedLinkException, AccessControlException, IOException {
+    return myFs.resolvePath(p);
+  }
 
   @Override
   public int getUriDefaultPort() {

src/java/org/apache/hadoop/fs/viewfs/ChRootedFileSystem.java

@@ -20,6 +20,7 @@ import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.List;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
@@ -34,6 +35,7 @@ import org.apache.hadoop.fs.FsServerDefaults;
 import org.apache.hadoop.fs.FsStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.util.Progressable;
 
 /**
@@ -114,7 +116,7 @@ class ChRootedFileSystem extends FileSystem {
           chRootPathPart.toString().substring(1));
 
     workingDir = getHomeDirectory();
-    // We don't use the wd of the myFs bu set it to root.
+    // We don't use the wd of the myFs,  (lets set it to root anyway)
     myFs.setWorkingDirectory(chRootPathPart);
   }
   
@@ -210,6 +212,7 @@ class ChRootedFileSystem extends FileSystem {
   
 
   @Override
+  @SuppressWarnings("deprecation")
   public boolean delete(Path f) throws IOException {
    return delete(f, true);
   }
@@ -303,4 +306,9 @@ class ChRootedFileSystem extends FileSystem {
   public void setVerifyChecksum(final boolean verifyChecksum)  {
     myFs.setVerifyChecksum(verifyChecksum);
   }
+  
+  @Override
+  public List<Token<?>> getDelegationTokens(String renewer) throws IOException {
+    return myFs.getDelegationTokens(renewer);
+  }
 }

src/java/org/apache/hadoop/fs/viewfs/ChRootedFs.java

@@ -16,10 +16,12 @@
  * limitations under the License.
  */
 package org.apache.hadoop.fs.viewfs;
+import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.EnumSet;
+import java.util.List;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
@@ -35,6 +37,7 @@ import org.apache.hadoop.fs.FsStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.UnresolvedLinkException;
 import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.util.Progressable;
 
 /**
@@ -143,6 +146,13 @@ class ChRootedFs extends AbstractFileSystem {
     return null;
   }
   
+  
+  public Path getResolvedQualifiedPath(final Path f)
+      throws FileNotFoundException {
+    return myFs.makeQualified(
+        new Path(chRootPathPartString + f.toUri().toString()));
+  }
+  
   @Override
   public FSDataOutputStream createInternal(final Path f,
       final EnumSet<CreateFlag> flag, final FsPermission absolutePermission,
@@ -288,4 +298,10 @@ class ChRootedFs extends AbstractFileSystem {
   public Path getLinkTarget(final Path f) throws IOException {
     return myFs.getLinkTarget(fullPath(f));
   }
+  
+  
+  @Override
+  public List<Token<?>> getDelegationTokens(String renewer) throws IOException {
+    return myFs.getDelegationTokens(renewer);
+  }
 }

src/java/org/apache/hadoop/fs/viewfs/InodeTree.java

@@ -21,7 +21,9 @@ import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.ArrayList;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
 
@@ -60,6 +62,19 @@ abstract class InodeTree<T> {
   
   final INodeDir<T> root; // the root of the mount table
   
+  List<MountPoint<T>> mountPoints = new ArrayList<MountPoint<T>>();
+  
+  
+  static class MountPoint<T> {
+    String src;
+    INodeLink<T> target;
+    MountPoint(String srcPath, INodeLink<T> mountLink) {
+      src = srcPath;
+      target = mountLink;
+    }
+
+  }
+  
   /**
    * Breaks file path into component names.
    * @param path
@@ -173,14 +188,16 @@ abstract class InodeTree<T> {
      */
     Path getTargetLink() {
       // is merge link - use "," as separator between the merged URIs
-      String result = targetDirLinkList[0].toString();
-      for (int i=1; i < targetDirLinkList.length; ++i) {
-        result += "," + targetDirLinkList[0].toString();  
+      //String result = targetDirLinkList[0].toString();
+      StringBuilder result = new StringBuilder(targetDirLinkList[0].toString());
+      for (int i=1; i < targetDirLinkList.length; ++i) { 
+        result.append(',').append(targetDirLinkList[i].toString());
       }
-      return new Path(result);
+      return new Path(result.toString());
     }
   }
 
+
   private void createLink(final String src, final String target,
       final boolean isLinkMerge, final UserGroupInformation aUgi)
       throws URISyntaxException, IOException,
@@ -218,11 +235,11 @@ abstract class InodeTree<T> {
     String iPath = srcPaths[i];// last component
     if (curInode.resolveInternal(iPath) != null) {
       //  directory/link already exists
-      String path = srcPaths[0];
+      StringBuilder strB = new StringBuilder(srcPaths[0]);
       for (int j = 1; j <= i; ++j) {
-        path += "/" + srcPaths[j];
+        strB.append('/').append(srcPaths[j]);
       }
-      throw new FileAlreadyExistsException("Path " + path +
+      throw new FileAlreadyExistsException("Path " + strB +
             " already exists as dir; cannot create link here");
     }
     
@@ -242,7 +259,8 @@ abstract class InodeTree<T> {
       newLink = new INodeLink<T>(fullPath, aUgi,
           getTargetFileSystem(new URI(target)), new URI(target));
     }
-    curInode.addLink(iPath, newLink); 
+    curInode.addLink(iPath, newLink);
+    mountPoints.add(new MountPoint<T>(src, newLink));
   }
   
   /**
@@ -273,18 +291,19 @@ abstract class InodeTree<T> {
    * @throws FileAlreadyExistsException
    * @throws IOException
    */
-  protected InodeTree(final Configuration config, String viewName)
+  protected InodeTree(final Configuration config, final String viewName)
       throws UnsupportedFileSystemException, URISyntaxException,
     FileAlreadyExistsException, IOException { 
-    if (viewName == null) {
-      viewName = Constants.CONFIG_VIEWFS_DEFAULT_MOUNT_TABLE;
+    String vName = viewName;
+    if (vName == null) {
+      vName = Constants.CONFIG_VIEWFS_DEFAULT_MOUNT_TABLE;
     }
     root = new INodeDir<T>("/", UserGroupInformation.getCurrentUser());
     root.InodeDirFs = getTargetFileSystem(root);
     root.isRoot = true;
     
     final String mtPrefix = Constants.CONFIG_VIEWFS_PREFIX + "." + 
-                            viewName + ".";
+                            vName + ".";
     final String linkPrefix = Constants.CONFIG_VIEWFS_LINK + ".";
     final String linkMergePrefix = Constants.CONFIG_VIEWFS_LINK_MERGE + ".";
     boolean gotMountTableEntry = false;
@@ -307,11 +326,12 @@ abstract class InodeTree<T> {
         }
         final String target = si.getValue(); // link or merge link
         createLink(src, target, isMergeLink, ugi); 
-      }  
+      }
     }
     if (!gotMountTableEntry) {
       throw new IOException(
-          "ViewFs: Cannot initialize: Empty Mount table in config");
+          "ViewFs: Cannot initialize: Empty Mount table in config for " + 
+             vName == null ? "viewfs:///" : ("viewfs://" + vName + "/"));
     }
   }
 
@@ -370,11 +390,11 @@ abstract class InodeTree<T> {
     for (i = 1; i < path.length - (resolveLastComponent ? 0 : 1); i++) {
       INode<T> nextInode = curInode.resolveInternal(path[i]);
       if (nextInode == null) {
-        String failedAt = path[0];
+        StringBuilder failedAt = new StringBuilder(path[0]);
         for ( int j = 1; j <=i; ++j) {
-          failedAt += "/" + path[j];
+          failedAt.append('/').append(path[j]);
         }
-        throw (new FileNotFoundException(failedAt));      
+        throw (new FileNotFoundException(failedAt.toString()));      
       }
 
       if (nextInode instanceof INodeLink) {
@@ -383,11 +403,11 @@ abstract class InodeTree<T> {
         if (i >= path.length-1) {
           remainingPath = SlashPath;
         } else {
-          String remainingPathStr = "/" + path[i+1];
+          StringBuilder remainingPathStr = new StringBuilder("/" + path[i+1]);
           for (int j = i+2; j< path.length; ++j) {
-            remainingPathStr += "/" + path[j];
+            remainingPathStr.append('/').append(path[j]);
           }
-          remainingPath = new Path(remainingPathStr);
+          remainingPath = new Path(remainingPathStr.toString());
         }
         final ResolveResult<T> res = 
           new ResolveResult<T>(ResultKind.isExternalDir,
@@ -407,15 +427,19 @@ abstract class InodeTree<T> {
       // for internal dirs rem-path does not start with / since the lookup
       // that follows will do a children.get(remaningPath) and will have to
       // strip-out the initial /
-      String remainingPathStr =  "/" + path[i];
+      StringBuilder remainingPathStr = new StringBuilder("/" + path[i]);
       for (int j = i+1; j< path.length; ++j) {
-        remainingPathStr += "/" + path[j];
+        remainingPathStr.append('/').append(path[j]);
       }
-      remainingPath = new Path(remainingPathStr);
+      remainingPath = new Path(remainingPathStr.toString());
     }
     final ResolveResult<T> res = 
        new ResolveResult<T>(ResultKind.isInternalDir,
            curInode.InodeDirFs, curInode.fullPath, remainingPath); 
     return res;
   }
+  
+  List<MountPoint<T>> getMountPoints() { 
+    return mountPoints;
+  }
 }

src/java/org/apache/hadoop/fs/viewfs/ViewFileSystem.java

@@ -23,6 +23,8 @@ import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.ArrayList;
+import java.util.List;
 import java.util.StringTokenizer;
 import java.util.Map.Entry;
 
@@ -45,6 +47,7 @@ import org.apache.hadoop.fs.viewfs.InodeTree.INode;
 import org.apache.hadoop.fs.viewfs.InodeTree.INodeLink;
 import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.util.Progressable;
 
 /**
@@ -55,8 +58,31 @@ import org.apache.hadoop.util.Progressable;
 @InterfaceAudience.Public
 @InterfaceStability.Evolving /*Evolving for a release,to be changed to Stable */
 public class ViewFileSystem extends FileSystem {
-  static final AccessControlException READONLY_MOUNTABLE =
-    new AccessControlException( "InternalDir of ViewFileSystem is readonly");
+  static AccessControlException readOnlyMountTable(final String operation,
+      final String p) {
+    return new AccessControlException( 
+        "InternalDir of ViewFileSystem is readonly; operation=" + operation + 
+        "Path=" + p);
+  }
+  static AccessControlException readOnlyMountTable(final String operation,
+      final Path p) {
+    return readOnlyMountTable(operation, p.toString());
+  }
+  
+  static public class MountPoint {
+    private Path src;       // the src of the mount
+    private URI[] targets; //  target of the mount; Multiple targets imply mergeMount
+    MountPoint(Path srcPath, URI[] targetURIs) {
+      src = srcPath;
+      targets = targetURIs;
+    }
+    Path getSrc() {
+      return src;
+    }
+    URI[] getTargets() {
+      return targets;
+    }
+  }
   
   final long creationTime; // of the the mount table
   final UserGroupInformation ugi; // the user/group of user who created mtable
@@ -197,23 +223,14 @@ public class ViewFileSystem extends FileSystem {
     return myUri;
   }
   
-  /**
-   * Return the fully-qualified path of path f - ie follow the path
-   * through the mount point.
-   * @param f path
-   * @return resolved fully-qualified path
-   * @throws FileNotFoundException
-   */
-  public Path getResolvedQualifiedPath(final Path f)
-      throws FileNotFoundException {
+  @Override
+  public Path resolvePath(final Path f) throws FileNotFoundException {
     final InodeTree.ResolveResult<FileSystem> res;
       res = fsState.resolve(getUriPath(f), true);
     if (res.isInternalDir()) {
       return f;
     }
-    final ChRootedFileSystem targetFs = 
-      (ChRootedFileSystem) res.targetFileSystem;
-    return targetFs.getResolvedQualifiedPath(res.remainingPath);
+    return res.targetFileSystem.resolvePath(res.remainingPath);
   }
   
   @Override
@@ -243,7 +260,7 @@ public class ViewFileSystem extends FileSystem {
     try {
       res = fsState.resolve(getUriPath(f), false);
     } catch (FileNotFoundException e) {
-        throw READONLY_MOUNTABLE;
+        throw readOnlyMountTable("create", f);
     }
     assert(res.remainingPath != null);
     return res.targetFileSystem.create(res.remainingPath, permission,
@@ -259,12 +276,13 @@ public class ViewFileSystem extends FileSystem {
       fsState.resolve(getUriPath(f), true);
     // If internal dir or target is a mount link (ie remainingPath is Slash)
     if (res.isInternalDir() || res.remainingPath == InodeTree.SlashPath) {
-      throw READONLY_MOUNTABLE;
+      throw readOnlyMountTable("delete", f);
     }
     return res.targetFileSystem.delete(res.remainingPath, recursive);
   }
   
   @Override
+  @SuppressWarnings("deprecation")
   public boolean delete(final Path f)
       throws AccessControlException, FileNotFoundException,
       IOException {
@@ -358,14 +376,13 @@ public class ViewFileSystem extends FileSystem {
       fsState.resolve(getUriPath(src), false); 
   
     if (resSrc.isInternalDir()) {
-      throw READONLY_MOUNTABLE;
+      throw readOnlyMountTable("rename", src);
     }
       
     InodeTree.ResolveResult<FileSystem> resDst = 
       fsState.resolve(getUriPath(dst), false);
     if (resDst.isInternalDir()) {
-      throw new AccessControlException(
-          "Cannot Rename within internal dirs of mount table: it is readOnly");
+          throw readOnlyMountTable("rename", dst);
     }
     /**
     // Alternate 1: renames within same file system - valid but we disallow
@@ -433,6 +450,38 @@ public class ViewFileSystem extends FileSystem {
     // points to many file systems. Noop for ViewFileSystem.
   }
   
+  public MountPoint[] getMountPoints() {
+    List<InodeTree.MountPoint<FileSystem>> mountPoints = 
+                  fsState.getMountPoints();
+    
+    MountPoint[] result = new MountPoint[mountPoints.size()];
+    for ( int i = 0; i < mountPoints.size(); ++i ) {
+      result[i] = new MountPoint(new Path(mountPoints.get(i).src), 
+                              mountPoints.get(i).target.targetDirLinkList);
+    }
+    return result;
+  }
+  
+ 
+  @Override
+  public List<Token<?>> getDelegationTokens(String renewer) throws IOException {
+    List<InodeTree.MountPoint<FileSystem>> mountPoints = 
+                fsState.getMountPoints();
+    int initialListSize  = 0;
+    for (InodeTree.MountPoint<FileSystem> im : mountPoints) {
+      initialListSize += im.target.targetDirLinkList.length; 
+    }
+    List<Token<?>> result = new ArrayList<Token<?>>(initialListSize);
+    for ( int i = 0; i < mountPoints.size(); ++i ) {
+      List<Token<?>> tokens = 
+        mountPoints.get(i).target.targetFileSystem.getDelegationTokens(renewer);
+      if (tokens != null) {
+        result.addAll(tokens);
+      }
+    }
+    return result;
+  }
+  
   /*
    * An instance of this class represents an internal dir of the viewFs 
    * that is internal dir of the mount table.
@@ -491,7 +540,7 @@ public class ViewFileSystem extends FileSystem {
     @Override
     public FSDataOutputStream append(final Path f, final int bufferSize,
         final Progressable progress) throws IOException {
-      throw READONLY_MOUNTABLE;
+      throw readOnlyMountTable("append", f);
     }
 
     @Override
@@ -499,17 +548,18 @@ public class ViewFileSystem extends FileSystem {
         final FsPermission permission, final boolean overwrite,
         final int bufferSize, final short replication, final long blockSize,
         final Progressable progress) throws AccessControlException {
-      throw READONLY_MOUNTABLE;
+      throw readOnlyMountTable("create", f);
     }
 
     @Override
     public boolean delete(final Path f, final boolean recursive)
         throws AccessControlException, IOException {
       checkPathIsSlash(f);
-      throw READONLY_MOUNTABLE;
+      throw readOnlyMountTable("delete", f);
     }
     
     @Override
+    @SuppressWarnings("deprecation")
     public boolean delete(final Path f)
         throws AccessControlException, IOException {
       return delete(f, true);
@@ -576,7 +626,11 @@ public class ViewFileSystem extends FileSystem {
       if (theInternalDir.isRoot & dir == null) {
         throw new FileAlreadyExistsException("/ already exits");
       }
-      throw READONLY_MOUNTABLE;
+      // Note dir starts with /
+      if (theInternalDir.children.containsKey(dir.toString().substring(1))) {
+        return true; // this is the stupid semantics of FileSystem
+      }
+      throw readOnlyMountTable("mkdirs",  dir);
     }
 
     @Override
@@ -591,35 +645,35 @@ public class ViewFileSystem extends FileSystem {
         IOException {
       checkPathIsSlash(src);
       checkPathIsSlash(dst);
-      throw READONLY_MOUNTABLE;     
+      throw readOnlyMountTable("rename", src);     
     }
 
     @Override
     public void setOwner(Path f, String username, String groupname)
         throws AccessControlException, IOException {
       checkPathIsSlash(f);
-      throw READONLY_MOUNTABLE;
+      throw readOnlyMountTable("setOwner", f);
     }
 
     @Override
     public void setPermission(Path f, FsPermission permission)
         throws AccessControlException, IOException {
       checkPathIsSlash(f);
-      throw READONLY_MOUNTABLE;    
+      throw readOnlyMountTable("setPermission", f);    
     }
 
     @Override
     public boolean setReplication(Path f, short replication)
         throws AccessControlException, IOException {
       checkPathIsSlash(f);
-      throw READONLY_MOUNTABLE;
+      throw readOnlyMountTable("setReplication", f);
     }
 
     @Override
     public void setTimes(Path f, long mtime, long atime)
         throws AccessControlException, IOException {
       checkPathIsSlash(f);
-      throw READONLY_MOUNTABLE;    
+      throw readOnlyMountTable("setTimes", f);    
     }
 
     @Override

src/java/org/apache/hadoop/fs/viewfs/ViewFs.java

@@ -23,7 +23,9 @@ import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.ArrayList;
 import java.util.EnumSet;
+import java.util.List;
 import java.util.Map.Entry;
 
 import org.apache.hadoop.classification.InterfaceAudience;
@@ -51,88 +53,10 @@ import org.apache.hadoop.fs.viewfs.InodeTree.INode;
 import org.apache.hadoop.fs.viewfs.InodeTree.INodeLink;
 import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.util.Progressable;
 
-/**
- * 
- * ViewFs (extends the AbstractFileSystem interface) implements a client-side
- * mount table. The viewFs file system is implemented completely in memory on
- * the client side. The client-side mount table allows a client to provide a 
- * customized view of a file system namespace that is composed from 
- * one or more individual file systems (a localFs or Hdfs, S3fs, etc).
- * For example one could have a mount table that provides links such as
- * <ul>
- * <li>  /user          -> hdfs://nnContainingUserDir/user
- * <li>  /project/foo   -> hdfs://nnProject1/projects/foo
- * <li>  /project/bar   -> hdfs://nnProject2/projects/bar
- * <li>  /tmp           -> hdfs://nnTmp/privateTmpForUserXXX
- * </ul> 
- * 
- * ViewFs is specified with the following URI: <b>viewfs:///</b> 
- * <p>
- * To use viewfs one would typically set the default file system in the
- * config  (i.e. fs.defaultFS = viewfs:///) along with the
- * mount table config variables as described below. 
- * 
- * <p>
- * <b> ** Config variables to specify the mount table entries ** </b>
- * <p>
- * 
- * The file system is initialized from the standard Hadoop config through
- * config variables.
- * See {@link FsConstants} for URI and Scheme constants; 
- * See {@link Constants} for config var constants; 
- * see {@link ConfigUtil} for convenient lib.
- * 
- * <p>
- * All the mount table config entries for view fs are prefixed by 
- * <b>fs.viewFs.</b>
- * For example the above example can be specfied with the following
- *  config variables:
- *  <ul>
- *  <li> fs.viewFs.defaultMT.link./user=hdfs://nnContainingUserDir/user
- *  <li> fs.viewFs.defaultMT.link./project/foo=hdfs://nnProject1/projects/foo
- *  <li> fs.viewFs.defaultMT.link./project/bar=hdfs://nnProject2/projects/bar
- *  <li> fs.viewFs.defaultMT.link./tmp=hdfs://nnTmp/privateTmpForUserXXX
- *  </ul>
- *  
- * The default mount table (when no authority is specified) is 
- * from config variables prefixed by <b>fs.viewFs.defaultMT </b>
- * The authority component of a URI can be used to specify a different mount
- * table. For example,
- * <ul>
- * <li>  viewfs://sanjayMountable/
- * </ul>
- * is initialized from the fs.viewFs.sanjayMountable.* config variables.
- * 
- *  <p> 
- *  <b> **** Merge Mounts **** </b>(NOTE: merge mounts are not implemented yet.)
- *  <p>
- *  
- *   One can also use "MergeMounts" to merge several directories (this is
- *   sometimes  called union-mounts or junction-mounts in the literature.
- *   For example of the home directories are stored on say two file systems
- *   (because they do not fit on one) then one could specify a mount
- *   entry such as following merges two dirs:
- *   <ul>
- *   <li> /user -> hdfs://nnUser1/user,hdfs://nnUser2/user
- *   </ul>
 
- *  Such a mergeLink can be specifed with the following config var where ","
- *  is used as the seperater for each of links to be merged:
- *  <ul>
- *  <li> fs.viewFs.defaultMT.linkMerge./user=
- *            hdfs://nnUser1/user,hdfs://nnUser1/user
- *  </ul>
- *   A special case of the merge mount is where mount table's root is merged
- *   with the root (slash) of another file system:
- *   <ul>
- *   <li>    fs.viewFs.defaultMT.linkMergeSlash=hdfs://nn99/
- *   </ul>
- *   In this cases the root of the mount table is merged with the root of
- *            <b>hdfs://nn99/ </b> 
- * 
- */
 /**
  * ViewFs (extends the AbstractFileSystem interface) implements a client-side
  * mount table. The viewFs file system is implemented completely in memory on
@@ -147,13 +71,11 @@ import org.apache.hadoop.util.Progressable;
  * <li>  /tmp           -> hdfs://nnTmp/privateTmpForUserXXX
  * </ul> 
  * 
- * ViewFileSystem is specified with the following URI: <b>viewfs:///</b> 
+ * ViewFs is specified with the following URI: <b>viewfs:///</b> 
  * <p>
  * To use viewfs one would typically set the default file system in the
  * config  (i.e. fs.default.name< = viewfs:///) along with the
  * mount table config variables as described below. 
- * If your core-site.xml does not have the following config value please add it
- * to your config: fs.viewfs.impl = org.apache.hadoop.fs.viewfs.ViewFileSystem
  * 
  * <p>
  * <b> ** Config variables to specify the mount table entries ** </b>
@@ -182,13 +104,13 @@ import org.apache.hadoop.util.Progressable;
  *  </ul>
  *  
  * The default mount table (when no authority is specified) is 
- * from config variables prefixed by <b>fs.viewFs.defaultMT </b>
+ * from config variables prefixed by <b>fs.viewFs.mounttable.default </b>
  * The authority component of a URI can be used to specify a different mount
  * table. For example,
  * <ul>
  * <li>  viewfs://sanjayMountable/
  * </ul>
- * is initialized from the fs.viewFs.sanjayMountable.* config variables.
+ * is initialized from fs.viewFs.mounttable.sanjayMountable.* config variables.
  * 
  *  <p> 
  *  <b> **** Merge Mounts **** </b>(NOTE: merge mounts are not implemented yet.)
@@ -202,7 +124,6 @@ import org.apache.hadoop.util.Progressable;
  *   <ul>
  *   <li> /user -> hdfs://nnUser1/user,hdfs://nnUser2/user
  *   </ul>
-
  *  Such a mergeLink can be specified with the following config var where ","
  *  is used as the separator for each of links to be merged:
  *  <ul>
@@ -226,8 +147,33 @@ public class ViewFs extends AbstractFileSystem {
   final Configuration config;
   InodeTree<AbstractFileSystem> fsState;  // the fs state; ie the mount table
   
-  static final AccessControlException READONLY_MOUNTABLE =
-    new AccessControlException("InternalDir of ViewFs is readonly");
+  static AccessControlException readOnlyMountTable(final String operation,
+      final String p) {
+    return new AccessControlException( 
+        "InternalDir of ViewFileSystem is readonly; operation=" + operation + 
+        "Path=" + p);
+  }
+  static AccessControlException readOnlyMountTable(final String operation,
+      final Path p) {
+    return readOnlyMountTable(operation, p.toString());
+  }
+  
+  
+  static public class MountPoint {
+    private Path src;       // the src of the mount
+    private URI[] targets; //  target of the mount; Multiple targets imply mergeMount
+    MountPoint(Path srcPath, URI[] targetURIs) {
+      src = srcPath;
+      targets = targetURIs;
+    }
+    Path getSrc() {
+      return src;
+    }
+    URI[] getTargets() {
+      return targets;
+    }
+  }
+  
   public ViewFs(final Configuration conf) throws IOException,
       URISyntaxException {
     this(FsConstants.VIEWFS_URI, conf);
@@ -288,6 +234,18 @@ public class ViewFs extends AbstractFileSystem {
     return -1;
   }
  
+  @Override
+  public Path resolvePath(final Path f) throws FileNotFoundException,
+          AccessControlException, UnresolvedLinkException, IOException {
+    final InodeTree.ResolveResult<AbstractFileSystem> res;
+      res = fsState.resolve(getUriPath(f), true);
+    if (res.isInternalDir()) {
+      return f;
+    }
+    return res.targetFileSystem.resolvePath(res.remainingPath);
+
+  }
+  
   @Override
   public FSDataOutputStream createInternal(final Path f,
       final EnumSet<CreateFlag> flag, final FsPermission absolutePermission,
@@ -302,7 +260,7 @@ public class ViewFs extends AbstractFileSystem {
       res = fsState.resolve(getUriPath(f), false);
     } catch (FileNotFoundException e) {
       if (createParent) {
-        throw READONLY_MOUNTABLE;
+        throw readOnlyMountTable("create", f);
       } else {
         throw e;
       }
@@ -526,7 +484,7 @@ public class ViewFs extends AbstractFileSystem {
       res = fsState.resolve(getUriPath(link), false);
     } catch (FileNotFoundException e) {
       if (createParent) {
-        throw READONLY_MOUNTABLE;
+        throw readOnlyMountTable("createSymlink", link);
       } else {
         throw e;
       }
@@ -587,6 +545,39 @@ public class ViewFs extends AbstractFileSystem {
     // points to many file systems. Noop for ViewFs. 
   }
   
+  public MountPoint[] getMountPoints() {
+    List<InodeTree.MountPoint<AbstractFileSystem>> mountPoints = 
+                  fsState.getMountPoints();
+    
+    MountPoint[] result = new MountPoint[mountPoints.size()];
+    for ( int i = 0; i < mountPoints.size(); ++i ) {
+      result[i] = new MountPoint(new Path(mountPoints.get(i).src), 
+                              mountPoints.get(i).target.targetDirLinkList);
+    }
+    return result;
+  }
+  
+  @Override
+  public List<Token<?>> getDelegationTokens(String renewer) throws IOException {
+    List<InodeTree.MountPoint<AbstractFileSystem>> mountPoints = 
+                fsState.getMountPoints();
+    int initialListSize  = 0;
+    for (InodeTree.MountPoint<AbstractFileSystem> im : mountPoints) {
+      initialListSize += im.target.targetDirLinkList.length; 
+    }
+    List<Token<?>> result = new ArrayList<Token<?>>(initialListSize);
+    for ( int i = 0; i < mountPoints.size(); ++i ) {
+      List<Token<?>> tokens = 
+        mountPoints.get(i).target.targetFileSystem.getDelegationTokens(renewer);
+      if (tokens != null) {
+        result.addAll(tokens);
+      }
+    }
+    return result;
+  }
+
+  
+  
   /*
    * An instance of this class represents an internal dir of the viewFs 
    * ie internal dir of the mount table.
@@ -631,14 +622,14 @@ public class ViewFs extends AbstractFileSystem {
         FileAlreadyExistsException, FileNotFoundException,
         ParentNotDirectoryException, UnsupportedFileSystemException,
         UnresolvedLinkException, IOException {
-      throw READONLY_MOUNTABLE;
+      throw readOnlyMountTable("create", f);
     }
 
     @Override
     public boolean delete(final Path f, final boolean recursive)
         throws AccessControlException, IOException {
       checkPathIsSlash(f);
-      throw READONLY_MOUNTABLE;
+      throw readOnlyMountTable("delete", f);
     }
 
     @Override
@@ -746,7 +737,7 @@ public class ViewFs extends AbstractFileSystem {
       if (theInternalDir.isRoot & dir == null) {
         throw new FileAlreadyExistsException("/ already exits");
       }
-      throw READONLY_MOUNTABLE;
+      throw readOnlyMountTable("mkdir", dir);
     }
 
     @Override
@@ -761,7 +752,7 @@ public class ViewFs extends AbstractFileSystem {
         throws AccessControlException, IOException {
       checkPathIsSlash(src);
       checkPathIsSlash(dst);
-      throw READONLY_MOUNTABLE;     
+      throw readOnlyMountTable("rename", src);     
     }
 
     @Override
@@ -772,7 +763,7 @@ public class ViewFs extends AbstractFileSystem {
     @Override
     public void createSymlink(final Path target, final Path link,
         final boolean createParent) throws AccessControlException {
-      throw READONLY_MOUNTABLE;    
+      throw readOnlyMountTable("createSymlink", link);    
     }
 
     @Override
@@ -785,34 +776,34 @@ public class ViewFs extends AbstractFileSystem {
     public void setOwner(final Path f, final String username,
         final String groupname) throws AccessControlException, IOException {
       checkPathIsSlash(f);
-      throw READONLY_MOUNTABLE;
+      throw readOnlyMountTable("setOwner", f);
     }
 
     @Override
     public void setPermission(final Path f, final FsPermission permission)
         throws AccessControlException, IOException {
       checkPathIsSlash(f);
-      throw READONLY_MOUNTABLE;    
+      throw readOnlyMountTable("setPermission", f);    
     }
 
     @Override
     public boolean setReplication(final Path f, final short replication)
         throws AccessControlException, IOException {
       checkPathIsSlash(f);
-      throw READONLY_MOUNTABLE;
+      throw readOnlyMountTable("setReplication", f);
     }
 
     @Override
     public void setTimes(final Path f, final long mtime, final long atime)
         throws AccessControlException, IOException {
       checkPathIsSlash(f);
-      throw READONLY_MOUNTABLE;    
+      throw readOnlyMountTable("setTimes", f);    
     }
 
     @Override
     public void setVerifyChecksum(final boolean verifyChecksum)
         throws AccessControlException {
-      throw READONLY_MOUNTABLE;   
+      throw readOnlyMountTable("setVerifyChecksum", "");   
     }
   }
 }

src/java/org/apache/hadoop/fs/viewfs/ViewFsFileStatus.java

@@ -1,3 +1,20 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package org.apache.hadoop.fs.viewfs;
 
 import java.io.IOException;
@@ -20,6 +37,15 @@ class ViewFsFileStatus extends FileStatus {
      modifiedPath = newPath;
    }
    
+   @Override
+   public boolean equals(Object o) {
+     return super.equals(o);
+   }
+   
+   public int hashCode() {
+     return super.hashCode();
+   }
+   
    @Override
    public long getLen() {
      return myFs.getLen();
@@ -36,6 +62,7 @@ class ViewFsFileStatus extends FileStatus {
    }
    
    @Override
+   @SuppressWarnings("deprecation")
    public boolean isDir() {
      return myFs.isDirectory();
    }

src/test/core/org/apache/hadoop/fs/FileContextPermissionBase.java

@@ -93,7 +93,8 @@ public abstract class FileContextPermissionBase {
       return;
     }
     String filename = "foo";
-    Path f = createFile(fc, filename);
+    Path f = getTestRootPath(fc, filename);
+    createFile(fc, filename);
     doFilePermissionCheck(FileContext.DEFAULT_PERM.applyUMask(fc.getUMask()),
                         fc.getFileStatus(f).getPermission());
   }
@@ -107,7 +108,8 @@ public abstract class FileContextPermissionBase {
     }
 
     String filename = "foo";
-    Path f = createFile(fc, filename);
+    Path f = getTestRootPath(fc, filename);
+    createFile(fc, f);
 
     try {
       // create files and manipulate them.
@@ -131,7 +133,8 @@ public abstract class FileContextPermissionBase {
     }
 
     String filename = "bar";
-    Path f = createFile(fc, filename);
+    Path f = getTestRootPath(fc, filename);
+    createFile(fc, f);
     List<String> groups = null;
     try {
       groups = getGroups();

src/test/core/org/apache/hadoop/fs/FileContextTestHelper.java

@@ -92,7 +92,7 @@ public final class FileContextTestHelper {
   /*
    * Create files with numBlocks blocks each with block size blockSize.
    */
-  public static void createFile(FileContext fc, Path path, int numBlocks,
+  public static long createFile(FileContext fc, Path path, int numBlocks,
       CreateOpts... options) throws IOException {
     BlockSize blockSizeOpt = 
       (BlockSize) CreateOpts.getOpt(CreateOpts.BlockSize.class, options);
@@ -103,33 +103,33 @@ public final class FileContextTestHelper {
     byte[] data = getFileData(numBlocks, blockSize);
     out.write(data, 0, data.length);
     out.close();
+    return data.length;
   }
 
-  public static void createFile(FileContext fc, Path path, int numBlocks,
+  public static long  createFile(FileContext fc, Path path, int numBlocks,
       int blockSize) throws IOException {
-    createFile(fc, path, numBlocks, CreateOpts.blockSize(blockSize), 
+    return createFile(fc, path, numBlocks, CreateOpts.blockSize(blockSize), 
         CreateOpts.createParent());
   }
 
-  public static void createFile(FileContext fc, Path path) throws IOException {
-    createFile(fc, path, DEFAULT_NUM_BLOCKS, CreateOpts.createParent());
+  public static long createFile(FileContext fc, Path path) throws IOException {
+    return createFile(fc, path, DEFAULT_NUM_BLOCKS, CreateOpts.createParent());
   }
 
-  public static Path createFile(FileContext fc, String name) throws IOException {
+  public static long createFile(FileContext fc, String name) throws IOException {
     Path path = getTestRootPath(fc, name);
-    createFile(fc, path);
-    return path;
+    return createFile(fc, path);
   }
   
-  public static void createFileNonRecursive(FileContext fc, String name)
+  public static long createFileNonRecursive(FileContext fc, String name)
   throws IOException {
     Path path = getTestRootPath(fc, name);
-    createFileNonRecursive(fc, path);
+    return createFileNonRecursive(fc, path);
   }
 
-  public static void createFileNonRecursive(FileContext fc, Path path)
+  public static long createFileNonRecursive(FileContext fc, Path path)
       throws IOException {
-    createFile(fc, path, DEFAULT_NUM_BLOCKS, CreateOpts.donotCreateParent());
+    return createFile(fc, path, DEFAULT_NUM_BLOCKS, CreateOpts.donotCreateParent());
   }
 
   public static void appendToFile(FileContext fc, Path path, int numBlocks,

src/test/core/org/apache/hadoop/fs/FileSystemTestHelper.java

@@ -98,7 +98,7 @@ public final class FileSystemTestHelper {
   /*
    * Create files with numBlocks blocks each with block size blockSize.
    */
-  public static void createFile(FileSystem fSys, Path path, int numBlocks,
+  public static long createFile(FileSystem fSys, Path path, int numBlocks,
       int blockSize, boolean createParent) throws IOException {
     FSDataOutputStream out = 
       fSys.create(path, false, 4096, fSys.getDefaultReplication(), blockSize );
@@ -106,21 +106,21 @@ public final class FileSystemTestHelper {
     byte[] data = getFileData(numBlocks, blockSize);
     out.write(data, 0, data.length);
     out.close();
+    return data.length;
   }
 
-  public static void createFile(FileSystem fSys, Path path, int numBlocks,
+  public static long createFile(FileSystem fSys, Path path, int numBlocks,
       int blockSize) throws IOException {
-    createFile(fSys, path, numBlocks, blockSize, true);
+      return createFile(fSys, path, numBlocks, blockSize, true);
     }
 
-  public static void createFile(FileSystem fSys, Path path) throws IOException {
-    createFile(fSys, path, DEFAULT_NUM_BLOCKS, DEFAULT_BLOCK_SIZE, true);
+  public static long createFile(FileSystem fSys, Path path) throws IOException {
+    return createFile(fSys, path, DEFAULT_NUM_BLOCKS, DEFAULT_BLOCK_SIZE, true);
   }
 
-  public static Path createFile(FileSystem fSys, String name) throws IOException {
+  public static long createFile(FileSystem fSys, String name) throws IOException {
     Path path = getTestRootPath(fSys, name);
-    createFile(fSys, path);
-    return path;
+    return createFile(fSys, path);
   }
 
   public static boolean exists(FileSystem fSys, Path p) throws IOException {

src/test/core/org/apache/hadoop/fs/viewfs/TestViewFileSystemLocalFileSystem.java

@@ -1,4 +1,20 @@
-
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package org.apache.hadoop.fs.viewfs;
 
 

src/test/core/org/apache/hadoop/fs/viewfs/TestViewFileSystemWithAuthorityLocalFileSystem.java

@@ -1,4 +1,20 @@
-
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package org.apache.hadoop.fs.viewfs;
 
 

src/test/core/org/apache/hadoop/fs/viewfs/TestViewFsConfig.java

@@ -1,3 +1,20 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package org.apache.hadoop.fs.viewfs;
 
 import java.io.IOException;

src/test/core/org/apache/hadoop/fs/viewfs/TestViewFsLocalFs.java

@@ -17,16 +17,15 @@
  */
 package org.apache.hadoop.fs.viewfs;
 
-
 import org.apache.hadoop.fs.FileContext;
 
+
 import org.junit.After;
 import org.junit.Before;
 
 
 public class TestViewFsLocalFs extends ViewFsBaseTest {
 
-
   @Before
   public void setUp() throws Exception {
     // create the test root on local_fs
@@ -39,5 +38,4 @@ public class TestViewFsLocalFs extends ViewFsBaseTest {
   public void tearDown() throws Exception {
     super.tearDown();
   }
- 
 }

src/test/core/org/apache/hadoop/fs/viewfs/TestViewFsWithAuthorityLocalFs.java

@@ -1,4 +1,20 @@
-
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package org.apache.hadoop.fs.viewfs;
 
 

src/test/core/org/apache/hadoop/fs/viewfs/TestViewfsFileStatus.java

@@ -0,0 +1,89 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.fs.viewfs;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.net.URISyntaxException;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.fs.FsConstants;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.DataInputBuffer;
+import org.apache.hadoop.io.DataOutputBuffer;
+
+import org.junit.AfterClass;
+import org.junit.Test;
+import static org.junit.Assert.*;
+
+/**
+ * The FileStatus is being serialized in MR as jobs are submitted.
+ * Since viewfs has overlayed ViewFsFileStatus, we ran into
+ * serialization problems. THis test is test the fix.
+ */
+public class TestViewfsFileStatus {
+
+  private static final File TEST_DIR =
+    new File(System.getProperty("test.build.data", "/tmp"),
+          TestViewfsFileStatus.class.getSimpleName());
+
+  @Test
+  public void testFileStatusSerialziation()
+      throws IOException, URISyntaxException {
+    String testfilename = "testFileStatusSerialziation";
+    TEST_DIR.mkdirs();
+    File infile = new File(TEST_DIR, testfilename);
+    final byte[] content = "dingos".getBytes();
+
+    FileOutputStream fos = null;
+    try {
+      fos = new FileOutputStream(infile);
+      fos.write(content);
+    } finally {
+      if (fos != null) {
+        fos.close();
+      }
+    }
+    assertEquals((long)content.length, infile.length());
+
+    Configuration conf = new Configuration();
+    ConfigUtil.addLink(conf, "/foo/bar/baz", TEST_DIR.toURI());
+    FileSystem vfs = FileSystem.get(FsConstants.VIEWFS_URI, conf);
+    assertEquals(ViewFileSystem.class, vfs.getClass());
+    FileStatus stat = vfs.getFileStatus(new Path("/foo/bar/baz", testfilename));
+    assertEquals(content.length, stat.getLen());
+    // check serialization/deserialization
+    DataOutputBuffer dob = new DataOutputBuffer();
+    stat.write(dob);
+    DataInputBuffer dib = new DataInputBuffer();
+    dib.reset(dob.getData(), 0, dob.getLength());
+    FileStatus deSer = new FileStatus();
+    deSer.readFields(dib);
+    assertEquals(content.length, deSer.getLen());
+  }
+
+  @AfterClass
+  public static void cleanup() throws IOException {
+    FileUtil.fullyDelete(TEST_DIR);
+  }
+
+}

src/test/core/org/apache/hadoop/fs/viewfs/ViewFileSystemBaseTest.java

@@ -19,6 +19,7 @@ package org.apache.hadoop.fs.viewfs;
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.util.List;
 
 
 import org.apache.hadoop.conf.Configuration;
@@ -31,7 +32,9 @@ import org.apache.hadoop.fs.FsConstants;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.viewfs.ConfigUtil;
 import org.apache.hadoop.fs.viewfs.ViewFileSystem;
+import org.apache.hadoop.fs.viewfs.ViewFileSystem.MountPoint;
 import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security.token.Token;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -106,6 +109,24 @@ public class ViewFileSystemBaseTest {
     fsTarget.delete(FileSystemTestHelper.getTestRootPath(fsTarget), true);
   }
   
+  @Test
+  public void testGetMountPoints() {
+    ViewFileSystem viewfs = (ViewFileSystem) fsView;
+    MountPoint[] mountPoints = viewfs.getMountPoints();
+    Assert.assertEquals(7, mountPoints.length); 
+  }
+  
+  /**
+   * This default implementation is when viewfs has mount points
+   * into file systems, such as LocalFs that do no have delegation tokens.
+   * It should be overridden for when mount points into hdfs.
+   */
+  @Test
+  public void testGetDelegationTokens() throws IOException {
+    List<Token<?>> delTokens = 
+        fsView.getDelegationTokens("sanjay");
+    Assert.assertEquals(0, delTokens.size()); 
+  }
   
   @Test
   public void testBasicPaths() {
@@ -136,75 +157,112 @@ public class ViewFileSystemBaseTest {
   public void testOperationsThroughMountLinks() throws IOException {
     // Create file 
     FileSystemTestHelper.createFile(fsView, "/user/foo");
-    Assert.assertTrue(fsView.isFile(new Path("/user/foo")));
-    Assert.assertTrue(fsTarget.isFile(new Path(targetTestRoot,"user/foo")));
+    Assert.assertTrue("Created file should be type file",
+        fsView.isFile(new Path("/user/foo")));
+    Assert.assertTrue("Target of created file should be type file",
+        fsTarget.isFile(new Path(targetTestRoot,"user/foo")));
     
     // Delete the created file
-    Assert.assertTrue(fsView.delete(new Path("/user/foo"), false));
-    Assert.assertFalse(fsView.exists(new Path("/user/foo")));
-    Assert.assertFalse(fsTarget.exists(new Path(targetTestRoot,"user/foo")));
+    Assert.assertTrue("Delete should suceed",
+        fsView.delete(new Path("/user/foo"), false));
+    Assert.assertFalse("File should not exist after delete",
+        fsView.exists(new Path("/user/foo")));
+    Assert.assertFalse("Target File should not exist after delete",
+        fsTarget.exists(new Path(targetTestRoot,"user/foo")));
     
     // Create file with a 2 component dirs
     FileSystemTestHelper.createFile(fsView, "/internalDir/linkToDir2/foo");
-    Assert.assertTrue(fsView.isFile(new Path("/internalDir/linkToDir2/foo")));
-    Assert.assertTrue(fsTarget.isFile(new Path(targetTestRoot,"dir2/foo")));
+    Assert.assertTrue("Created file should be type file",
+        fsView.isFile(new Path("/internalDir/linkToDir2/foo")));
+    Assert.assertTrue("Target of created file should be type file",
+        fsTarget.isFile(new Path(targetTestRoot,"dir2/foo")));
     
     // Delete the created file
-    Assert.assertTrue(fsView.delete(new Path("/internalDir/linkToDir2/foo"), false));
-    Assert.assertFalse(fsView.exists(new Path("/internalDir/linkToDir2/foo")));
-    Assert.assertFalse(fsTarget.exists(new Path(targetTestRoot,"dir2/foo")));
+    Assert.assertTrue("Delete should suceed",
+        fsView.delete(new Path("/internalDir/linkToDir2/foo"), false));
+    Assert.assertFalse("File should not exist after delete",
+        fsView.exists(new Path("/internalDir/linkToDir2/foo")));
+    Assert.assertFalse("Target File should not exist after delete",
+        fsTarget.exists(new Path(targetTestRoot,"dir2/foo")));
     
     
     // Create file with a 3 component dirs
     FileSystemTestHelper.createFile(fsView, "/internalDir/internalDir2/linkToDir3/foo");
-    Assert.assertTrue(fsView.isFile(new Path("/internalDir/internalDir2/linkToDir3/foo")));
-    Assert.assertTrue(fsTarget.isFile(new Path(targetTestRoot,"dir3/foo")));
+    Assert.assertTrue("Created file should be type file",
+        fsView.isFile(new Path("/internalDir/internalDir2/linkToDir3/foo")));
+    Assert.assertTrue("Target of created file should be type file",
+        fsTarget.isFile(new Path(targetTestRoot,"dir3/foo")));
     
     // Recursive Create file with missing dirs
-    FileSystemTestHelper.createFile(fsView, "/internalDir/linkToDir2/missingDir/miss2/foo");
-    Assert.assertTrue(fsView.isFile(new Path("/internalDir/linkToDir2/missingDir/miss2/foo")));
-    Assert.assertTrue(fsTarget.isFile(new Path(targetTestRoot,"dir2/missingDir/miss2/foo")));
+    FileSystemTestHelper.createFile(fsView,
+        "/internalDir/linkToDir2/missingDir/miss2/foo");
+    Assert.assertTrue("Created file should be type file",
+        fsView.isFile(new Path("/internalDir/linkToDir2/missingDir/miss2/foo")));
+    Assert.assertTrue("Target of created file should be type file",
+        fsTarget.isFile(new Path(targetTestRoot,"dir2/missingDir/miss2/foo")));
 
     
     // Delete the created file
-    Assert.assertTrue(fsView.delete(new Path("/internalDir/internalDir2/linkToDir3/foo"), false));
-    Assert.assertFalse(fsView.exists(new Path("/internalDir/internalDir2/linkToDir3/foo")));
-    Assert.assertFalse(fsTarget.exists(new Path(targetTestRoot,"dir3/foo")));
+    Assert.assertTrue("Delete should succeed",
+        fsView.delete(
+            new Path("/internalDir/internalDir2/linkToDir3/foo"), false));
+    Assert.assertFalse("File should not exist after delete",
+        fsView.exists(new Path("/internalDir/internalDir2/linkToDir3/foo")));
+    Assert.assertFalse("Target File should not exist after delete",
+        fsTarget.exists(new Path(targetTestRoot,"dir3/foo")));
     
       
     // mkdir
     fsView.mkdirs(FileSystemTestHelper.getTestRootPath(fsView, "/user/dirX"));
-    Assert.assertTrue(fsView.isDirectory(new Path("/user/dirX")));
-    Assert.assertTrue(fsTarget.isDirectory(new Path(targetTestRoot,"user/dirX")));
+    Assert.assertTrue("New dir should be type dir", 
+        fsView.isDirectory(new Path("/user/dirX")));
+    Assert.assertTrue("Target of new dir should be of type dir",
+        fsTarget.isDirectory(new Path(targetTestRoot,"user/dirX")));
     
-    fsView.mkdirs(FileSystemTestHelper.getTestRootPath(fsView, "/user/dirX/dirY"));
-    Assert.assertTrue(fsView.isDirectory(new Path("/user/dirX/dirY")));
-    Assert.assertTrue(fsTarget.isDirectory(new Path(targetTestRoot,"user/dirX/dirY")));
+    fsView.mkdirs(
+        FileSystemTestHelper.getTestRootPath(fsView, "/user/dirX/dirY"));
+    Assert.assertTrue("New dir should be type dir", 
+        fsView.isDirectory(new Path("/user/dirX/dirY")));
+    Assert.assertTrue("Target of new dir should be of type dir",
+        fsTarget.isDirectory(new Path(targetTestRoot,"user/dirX/dirY")));
     
 
     // Delete the created dir
-    Assert.assertTrue(fsView.delete(new Path("/user/dirX/dirY"), false));
-    Assert.assertFalse(fsView.exists(new Path("/user/dirX/dirY")));
-    Assert.assertFalse(fsTarget.exists(new Path(targetTestRoot,"user/dirX/dirY")));
+    Assert.assertTrue("Delete should succeed",
+        fsView.delete(new Path("/user/dirX/dirY"), false));
+    Assert.assertFalse("File should not exist after delete",
+        fsView.exists(new Path("/user/dirX/dirY")));
+    Assert.assertFalse("Target File should not exist after delete",
+        fsTarget.exists(new Path(targetTestRoot,"user/dirX/dirY")));
     
-    Assert.assertTrue(fsView.delete(new Path("/user/dirX"), false));
-    Assert.assertFalse(fsView.exists(new Path("/user/dirX")));
+    Assert.assertTrue("Delete should succeed",
+        fsView.delete(new Path("/user/dirX"), false));
+    Assert.assertFalse("File should not exist after delete",
+        fsView.exists(new Path("/user/dirX")));
     Assert.assertFalse(fsTarget.exists(new Path(targetTestRoot,"user/dirX")));
     
     // Rename a file 
     FileSystemTestHelper.createFile(fsView, "/user/foo");
     fsView.rename(new Path("/user/foo"), new Path("/user/fooBar"));
-    Assert.assertFalse(fsView.exists(new Path("/user/foo")));
-    Assert.assertFalse(fsTarget.exists(new Path(targetTestRoot,"user/foo")));
-    Assert.assertTrue(fsView.isFile(FileSystemTestHelper.getTestRootPath(fsView,"/user/fooBar")));
-    Assert.assertTrue(fsTarget.isFile(new Path(targetTestRoot,"user/fooBar")));
+    Assert.assertFalse("Renamed src should not exist", 
+        fsView.exists(new Path("/user/foo")));
+    Assert.assertFalse("Renamed src should not exist in target",
+        fsTarget.exists(new Path(targetTestRoot,"user/foo")));
+    Assert.assertTrue("Renamed dest should  exist as file",
+        fsView.isFile(FileSystemTestHelper.getTestRootPath(fsView,"/user/fooBar")));
+    Assert.assertTrue("Renamed dest should  exist as file in target",
+        fsTarget.isFile(new Path(targetTestRoot,"user/fooBar")));
     
     fsView.mkdirs(new Path("/user/dirFoo"));
     fsView.rename(new Path("/user/dirFoo"), new Path("/user/dirFooBar"));
-    Assert.assertFalse(fsView.exists(new Path("/user/dirFoo")));
-    Assert.assertFalse(fsTarget.exists(new Path(targetTestRoot,"user/dirFoo")));
-    Assert.assertTrue(fsView.isDirectory(FileSystemTestHelper.getTestRootPath(fsView,"/user/dirFooBar")));
-    Assert.assertTrue(fsTarget.isDirectory(new Path(targetTestRoot,"user/dirFooBar")));
+    Assert.assertFalse("Renamed src should not exist", 
+        fsView.exists(new Path("/user/dirFoo")));
+    Assert.assertFalse("Renamed src should not exist in target",
+        fsTarget.exists(new Path(targetTestRoot,"user/dirFoo")));
+    Assert.assertTrue("Renamed dest should  exist as dir",
+        fsView.isDirectory(FileSystemTestHelper.getTestRootPath(fsView,"/user/dirFooBar")));
+    Assert.assertTrue("Renamed dest should  exist as dir in target",
+        fsTarget.isDirectory(new Path(targetTestRoot,"user/dirFooBar")));
     
   }
   
@@ -239,7 +297,8 @@ public class ViewFileSystemBaseTest {
     FileSystemTestHelper.createFile(fsTarget, 
         targetFilePath, 10, 1024);
     Path viewFilePath = new Path("/data/largeFile");
-    Assert.assertTrue(fsView.isFile(viewFilePath));
+    Assert.assertTrue("Created File should be type File",
+        fsView.isFile(viewFilePath));
     BlockLocation[] viewBL = fsView.getFileBlockLocations(fsView.getFileStatus(viewFilePath), 0, 10240+100);
     Assert.assertEquals(SupportsBlocks ? 10 : 1, viewBL.length);
     BlockLocation[] targetBL = fsTarget.getFileBlockLocations(fsTarget.getFileStatus(targetFilePath), 0, 10240+100);
@@ -284,19 +343,19 @@ public class ViewFileSystemBaseTest {
     Assert.assertEquals(6, dirPaths.length);
     fs = FileSystemTestHelper.containsPath(fsView, "/user", dirPaths);
       Assert.assertNotNull(fs);
-      Assert.assertTrue(fs.isSymlink());
+      Assert.assertTrue("A mount should appear as symlink", fs.isSymlink());
     fs = FileSystemTestHelper.containsPath(fsView, "/data", dirPaths);
       Assert.assertNotNull(fs);
-      Assert.assertTrue(fs.isSymlink());
+      Assert.assertTrue("A mount should appear as symlink", fs.isSymlink());
     fs = FileSystemTestHelper.containsPath(fsView, "/internalDir", dirPaths);
       Assert.assertNotNull(fs);
-      Assert.assertTrue(fs.isDirectory());
+      Assert.assertTrue("A mount should appear as symlink", fs.isDirectory());
     fs = FileSystemTestHelper.containsPath(fsView, "/danglingLink", dirPaths);
       Assert.assertNotNull(fs);
-      Assert.assertTrue(fs.isSymlink());
+      Assert.assertTrue("A mount should appear as symlink", fs.isSymlink());
     fs = FileSystemTestHelper.containsPath(fsView, "/linkToAFile", dirPaths);
       Assert.assertNotNull(fs);
-      Assert.assertTrue(fs.isSymlink());
+      Assert.assertTrue("A mount should appear as symlink", fs.isSymlink());
       
       
       
@@ -306,10 +365,11 @@ public class ViewFileSystemBaseTest {
 
       fs = FileSystemTestHelper.containsPath(fsView, "/internalDir/internalDir2", dirPaths);
         Assert.assertNotNull(fs);
-        Assert.assertTrue(fs.isDirectory());
-      fs = FileSystemTestHelper.containsPath(fsView, "/internalDir/linkToDir2", dirPaths);
+        Assert.assertTrue("A mount should appear as symlink", fs.isDirectory());
+      fs = FileSystemTestHelper.containsPath(fsView, "/internalDir/linkToDir2",
+          dirPaths);
         Assert.assertNotNull(fs);
-        Assert.assertTrue(fs.isSymlink());
+        Assert.assertTrue("A mount should appear as symlink", fs.isSymlink());
   }
   
   @Test
@@ -319,12 +379,13 @@ public class ViewFileSystemBaseTest {
     Assert.assertEquals(0, dirPaths.length);
     
     // add a file
-    FileSystemTestHelper.createFile(fsView, "/data/foo");
+    long len = FileSystemTestHelper.createFile(fsView, "/data/foo");
     dirPaths = fsView.listStatus(new Path("/data"));
     Assert.assertEquals(1, dirPaths.length);
     fs = FileSystemTestHelper.containsPath(fsView, "/data/foo", dirPaths);
     Assert.assertNotNull(fs);
-    Assert.assertTrue(fs.isFile());
+    Assert.assertTrue("Created file shoudl appear as a file", fs.isFile());
+    Assert.assertEquals(len, fs.getLen());
     
     // add a dir
     fsView.mkdirs(FileSystemTestHelper.getTestRootPath(fsView, "/data/dirX"));
@@ -332,10 +393,10 @@ public class ViewFileSystemBaseTest {
     Assert.assertEquals(2, dirPaths.length);
     fs = FileSystemTestHelper.containsPath(fsView, "/data/foo", dirPaths);
     Assert.assertNotNull(fs);
-    Assert.assertTrue(fs.isFile());
+    Assert.assertTrue("Created file shoudl appear as a file", fs.isFile());
     fs = FileSystemTestHelper.containsPath(fsView, "/data/dirX", dirPaths);
     Assert.assertNotNull(fs);
-    Assert.assertTrue(fs.isDirectory()); 
+    Assert.assertTrue("Created dir should appear as a dir", fs.isDirectory()); 
   }
       
   @Test
@@ -356,15 +417,69 @@ public class ViewFileSystemBaseTest {
     fsView.getFileStatus(new Path("/danglingLink"));
   }
   
-  
   @Test(expected=FileNotFoundException.class) 
   public void testgetFSonNonExistingInternalDir() throws IOException {
     fsView.getFileStatus(new Path("/internalDir/nonExisting"));
   }
+  
+  /*
+   * Test resolvePath(p) 
+   */
+  
+  @Test
+  public void testResolvePathInternalPaths() throws IOException {
+    Assert.assertEquals(new Path("/"), fsView.resolvePath(new Path("/")));
+    Assert.assertEquals(new Path("/internalDir"),
+                          fsView.resolvePath(new Path("/internalDir")));
+  }
+  @Test
+  public void testResolvePathMountPoints() throws IOException {
+    Assert.assertEquals(new Path(targetTestRoot,"user"),
+                          fsView.resolvePath(new Path("/user")));
+    Assert.assertEquals(new Path(targetTestRoot,"data"),
+        fsView.resolvePath(new Path("/data")));
+    Assert.assertEquals(new Path(targetTestRoot,"dir2"),
+        fsView.resolvePath(new Path("/internalDir/linkToDir2")));
+    Assert.assertEquals(new Path(targetTestRoot,"dir3"),
+        fsView.resolvePath(new Path("/internalDir/internalDir2/linkToDir3")));
 
+  }
   
+  @Test
+  public void testResolvePathThroughMountPoints() throws IOException {
+    FileSystemTestHelper.createFile(fsView, "/user/foo");
+    Assert.assertEquals(new Path(targetTestRoot,"user/foo"),
+                          fsView.resolvePath(new Path("/user/foo")));
+    
+    fsView.mkdirs(
+        FileSystemTestHelper.getTestRootPath(fsView, "/user/dirX"));
+    Assert.assertEquals(new Path(targetTestRoot,"user/dirX"),
+        fsView.resolvePath(new Path("/user/dirX")));
+
+    
+    fsView.mkdirs(
+        FileSystemTestHelper.getTestRootPath(fsView, "/user/dirX/dirY"));
+    Assert.assertEquals(new Path(targetTestRoot,"user/dirX/dirY"),
+        fsView.resolvePath(new Path("/user/dirX/dirY")));
+  }
+
+  @Test(expected=FileNotFoundException.class) 
+  public void testResolvePathDanglingLink() throws IOException {
+      fsView.resolvePath(new Path("/danglingLink"));
+  }
   
+  @Test(expected=FileNotFoundException.class) 
+  public void testResolvePathMissingThroughMountPoints() throws IOException {
+    fsView.resolvePath(new Path("/user/nonExisting"));
+  }
   
+
+  @Test(expected=FileNotFoundException.class) 
+  public void testResolvePathMissingThroughMountPoints2() throws IOException {
+    fsView.mkdirs(
+        FileSystemTestHelper.getTestRootPath(fsView, "/user/dirX"));
+    fsView.resolvePath(new Path("/user/dirX/nonExisting"));
+  }
   
   /**
    * Test modify operations (create, mkdir, rename, etc) 
@@ -375,20 +490,25 @@ public class ViewFileSystemBaseTest {
    */
  
  
-  // Mkdir on internal mount table should fail
+  // Mkdir on existing internal mount table succeed except for /
   @Test(expected=AccessControlException.class) 
   public void testInternalMkdirSlash() throws IOException {
     fsView.mkdirs(FileSystemTestHelper.getTestRootPath(fsView, "/"));
   }
   
-  @Test(expected=AccessControlException.class) 
   public void testInternalMkdirExisting1() throws IOException {
-    fsView.mkdirs(FileSystemTestHelper.getTestRootPath(fsView, "/internalDir"));
+    Assert.assertTrue("mkdir of existing dir should succeed", 
+        fsView.mkdirs(FileSystemTestHelper.getTestRootPath(fsView,
+        "/internalDir")));
   }
-  @Test(expected=AccessControlException.class) 
+
   public void testInternalMkdirExisting2() throws IOException {
-    fsView.mkdirs(FileSystemTestHelper.getTestRootPath(fsView, "/internalDir/linkToDir2"));
+    Assert.assertTrue("mkdir of existing dir should succeed", 
+        fsView.mkdirs(FileSystemTestHelper.getTestRootPath(fsView,
+        "/internalDir/linkToDir2")));
   }
+  
+  // Mkdir for new internal mount table should fail
   @Test(expected=AccessControlException.class) 
   public void testInternalMkdirNew() throws IOException {
     fsView.mkdirs(FileSystemTestHelper.getTestRootPath(fsView, "/dirNew"));
@@ -398,7 +518,7 @@ public class ViewFileSystemBaseTest {
     fsView.mkdirs(FileSystemTestHelper.getTestRootPath(fsView, "/internalDir/dirNew"));
   }
   
-  // Create on internal mount table should fail
+  // Create File on internal mount table should fail
   
   @Test(expected=AccessControlException.class) 
   public void testInternalCreate1() throws IOException {
@@ -442,12 +562,19 @@ public class ViewFileSystemBaseTest {
   }
   @Test(expected=AccessControlException.class) 
   public void testInternalDeleteExisting2() throws IOException {
-    Assert.assertTrue(
         fsView.getFileStatus(
-            new Path("/internalDir/linkToDir2")).isDirectory());
+            new Path("/internalDir/linkToDir2")).isDirectory();
     fsView.delete(new Path("/internalDir/linkToDir2"), false);
   } 
   
+  @Test
+  public void testMkdirOfMountLink() throws IOException {
+    // data exists - mkdirs returns true even though no permission in internal
+    // mount table
+    Assert.assertTrue("mkdir of existing mount link should succeed", 
+        fsView.mkdirs(new Path("/data")));
+  }
+  
   
   // Rename on internal mount table should fail
   
@@ -457,8 +584,7 @@ public class ViewFileSystemBaseTest {
   }
   @Test(expected=AccessControlException.class) 
   public void testInternalRename2() throws IOException {
-    Assert.assertTrue(
-        fsView.getFileStatus(new Path("/internalDir/linkToDir2")).isDirectory());
+       fsView.getFileStatus(new Path("/internalDir/linkToDir2")).isDirectory();
     fsView.rename(new Path("/internalDir/linkToDir2"),
         new Path("/internalDir/dir1"));
   }

src/test/core/org/apache/hadoop/fs/viewfs/ViewFileSystemTestSetup.java

@@ -1,3 +1,20 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package org.apache.hadoop.fs.viewfs;
 
 import org.apache.hadoop.conf.Configuration;
@@ -55,10 +72,10 @@ public class ViewFileSystemTestSetup {
     ConfigUtil.addLink(conf, srcTestFirstDir,
         targetOfTests.toUri());
     
-    FileSystem fcView = FileSystem.get(FsConstants.VIEWFS_URI, conf);
-    //System.out.println("SRCOfTests = "+ getTestRootPath(fc, "test"));
+    FileSystem fsView = FileSystem.get(FsConstants.VIEWFS_URI, conf);
+    //System.out.println("SRCOfTests = "+ getTestRootPath(fs, "test"));
     //System.out.println("TargetOfTests = "+ targetOfTests.toUri());
-    return fcView;
+    return fsView;
   }
 
   /**

src/test/core/org/apache/hadoop/fs/viewfs/ViewFsBaseTest.java

@@ -26,6 +26,7 @@ import static org.apache.hadoop.fs.FileContextTestHelper.isFile;
 import java.io.FileNotFoundException;
 import java.io.IOException;
 import java.net.URI;
+import java.util.List;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.BlockLocation;
@@ -35,7 +36,9 @@ import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FsConstants;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.FileContextTestHelper.fileType;
+import org.apache.hadoop.fs.viewfs.ViewFs.MountPoint;
 import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security.token.Token;
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Before;
@@ -118,6 +121,26 @@ public class ViewFsBaseTest {
     fcTarget.delete(FileContextTestHelper.getTestRootPath(fcTarget), true);
   }
   
+  @Test
+  public void testGetMountPoints() {
+    ViewFs viewfs = (ViewFs) fcView.getDefaultFileSystem();
+    MountPoint[] mountPoints = viewfs.getMountPoints();
+    Assert.assertEquals(7, mountPoints.length); 
+  }
+  
+  /**
+   * This default implementation is when viewfs has mount points
+   * into file systems, such as LocalFs that do no have delegation tokens.
+   * It should be overridden for when mount points into hdfs.
+   */
+  @Test
+  public void testGetDelegationTokens() throws IOException {
+    List<Token<?>> delTokens = 
+        fcView.getDelegationTokens(new Path("/"), "sanjay");
+    Assert.assertEquals(0, delTokens.size()); 
+  }
+
+  
   @Test
   public void testBasicPaths() {
     Assert.assertEquals(FsConstants.VIEWFS_URI,
@@ -146,76 +169,98 @@ public class ViewFsBaseTest {
   public void testOperationsThroughMountLinks() throws IOException {
     // Create file 
     FileContextTestHelper.createFileNonRecursive(fcView, "/user/foo");
-    Assert.assertTrue(isFile(fcView, new Path("/user/foo")));
-    Assert.assertTrue(isFile(fcTarget, new Path(targetTestRoot,"user/foo")));
+    Assert.assertTrue("Create file should be file",
+		isFile(fcView, new Path("/user/foo")));
+    Assert.assertTrue("Target of created file should be type file",
+        isFile(fcTarget, new Path(targetTestRoot,"user/foo")));
     
     // Delete the created file
-    Assert.assertTrue(fcView.delete(new Path("/user/foo"), false));
-    Assert.assertFalse(exists(fcView, new Path("/user/foo")));
-    Assert.assertFalse(exists(fcTarget, new Path(targetTestRoot,"user/foo")));
+    Assert.assertTrue("Delete should succeed",
+        fcView.delete(new Path("/user/foo"), false));
+    Assert.assertFalse("File should not exist after delete",
+        exists(fcView, new Path("/user/foo")));
+    Assert.assertFalse("Target File should not exist after delete",
+        exists(fcTarget, new Path(targetTestRoot,"user/foo")));
     
     // Create file with a 2 component dirs
     FileContextTestHelper.createFileNonRecursive(fcView,
         "/internalDir/linkToDir2/foo");
-    Assert.assertTrue(isFile(fcView, new Path("/internalDir/linkToDir2/foo")));
-    Assert.assertTrue(isFile(fcTarget, new Path(targetTestRoot,"dir2/foo")));
+    Assert.assertTrue("Created file should be type file",
+        isFile(fcView, new Path("/internalDir/linkToDir2/foo")));
+    Assert.assertTrue("Target of created file should be type file",
+        isFile(fcTarget, new Path(targetTestRoot,"dir2/foo")));
     
     // Delete the created file
-    Assert.assertTrue(fcView.delete(new Path("/internalDir/linkToDir2/foo"),false));
-    Assert.assertFalse(exists(fcView, new Path("/internalDir/linkToDir2/foo")));
-    Assert.assertFalse(exists(fcTarget, new Path(targetTestRoot,"dir2/foo")));
+    Assert.assertTrue("Delete should suceed",
+        fcView.delete(new Path("/internalDir/linkToDir2/foo"),false));
+    Assert.assertFalse("File should not exist after deletion",
+        exists(fcView, new Path("/internalDir/linkToDir2/foo")));
+    Assert.assertFalse("Target should not exist after deletion",
+        exists(fcTarget, new Path(targetTestRoot,"dir2/foo")));
     
     
     // Create file with a 3 component dirs
     FileContextTestHelper.createFileNonRecursive(fcView,
         "/internalDir/internalDir2/linkToDir3/foo");
-    Assert.assertTrue(isFile(fcView,
-        new Path("/internalDir/internalDir2/linkToDir3/foo")));
-    Assert.assertTrue(isFile(fcTarget, new Path(targetTestRoot,"dir3/foo")));
+    Assert.assertTrue("Created file should be of type file", 
+        isFile(fcView, new Path("/internalDir/internalDir2/linkToDir3/foo")));
+    Assert.assertTrue("Target of created file should also be type file",
+        isFile(fcTarget, new Path(targetTestRoot,"dir3/foo")));
     
     // Recursive Create file with missing dirs
     FileContextTestHelper.createFile(fcView,
         "/internalDir/linkToDir2/missingDir/miss2/foo");
-    Assert.assertTrue(isFile(fcView,
-        new Path("/internalDir/linkToDir2/missingDir/miss2/foo")));
-    Assert.assertTrue(isFile(fcTarget,
-        new Path(targetTestRoot,"dir2/missingDir/miss2/foo")));
+    Assert.assertTrue("Created file should be of type file",
+      isFile(fcView, new Path("/internalDir/linkToDir2/missingDir/miss2/foo")));
+    Assert.assertTrue("Target of created file should also be type file",
+        isFile(fcTarget, new Path(targetTestRoot,"dir2/missingDir/miss2/foo")));
 
     
     // Delete the created file
-    Assert.assertTrue(fcView.delete(
+    Assert.assertTrue("Delete should succeed",  fcView.delete(
         new Path("/internalDir/internalDir2/linkToDir3/foo"), false));
-    Assert.assertFalse(exists(fcView,
-        new Path("/internalDir/internalDir2/linkToDir3/foo")));
-    Assert.assertFalse(exists(fcTarget, new Path(targetTestRoot,"dir3/foo")));
+    Assert.assertFalse("Deleted File should not exist", 
+        exists(fcView, new Path("/internalDir/internalDir2/linkToDir3/foo")));
+    Assert.assertFalse("Target of deleted file should not exist", 
+        exists(fcTarget, new Path(targetTestRoot,"dir3/foo")));
     
       
     // mkdir
     fcView.mkdir(FileContextTestHelper.getTestRootPath(fcView, "/user/dirX"),
         FileContext.DEFAULT_PERM, false);
-    Assert.assertTrue(isDir(fcView, new Path("/user/dirX")));
-    Assert.assertTrue(isDir(fcTarget, new Path(targetTestRoot,"user/dirX")));
+    Assert.assertTrue("New dir should be type dir", 
+        isDir(fcView, new Path("/user/dirX")));
+    Assert.assertTrue("Target of new dir should be of type dir",
+        isDir(fcTarget, new Path(targetTestRoot,"user/dirX")));
     
     fcView.mkdir(FileContextTestHelper.getTestRootPath(fcView, "/user/dirX/dirY"),
         FileContext.DEFAULT_PERM, false);
-    Assert.assertTrue(isDir(fcView, new Path("/user/dirX/dirY")));
-    Assert.assertTrue(isDir(fcTarget,new Path(targetTestRoot,"user/dirX/dirY")));
+    Assert.assertTrue("New dir should be type dir", 
+        isDir(fcView, new Path("/user/dirX/dirY")));
+    Assert.assertTrue("Target of new dir should be of type dir",
+        isDir(fcTarget,new Path(targetTestRoot,"user/dirX/dirY")));
     
 
     // Delete the created dir
-    Assert.assertTrue(fcView.delete(new Path("/user/dirX/dirY"), false));
-    Assert.assertFalse(exists(fcView, new Path("/user/dirX/dirY")));
-    Assert.assertFalse(exists(fcTarget,
-        new Path(targetTestRoot,"user/dirX/dirY")));
+    Assert.assertTrue("Delete should succeed",
+        fcView.delete(new Path("/user/dirX/dirY"), false));
+    Assert.assertFalse("Deleted File should not exist",
+        exists(fcView, new Path("/user/dirX/dirY")));
+    Assert.assertFalse("Deleted Target should not exist", 
+        exists(fcTarget, new Path(targetTestRoot,"user/dirX/dirY")));
     
-    Assert.assertTrue(fcView.delete(new Path("/user/dirX"), false));
-    Assert.assertFalse(exists(fcView, new Path("/user/dirX")));
-    Assert.assertFalse(exists(fcTarget, new Path(targetTestRoot,"user/dirX")));
+    Assert.assertTrue("Delete should succeed",
+        fcView.delete(new Path("/user/dirX"), false));
+    Assert.assertFalse("Deleted File should not exist",
+        exists(fcView, new Path("/user/dirX")));
+    Assert.assertFalse("Deleted Target should not exist",
+        exists(fcTarget, new Path(targetTestRoot,"user/dirX")));
     
     // Rename a file 
     FileContextTestHelper.createFile(fcView, "/user/foo");
     fcView.rename(new Path("/user/foo"), new Path("/user/fooBar"));
-    Assert.assertFalse(exists(fcView, new Path("/user/foo")));
+    Assert.assertFalse("Renamed src should not exist", 
+        exists(fcView, new Path("/user/foo")));
     Assert.assertFalse(exists(fcTarget, new Path(targetTestRoot,"user/foo")));
     Assert.assertTrue(isFile(fcView,
         FileContextTestHelper.getTestRootPath(fcView,"/user/fooBar")));
@@ -223,11 +268,15 @@ public class ViewFsBaseTest {
     
     fcView.mkdir(new Path("/user/dirFoo"), FileContext.DEFAULT_PERM, false);
     fcView.rename(new Path("/user/dirFoo"), new Path("/user/dirFooBar"));
-    Assert.assertFalse(exists(fcView, new Path("/user/dirFoo")));
-    Assert.assertFalse(exists(fcTarget, new Path(targetTestRoot,"user/dirFoo")));
-    Assert.assertTrue(isDir(fcView,
+    Assert.assertFalse("Renamed src should not exist",
+        exists(fcView, new Path("/user/dirFoo")));
+    Assert.assertFalse("Renamed src should not exist in target",
+        exists(fcTarget, new Path(targetTestRoot,"user/dirFoo")));
+    Assert.assertTrue("Renamed dest should  exist as dir",
+        isDir(fcView,
         FileContextTestHelper.getTestRootPath(fcView,"/user/dirFooBar")));
-    Assert.assertTrue(isDir(fcTarget,new Path(targetTestRoot,"user/dirFooBar")));
+    Assert.assertTrue("Renamed dest should  exist as dir in target",
+        isDir(fcTarget,new Path(targetTestRoot,"user/dirFooBar")));
     
   }
   
@@ -236,7 +285,7 @@ public class ViewFsBaseTest {
   public void testRenameAcrossMounts1() throws IOException {
     FileContextTestHelper.createFile(fcView, "/user/foo");
     fcView.rename(new Path("/user/foo"), new Path("/user2/fooBarBar"));
-    /* - code if we had wanted this to suceed
+    /* - code if we had wanted this to succeed
     Assert.assertFalse(exists(fc, new Path("/user/foo")));
     Assert.assertFalse(exists(fclocal, new Path(targetTestRoot,"user/foo")));
     Assert.assertTrue(isFile(fc,
@@ -309,19 +358,19 @@ public class ViewFsBaseTest {
     Assert.assertEquals(6, dirPaths.length);
     fs = FileContextTestHelper.containsPath(fcView, "/user", dirPaths);
       Assert.assertNotNull(fs);
-      Assert.assertTrue(fs.isSymlink());
+      Assert.assertTrue("A mount should appear as symlink", fs.isSymlink());
     fs = FileContextTestHelper.containsPath(fcView, "/data", dirPaths);
       Assert.assertNotNull(fs);
-      Assert.assertTrue(fs.isSymlink());
+      Assert.assertTrue("A mount should appear as symlink", fs.isSymlink());
     fs = FileContextTestHelper.containsPath(fcView, "/internalDir", dirPaths);
       Assert.assertNotNull(fs);
-      Assert.assertTrue(fs.isDirectory());
+      Assert.assertTrue("InternalDirs should appear as dir", fs.isDirectory());
     fs = FileContextTestHelper.containsPath(fcView, "/danglingLink", dirPaths);
       Assert.assertNotNull(fs);
-      Assert.assertTrue(fs.isSymlink());
+      Assert.assertTrue("A mount should appear as symlink", fs.isSymlink());
     fs = FileContextTestHelper.containsPath(fcView, "/linkToAFile", dirPaths);
       Assert.assertNotNull(fs);
-      Assert.assertTrue(fs.isSymlink());
+      Assert.assertTrue("A mount should appear as symlink", fs.isSymlink());
       
       
       
@@ -332,16 +381,17 @@ public class ViewFsBaseTest {
       fs = FileContextTestHelper.containsPath(fcView,
           "/internalDir/internalDir2", dirPaths);
         Assert.assertNotNull(fs);
-        Assert.assertTrue(fs.isDirectory());
+        Assert.assertTrue("InternalDirs should appear as dir",fs.isDirectory());
       fs = FileContextTestHelper.containsPath(fcView,
           "/internalDir/linkToDir2", dirPaths);
         Assert.assertNotNull(fs);
-        Assert.assertTrue(fs.isSymlink());
+        Assert.assertTrue("A mount should appear as symlink", fs.isSymlink());
   }
       
   @Test
   public void testFileStatusOnMountLink() throws IOException {
-    Assert.assertTrue(fcView.getFileStatus(new Path("/")).isDirectory());
+    Assert.assertTrue("Slash should appear as dir", 
+        fcView.getFileStatus(new Path("/")).isDirectory());
     checkFileStatus(fcView, "/", fileType.isDir);
     checkFileStatus(fcView, "/user", fileType.isDir);
     checkFileStatus(fcView, "/data", fileType.isDir);
@@ -391,17 +441,18 @@ public class ViewFsBaseTest {
   public void testSymlinkTarget() throws IOException {
 
     // get link target`
-    Assert.assertTrue(fcView.getLinkTarget(new Path("/user"
-      )).equals(new Path(targetTestRoot,"user")));
-    Assert.assertTrue(fcView.getLinkTarget(new Path("/data"
-      )).equals(new Path(targetTestRoot,"data")));
-    Assert.assertTrue(fcView.getLinkTarget(new Path("/internalDir/linkToDir2"
-      )).equals(new Path(targetTestRoot,"dir2")));
-    Assert.assertTrue(fcView.getLinkTarget(new Path(
-     "/internalDir/internalDir2/linkToDir3"
-      )).equals(new Path(targetTestRoot,"dir3")));
-    Assert.assertTrue(fcView.getLinkTarget(new Path("/linkToAFile"
-      )).equals(new Path(targetTestRoot,"aFile")));
+    Assert.assertEquals(fcView.getLinkTarget(new Path("/user")),
+        (new Path(targetTestRoot,"user")));
+    Assert.assertEquals(fcView.getLinkTarget(new Path("/data")),
+        (new Path(targetTestRoot,"data")));
+    Assert.assertEquals(
+        fcView.getLinkTarget(new Path("/internalDir/linkToDir2")),
+        (new Path(targetTestRoot,"dir2")));
+    Assert.assertEquals(
+        fcView.getLinkTarget(new Path("/internalDir/internalDir2/linkToDir3")),
+        (new Path(targetTestRoot,"dir3")));
+    Assert.assertEquals(fcView.getLinkTarget(new Path("/linkToAFile")),
+        (new Path(targetTestRoot,"aFile")));
   }
   
   @Test(expected=IOException.class) 
@@ -409,7 +460,69 @@ public class ViewFsBaseTest {
     fcView.getLinkTarget(new Path("/internalDir/internalDir2"));
   }
   
+  /*
+   * Test resolvePath(p) 
+   * TODO In the tests below replace 
+   * fcView.getDefaultFileSystem().resolvePath() fcView.resolvePath()
+   */
+  
+  @Test
+  public void testResolvePathInternalPaths() throws IOException {
+    Assert.assertEquals(new Path("/"), fcView.getDefaultFileSystem().resolvePath(new Path("/")));
+    Assert.assertEquals(new Path("/internalDir"),
+                          fcView.getDefaultFileSystem().resolvePath(new Path("/internalDir")));
+  }
+  @Test
+  public void testResolvePathMountPoints() throws IOException {
+    Assert.assertEquals(new Path(targetTestRoot,"user"),
+                          fcView.getDefaultFileSystem().resolvePath(new Path("/user")));
+    Assert.assertEquals(new Path(targetTestRoot,"data"),
+        fcView.getDefaultFileSystem().resolvePath(new Path("/data")));
+    Assert.assertEquals(new Path(targetTestRoot,"dir2"),
+        fcView.getDefaultFileSystem().resolvePath(new Path("/internalDir/linkToDir2")));
+    Assert.assertEquals(new Path(targetTestRoot,"dir3"),
+        fcView.getDefaultFileSystem().resolvePath(new Path("/internalDir/internalDir2/linkToDir3")));
+
+  }
+  
+  @Test
+  public void testResolvePathThroughMountPoints() throws IOException {
+    FileContextTestHelper.createFile(fcView, "/user/foo");
+    Assert.assertEquals(new Path(targetTestRoot,"user/foo"),
+                          fcView.getDefaultFileSystem().resolvePath(new Path("/user/foo")));
+    
+    fcView.mkdir(
+        FileContextTestHelper.getTestRootPath(fcView, "/user/dirX"),
+        FileContext.DEFAULT_PERM, false);
+    Assert.assertEquals(new Path(targetTestRoot,"user/dirX"),
+        fcView.getDefaultFileSystem().resolvePath(new Path("/user/dirX")));
+
+    
+    fcView.mkdir(
+        FileContextTestHelper.getTestRootPath(fcView, "/user/dirX/dirY"),
+        FileContext.DEFAULT_PERM, false);
+    Assert.assertEquals(new Path(targetTestRoot,"user/dirX/dirY"),
+        fcView.getDefaultFileSystem().resolvePath(new Path("/user/dirX/dirY")));
+  }
+
+  @Test(expected=FileNotFoundException.class) 
+  public void testResolvePathDanglingLink() throws IOException {
+      fcView.getDefaultFileSystem().resolvePath(new Path("/danglingLink"));
+  }
+  
+  @Test(expected=FileNotFoundException.class) 
+  public void testResolvePathMissingThroughMountPoints() throws IOException {
+    fcView.getDefaultFileSystem().resolvePath(new Path("/user/nonExisting"));
+  }
   
+
+  @Test(expected=FileNotFoundException.class) 
+  public void testResolvePathMissingThroughMountPoints2() throws IOException {
+    fcView.mkdir(
+        FileContextTestHelper.getTestRootPath(fcView, "/user/dirX"),
+        FileContext.DEFAULT_PERM, false);
+    fcView.getDefaultFileSystem().resolvePath(new Path("/user/dirX/nonExisting"));
+  }
   
   
   /**
@@ -494,7 +607,7 @@ public class ViewFsBaseTest {
   }
   @Test(expected=AccessControlException.class) 
   public void testInternalDeleteExisting2() throws IOException {
-    Assert.assertTrue(
+    Assert.assertTrue("Delete of link to dir should succeed",
         fcView.getFileStatus(new Path("/internalDir/linkToDir2")).isDirectory());
     fcView.delete(new Path("/internalDir/linkToDir2"), false);
   } 
@@ -508,7 +621,7 @@ public class ViewFsBaseTest {
   }
   @Test(expected=AccessControlException.class) 
   public void testInternalRename2() throws IOException {
-    Assert.assertTrue(
+    Assert.assertTrue("linkTODir2 should be a dir", 
         fcView.getFileStatus(new Path("/internalDir/linkToDir2")).isDirectory());
     fcView.rename(new Path("/internalDir/linkToDir2"),
         new Path("/internalDir/dir1"));

src/test/core/org/apache/hadoop/fs/viewfs/ViewFsTestSetup.java

@@ -1,3 +1,20 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
 package org.apache.hadoop.fs.viewfs;
 
 import org.apache.hadoop.conf.Configuration;