diff --git a/CHANGES.txt b/CHANGES.txt
index dd986ab63d6..b5b25a91920 100644
--- a/CHANGES.txt
+++ b/CHANGES.txt
@@ -171,6 +171,10 @@ Trunk (unreleased changes)
     HADOOP-7024. Create a test method for adding file systems during tests.
     (Kan Zhang via jghoman)
 
+    HADOOP-6903 Make AbstractFSileSystem methods and some FileContext methods to be public
+    (Sanjay Radia via Sanjay Radia)
+
+
   OPTIMIZATIONS
 
     HADOOP-6884. Add LOG.isDebugEnabled() guard for each LOG.debug(..).
diff --git a/src/java/org/apache/hadoop/fs/AbstractFileSystem.java b/src/java/org/apache/hadoop/fs/AbstractFileSystem.java
index 946455b843c..2041ad05a27 100644
--- a/src/java/org/apache/hadoop/fs/AbstractFileSystem.java
+++ b/src/java/org/apache/hadoop/fs/AbstractFileSystem.java
@@ -25,7 +25,6 @@
 import java.net.URISyntaxException;
 import java.util.EnumSet;
 import java.util.IdentityHashMap;
-import java.util.Iterator;
 import java.util.Map;
 import java.util.NoSuchElementException;
 import java.util.StringTokenizer;
@@ -77,7 +76,7 @@ public abstract class AbstractFileSystem {
   
   private final URI myUri;
   
-  protected Statistics getStatistics() {
+  public Statistics getStatistics() {
     return statistics;
   }
   
@@ -135,7 +134,7 @@ static <T> T newInstance(Class<T> theClass,
    * @throws UnsupportedFileSystemException file system for <code>uri</code> is
    *           not found
    */
-  private static AbstractFileSystem createFileSystem(URI uri, Configuration conf)
+  public static AbstractFileSystem createFileSystem(URI uri, Configuration conf)
       throws UnsupportedFileSystemException {
     Class<?> clazz = conf.getClass("fs.AbstractFileSystem." + 
                                 uri.getScheme() + ".impl", null);
@@ -152,7 +151,7 @@ private static AbstractFileSystem createFileSystem(URI uri, Configuration conf)
    * @param cls the class to lookup
    * @return a statistics object
    */
-  protected static synchronized Statistics getStatistics(String scheme,
+  public static synchronized Statistics getStatistics(String scheme,
       Class<? extends AbstractFileSystem> cls) {
     Statistics result = STATISTICS_TABLE.get(cls);
     if (result == null) {
@@ -162,13 +161,13 @@ protected static synchronized Statistics getStatistics(String scheme,
     return result;
   }
   
-  protected static synchronized void clearStatistics() {
+  public static synchronized void clearStatistics() {
     for(Statistics stat: STATISTICS_TABLE.values()) {
       stat.reset();
     }
   }
 
-  protected static synchronized void printStatistics() {
+  public static synchronized void printStatistics() {
     for (Map.Entry<Class<? extends AbstractFileSystem>, Statistics> pair: 
             STATISTICS_TABLE.entrySet()) {
       System.out.println("  FileSystem " + pair.getKey().getName() + 
@@ -193,7 +192,7 @@ protected static synchronized void printStatistics() {
    * @throws UnsupportedFileSystemException if the file system for
    *           <code>uri</code> is not supported.
    */
-  static AbstractFileSystem get(final URI uri, final Configuration conf)
+  public static AbstractFileSystem get(final URI uri, final Configuration conf)
       throws UnsupportedFileSystemException {
     return createFileSystem(uri, conf);
   }
@@ -208,14 +207,19 @@ static AbstractFileSystem get(final URI uri, final Configuration conf)
    *
    * @throws URISyntaxException <code>uri</code> has syntax error
    */
-  protected AbstractFileSystem(final URI uri, final String supportedScheme,
+  public AbstractFileSystem(final URI uri, final String supportedScheme,
       final boolean authorityNeeded, final int defaultPort)
       throws URISyntaxException {
     myUri = getUri(uri, supportedScheme, authorityNeeded, defaultPort);
     statistics = getStatistics(supportedScheme, getClass()); 
   }
   
-  protected void checkScheme(URI uri, String supportedScheme) {
+  /**
+   * Check that the Uri's scheme matches
+   * @param uri
+   * @param supportedScheme
+   */
+  public void checkScheme(URI uri, String supportedScheme) {
     String scheme = uri.getScheme();
     if (scheme == null) {
       throw new HadoopIllegalArgumentException("Uri without scheme: " + uri);
@@ -272,14 +276,14 @@ private URI getUri(URI uri, String supportedScheme,
    * @return default port of this file system's Uri scheme
    *         A uri with a port of -1 => default port;
    */
-  protected abstract int getUriDefaultPort();
+  public abstract int getUriDefaultPort();
 
   /**
    * Returns a URI whose scheme and authority identify this FileSystem.
    * 
    * @return the uri of this file system.
    */
-  protected URI getUri() {
+  public URI getUri() {
     return myUri;
   }
   
@@ -292,7 +296,7 @@ protected URI getUri() {
    * 
    * @throws InvalidPathException if the path is invalid
    */
-  protected void checkPath(Path path) {
+  public void checkPath(Path path) {
     URI uri = path.toUri();
     String thatScheme = uri.getScheme();
     String thatAuthority = uri.getAuthority();
@@ -340,7 +344,7 @@ protected void checkPath(Path path) {
    * 
    * @return path-part of the Path p
    */
-  protected String getUriPath(final Path p) {
+  public String getUriPath(final Path p) {
     checkPath(p);
     String s = p.toUri().getPath();
     if (!isValidName(s)) {
@@ -350,6 +354,16 @@ protected String getUriPath(final Path p) {
     return s;
   }
   
+  /**
+   * Make the path fully qualified to this file system
+   * @param path
+   * @return the qualified path
+   */
+  public Path makeQualified(Path path) {
+    checkPath(path);
+    return path.makeQualified(this.getUri(), null);
+  }
+  
   /**
    * Some file systems like LocalFileSystem have an initial workingDir
    * that is used as the starting workingDir. For other file systems
@@ -358,7 +372,7 @@ protected String getUriPath(final Path p) {
    * @return the initial workingDir if the file system has such a notion
    *         otherwise return a null.
    */
-  protected Path getInitialWorkingDirectory() {
+  public Path getInitialWorkingDirectory() {
     return null;
   }
   
@@ -368,7 +382,7 @@ protected Path getInitialWorkingDirectory() {
    * 
    * @return current user's home directory.
    */
-  protected Path getHomeDirectory() {
+  public Path getHomeDirectory() {
     return new Path("/user/"+System.getProperty("user.name")).makeQualified(
                                                                 getUri(), null);
   }
@@ -380,7 +394,7 @@ protected Path getHomeDirectory() {
    * 
    * @throws IOException an I/O error occurred
    */
-  protected abstract FsServerDefaults getServerDefaults() throws IOException; 
+  public abstract FsServerDefaults getServerDefaults() throws IOException; 
 
   /**
    * The specification of this method matches that of
@@ -388,7 +402,7 @@ protected Path getHomeDirectory() {
    * that the Path f must be fully qualified and the permission is absolute
    * (i.e. umask has been applied).
    */
-  protected final FSDataOutputStream create(final Path f,
+  public final FSDataOutputStream create(final Path f,
       final EnumSet<CreateFlag> createFlag, Options.CreateOpts... opts)
       throws AccessControlException, FileAlreadyExistsException,
       FileNotFoundException, ParentNotDirectoryException,
@@ -491,7 +505,7 @@ protected final FSDataOutputStream create(final Path f,
    * {@link #create(Path, EnumSet, Options.CreateOpts...)} except that the opts
    * have been declared explicitly.
    */
-  protected abstract FSDataOutputStream createInternal(Path f,
+  public abstract FSDataOutputStream createInternal(Path f,
       EnumSet<CreateFlag> flag, FsPermission absolutePermission,
       int bufferSize, short replication, long blockSize, Progressable progress,
       int bytesPerChecksum, boolean createParent)
@@ -505,7 +519,7 @@ protected abstract FSDataOutputStream createInternal(Path f,
    * f must be fully qualified and the permission is absolute (i.e. 
    * umask has been applied).
    */
-  protected abstract void mkdir(final Path dir, final FsPermission permission,
+  public abstract void mkdir(final Path dir, final FsPermission permission,
       final boolean createParent) throws AccessControlException,
       FileAlreadyExistsException, FileNotFoundException,
       UnresolvedLinkException, IOException;
@@ -515,7 +529,7 @@ protected abstract void mkdir(final Path dir, final FsPermission permission,
    * {@link FileContext#delete(Path, boolean)} except that Path f must be for
    * this file system.
    */
-  protected abstract boolean delete(final Path f, final boolean recursive)
+  public abstract boolean delete(final Path f, final boolean recursive)
       throws AccessControlException, FileNotFoundException,
       UnresolvedLinkException, IOException;
 
@@ -524,7 +538,7 @@ protected abstract boolean delete(final Path f, final boolean recursive)
    * {@link FileContext#open(Path)} except that Path f must be for this
    * file system.
    */
-  protected FSDataInputStream open(final Path f) throws AccessControlException,
+  public FSDataInputStream open(final Path f) throws AccessControlException,
       FileNotFoundException, UnresolvedLinkException, IOException {
     return open(f, getServerDefaults().getFileBufferSize());
   }
@@ -534,7 +548,7 @@ protected FSDataInputStream open(final Path f) throws AccessControlException,
    * {@link FileContext#open(Path, int)} except that Path f must be for this
    * file system.
    */
-  protected abstract FSDataInputStream open(final Path f, int bufferSize)
+  public abstract FSDataInputStream open(final Path f, int bufferSize)
       throws AccessControlException, FileNotFoundException,
       UnresolvedLinkException, IOException;
 
@@ -543,7 +557,7 @@ protected abstract FSDataInputStream open(final Path f, int bufferSize)
    * {@link FileContext#setReplication(Path, short)} except that Path f must be
    * for this file system.
    */
-  protected abstract boolean setReplication(final Path f,
+  public abstract boolean setReplication(final Path f,
       final short replication) throws AccessControlException,
       FileNotFoundException, UnresolvedLinkException, IOException;
 
@@ -552,7 +566,7 @@ protected abstract boolean setReplication(final Path f,
    * {@link FileContext#rename(Path, Path, Options.Rename...)} except that Path
    * f must be for this file system.
    */
-  protected final void rename(final Path src, final Path dst,
+  public final void rename(final Path src, final Path dst,
       final Options.Rename... options) throws AccessControlException,
       FileAlreadyExistsException, FileNotFoundException,
       ParentNotDirectoryException, UnresolvedLinkException, IOException {
@@ -576,7 +590,7 @@ protected final void rename(final Path src, final Path dst,
    * method and can take advantage of the default impl of the other
    * {@link #renameInternal(Path, Path, boolean)}
    */
-  protected abstract void renameInternal(final Path src, final Path dst)
+  public abstract void renameInternal(final Path src, final Path dst)
       throws AccessControlException, FileAlreadyExistsException,
       FileNotFoundException, ParentNotDirectoryException,
       UnresolvedLinkException, IOException;
@@ -586,7 +600,7 @@ protected abstract void renameInternal(final Path src, final Path dst)
    * {@link FileContext#rename(Path, Path, Options.Rename...)} except that Path
    * f must be for this file system.
    */
-  protected void renameInternal(final Path src, final Path dst,
+  public void renameInternal(final Path src, final Path dst,
       boolean overwrite) throws AccessControlException,
       FileAlreadyExistsException, FileNotFoundException,
       ParentNotDirectoryException, UnresolvedLinkException, IOException {
@@ -640,7 +654,7 @@ protected void renameInternal(final Path src, final Path dst,
   /**
    * Returns true if the file system supports symlinks, false otherwise.
    */
-  protected boolean supportsSymlinks() {
+  public boolean supportsSymlinks() {
     return false;
   }
   
@@ -648,7 +662,7 @@ protected boolean supportsSymlinks() {
    * The specification of this method matches that of  
    * {@link FileContext#createSymlink(Path, Path, boolean)};
    */
-  protected void createSymlink(final Path target, final Path link,
+  public void createSymlink(final Path target, final Path link,
       final boolean createParent) throws IOException, UnresolvedLinkException {
     throw new IOException("File system does not support symlinks");    
   }
@@ -657,7 +671,7 @@ protected void createSymlink(final Path target, final Path link,
    * The specification of this method matches that of  
    * {@link FileContext#getLinkTarget(Path)};
    */
-  protected Path getLinkTarget(final Path f) throws IOException {
+  public Path getLinkTarget(final Path f) throws IOException {
     /* We should never get here. Any file system that threw an
      * UnresolvedLinkException, causing this function to be called,
      * needs to override this method.
@@ -670,7 +684,7 @@ protected Path getLinkTarget(final Path f) throws IOException {
    * {@link FileContext#setPermission(Path, FsPermission)} except that Path f
    * must be for this file system.
    */
-  protected abstract void setPermission(final Path f,
+  public abstract void setPermission(final Path f,
       final FsPermission permission) throws AccessControlException,
       FileNotFoundException, UnresolvedLinkException, IOException;
 
@@ -679,7 +693,7 @@ protected abstract void setPermission(final Path f,
    * {@link FileContext#setOwner(Path, String, String)} except that Path f must
    * be for this file system.
    */
-  protected abstract void setOwner(final Path f, final String username,
+  public abstract void setOwner(final Path f, final String username,
       final String groupname) throws AccessControlException,
       FileNotFoundException, UnresolvedLinkException, IOException;
 
@@ -688,7 +702,7 @@ protected abstract void setOwner(final Path f, final String username,
    * {@link FileContext#setTimes(Path, long, long)} except that Path f must be
    * for this file system.
    */
-  protected abstract void setTimes(final Path f, final long mtime,
+  public abstract void setTimes(final Path f, final long mtime,
     final long atime) throws AccessControlException, FileNotFoundException,
       UnresolvedLinkException, IOException;
 
@@ -697,7 +711,7 @@ protected abstract void setTimes(final Path f, final long mtime,
    * {@link FileContext#getFileChecksum(Path)} except that Path f must be for
    * this file system.
    */
-  protected abstract FileChecksum getFileChecksum(final Path f)
+  public abstract FileChecksum getFileChecksum(final Path f)
       throws AccessControlException, FileNotFoundException,
       UnresolvedLinkException, IOException;
   
@@ -707,7 +721,7 @@ protected abstract FileChecksum getFileChecksum(final Path f)
    * except that an UnresolvedLinkException may be thrown if a symlink is 
    * encountered in the path.
    */
-  protected abstract FileStatus getFileStatus(final Path f)
+  public abstract FileStatus getFileStatus(final Path f)
       throws AccessControlException, FileNotFoundException,
       UnresolvedLinkException, IOException;
 
@@ -719,7 +733,7 @@ protected abstract FileStatus getFileStatus(final Path f)
    * If the file system does not support symlinks then the behavior is
    * equivalent to {@link AbstractFileSystem#getFileStatus(Path)}.
    */
-  protected FileStatus getFileLinkStatus(final Path f)
+  public FileStatus getFileLinkStatus(final Path f)
       throws AccessControlException, FileNotFoundException,
       UnsupportedFileSystemException, IOException {
     return getFileStatus(f);
@@ -730,7 +744,7 @@ protected FileStatus getFileLinkStatus(final Path f)
    * {@link FileContext#getFileBlockLocations(Path, long, long)} except that
    * Path f must be for this file system.
    */
-  protected abstract BlockLocation[] getFileBlockLocations(final Path f,
+  public abstract BlockLocation[] getFileBlockLocations(final Path f,
       final long start, final long len) throws AccessControlException,
       FileNotFoundException, UnresolvedLinkException, IOException;
 
@@ -739,7 +753,7 @@ protected abstract BlockLocation[] getFileBlockLocations(final Path f,
    * {@link FileContext#getFsStatus(Path)} except that Path f must be for this
    * file system.
    */
-  protected FsStatus getFsStatus(final Path f) throws AccessControlException,
+  public FsStatus getFsStatus(final Path f) throws AccessControlException,
       FileNotFoundException, UnresolvedLinkException, IOException {
     // default impl gets FsStatus of root
     return getFsStatus();
@@ -749,7 +763,7 @@ protected FsStatus getFsStatus(final Path f) throws AccessControlException,
    * The specification of this method matches that of
    * {@link FileContext#getFsStatus(Path)}.
    */
-  protected abstract FsStatus getFsStatus() throws AccessControlException,
+  public abstract FsStatus getFsStatus() throws AccessControlException,
       FileNotFoundException, IOException;
 
   /**
@@ -757,7 +771,7 @@ protected abstract FsStatus getFsStatus() throws AccessControlException,
    * {@link FileContext#listStatus(Path)} except that Path f must be for this
    * file system.
    */
-  protected RemoteIterator<FileStatus> listStatusIterator(final Path f)
+  public RemoteIterator<FileStatus> listStatusIterator(final Path f)
       throws AccessControlException, FileNotFoundException,
       UnresolvedLinkException, IOException {
     return new RemoteIterator<FileStatus>() {
@@ -784,7 +798,7 @@ public FileStatus next() {
    * {@link FileContext#listLocatedStatus(Path)} except that Path f 
    * must be for this file system.
    */
-  protected RemoteIterator<LocatedFileStatus> listLocatedStatus(final Path f)
+  public RemoteIterator<LocatedFileStatus> listLocatedStatus(final Path f)
       throws AccessControlException, FileNotFoundException,
       UnresolvedLinkException, IOException {
     return new RemoteIterator<LocatedFileStatus>() {
@@ -816,7 +830,7 @@ public LocatedFileStatus next() throws IOException {
    * {@link FileContext.Util#listStatus(Path)} except that Path f must be 
    * for this file system.
    */
-  protected abstract FileStatus[] listStatus(final Path f)
+  public abstract FileStatus[] listStatus(final Path f)
       throws AccessControlException, FileNotFoundException,
       UnresolvedLinkException, IOException;
 
@@ -825,6 +839,6 @@ protected abstract FileStatus[] listStatus(final Path f)
    * {@link FileContext#setVerifyChecksum(boolean, Path)} except that Path f
    * must be for this file system.
    */
-  protected abstract void setVerifyChecksum(final boolean verifyChecksum)
+  public abstract void setVerifyChecksum(final boolean verifyChecksum)
       throws AccessControlException, IOException;
 }
diff --git a/src/java/org/apache/hadoop/fs/ChecksumFs.java b/src/java/org/apache/hadoop/fs/ChecksumFs.java
index 89f386611cf..a55108598be 100644
--- a/src/java/org/apache/hadoop/fs/ChecksumFs.java
+++ b/src/java/org/apache/hadoop/fs/ChecksumFs.java
@@ -358,7 +358,7 @@ protected void writeChunk(byte[] b, int offset, int len, byte[] checksum)
   }
 
   @Override
-  protected FSDataOutputStream createInternal(Path f,
+  public FSDataOutputStream createInternal(Path f,
       EnumSet<CreateFlag> createFlag, FsPermission absolutePermission,
       int bufferSize, short replication, long blockSize, Progressable progress,
       int bytesPerChecksum, boolean createParent) throws IOException {
@@ -481,7 +481,7 @@ public boolean reportChecksumFailure(Path f, FSDataInputStream in,
   }
 
   @Override
-  protected FileStatus[] listStatus(Path f) throws IOException,
+  public FileStatus[] listStatus(Path f) throws IOException,
       UnresolvedLinkException {
     ArrayList<FileStatus> results = new ArrayList<FileStatus>();
     FileStatus[] listing = getMyFs().listStatus(f);
diff --git a/src/java/org/apache/hadoop/fs/DelegateToFileSystem.java b/src/java/org/apache/hadoop/fs/DelegateToFileSystem.java
index 64d4e04797c..14da6e0ad32 100644
--- a/src/java/org/apache/hadoop/fs/DelegateToFileSystem.java
+++ b/src/java/org/apache/hadoop/fs/DelegateToFileSystem.java
@@ -50,13 +50,13 @@ protected DelegateToFileSystem(URI theUri, FileSystem theFsImpl,
   }
 
   @Override
-  protected Path getInitialWorkingDirectory() {
+  public Path getInitialWorkingDirectory() {
     return fsImpl.getInitialWorkingDirectory();
   }
   
   @Override
   @SuppressWarnings("deprecation") // call to primitiveCreate
-  protected FSDataOutputStream createInternal (Path f,
+  public FSDataOutputStream createInternal (Path f,
       EnumSet<CreateFlag> flag, FsPermission absolutePermission, int bufferSize,
       short replication, long blockSize, Progressable progress,
       int bytesPerChecksum, boolean createParent) throws IOException {
@@ -83,59 +83,59 @@ protected FSDataOutputStream createInternal (Path f,
   }
 
   @Override
-  protected boolean delete(Path f, boolean recursive) throws IOException {
+  public boolean delete(Path f, boolean recursive) throws IOException {
     checkPath(f);
     return fsImpl.delete(f, recursive);
   }
 
   @Override
-  protected BlockLocation[] getFileBlockLocations(Path f, long start, long len)
+  public BlockLocation[] getFileBlockLocations(Path f, long start, long len)
       throws IOException {
     checkPath(f);
     return fsImpl.getFileBlockLocations(f, start, len);
   }
 
   @Override
-  protected FileChecksum getFileChecksum(Path f) throws IOException {
+  public FileChecksum getFileChecksum(Path f) throws IOException {
     checkPath(f);
     return fsImpl.getFileChecksum(f);
   }
 
   @Override
-  protected FileStatus getFileStatus(Path f) throws IOException {
+  public FileStatus getFileStatus(Path f) throws IOException {
     checkPath(f);
     return fsImpl.getFileStatus(f);
   }
 
   @Override
-  protected FileStatus getFileLinkStatus(final Path f) throws IOException {
+  public FileStatus getFileLinkStatus(final Path f) throws IOException {
     return getFileStatus(f);
   }
 
   @Override
-  protected FsStatus getFsStatus() throws IOException {
+  public FsStatus getFsStatus() throws IOException {
     return fsImpl.getStatus();
   }
 
   @Override
-  protected FsServerDefaults getServerDefaults() throws IOException {
+  public FsServerDefaults getServerDefaults() throws IOException {
     return fsImpl.getServerDefaults();
   }
 
   @Override
-  protected int getUriDefaultPort() {
+  public int getUriDefaultPort() {
     return 0;
   }
 
   @Override
-  protected FileStatus[] listStatus(Path f) throws IOException {
+  public FileStatus[] listStatus(Path f) throws IOException {
     checkPath(f);
     return fsImpl.listStatus(f);
   }
 
   @Override
   @SuppressWarnings("deprecation") // call to primitiveMkdir
-  protected void mkdir(Path dir, FsPermission permission, boolean createParent)
+  public void mkdir(Path dir, FsPermission permission, boolean createParent)
       throws IOException {
     checkPath(dir);
     fsImpl.primitiveMkdir(dir, permission, createParent);
@@ -143,64 +143,64 @@ protected void mkdir(Path dir, FsPermission permission, boolean createParent)
   }
 
   @Override
-  protected FSDataInputStream open(Path f, int bufferSize) throws IOException {
+  public FSDataInputStream open(Path f, int bufferSize) throws IOException {
     checkPath(f);
     return fsImpl.open(f, bufferSize);
   }
 
   @Override
   @SuppressWarnings("deprecation") // call to rename
-  protected void renameInternal(Path src, Path dst) throws IOException {
+  public void renameInternal(Path src, Path dst) throws IOException {
     checkPath(src);
     checkPath(dst);
     fsImpl.rename(src, dst, Options.Rename.NONE);
   }
 
   @Override
-  protected void setOwner(Path f, String username, String groupname)
+  public void setOwner(Path f, String username, String groupname)
       throws IOException {
     checkPath(f);
     fsImpl.setOwner(f, username, groupname);
   }
 
   @Override
-  protected void setPermission(Path f, FsPermission permission)
+  public void setPermission(Path f, FsPermission permission)
       throws IOException {
     checkPath(f);
     fsImpl.setPermission(f, permission);
   }
 
   @Override
-  protected boolean setReplication(Path f, short replication)
+  public boolean setReplication(Path f, short replication)
       throws IOException {
     checkPath(f);
     return fsImpl.setReplication(f, replication);
   }
 
   @Override
-  protected void setTimes(Path f, long mtime, long atime) throws IOException {
+  public void setTimes(Path f, long mtime, long atime) throws IOException {
     checkPath(f);
     fsImpl.setTimes(f, mtime, atime);
   }
 
   @Override
-  protected void setVerifyChecksum(boolean verifyChecksum) throws IOException {
+  public void setVerifyChecksum(boolean verifyChecksum) throws IOException {
     fsImpl.setVerifyChecksum(verifyChecksum);
   }
 
   @Override
-  protected boolean supportsSymlinks() {
+  public boolean supportsSymlinks() {
     return false;
   }  
   
   @Override
-  protected void createSymlink(Path target, Path link, boolean createParent) 
+  public void createSymlink(Path target, Path link, boolean createParent) 
       throws IOException { 
     throw new IOException("File system does not support symlinks");
   } 
   
   @Override
-  protected Path getLinkTarget(final Path f) throws IOException {
+  public Path getLinkTarget(final Path f) throws IOException {
     /* We should never get here. Any file system that threw an 
      * UnresolvedLinkException, causing this function to be called,
      * should override getLinkTarget. 
diff --git a/src/java/org/apache/hadoop/fs/FileContext.java b/src/java/org/apache/hadoop/fs/FileContext.java
index 01765743f8c..01014587d6a 100644
--- a/src/java/org/apache/hadoop/fs/FileContext.java
+++ b/src/java/org/apache/hadoop/fs/FileContext.java
@@ -26,7 +26,6 @@
 import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.IdentityHashMap;
-import java.util.Iterator;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
@@ -306,7 +305,7 @@ private AbstractFileSystem getFSofPath(final Path absOrFqPath)
    * @param aConf
    * @return new FileContext with specifed FS as default.
    */
-  protected static FileContext getFileContext(final AbstractFileSystem defFS,
+  public static FileContext getFileContext(final AbstractFileSystem defFS,
                     final Configuration aConf) {
     return new FileContext(defFS, FsPermission.getUMask(aConf), aConf);
   }
@@ -426,21 +425,33 @@ public static FileContext getLocalFSFileContext(final Configuration aConf)
   @InterfaceAudience.Private
   @InterfaceStability.Unstable /* return type will change to AFS once
                                   HADOOP-6223 is completed */
-  protected AbstractFileSystem getDefaultFileSystem() {
+  public AbstractFileSystem getDefaultFileSystem() {
     return defaultFS;
   }
   
   /**
-   * Set the working directory for wd-relative names (such a "foo/bar").
-   * @param newWDir
-   * @throws IOException
+   * Set the working directory for wd-relative names (such a "foo/bar"). Working
+   * directory feature is provided by simply prefixing relative names with the
+   * working dir. Note this is different from Unix where the wd is actually set
+   * to the inode. Hence setWorkingDir does not follow symlinks etc. This works
+   * better in a distributed environment that has multiple independent roots.
+   * {@link #getWorkingDirectory()} should return what setWorkingDir() set.
    * 
-   * newWdir can be one of 
-   *     - relative path:  "foo/bar";
-   *     - absolute without scheme: "/foo/bar"
-   *     - fully qualified with scheme: "xx://auth/foo/bar"
-   *  Illegal WDs:
-   *      - relative with scheme: "xx:foo/bar" 
+   * @param newWDir new working directory
+   * @throws IOException 
+   * <br>
+   *           NewWdir can be one of:
+   *           <ul>
+   *           <li>relative path: "foo/bar";</li>
+   *           <li>absolute without scheme: "/foo/bar"</li>
+   *           <li>fully qualified with scheme: "xx://auth/foo/bar"</li>
+   *           </ul>
+   * <br>
+   *           Illegal WDs:
+   *           <ul>
+   *           <li>relative with scheme: "xx:foo/bar"</li>
+   *           <li>non existent directory</li>
+   *           </ul>
    */
   public void setWorkingDirectory(final Path newWDir) throws IOException {
     checkNotSchemeWithRelative(newWDir);
@@ -448,7 +459,7 @@ public void setWorkingDirectory(final Path newWDir) throws IOException {
      * path is not relative first since resolve requires and returns 
      * an absolute path.
      */  
-    final Path newWorkingDir = resolve(new Path(workingDir, newWDir));
+    final Path newWorkingDir = new Path(workingDir, newWDir);
     FileStatus status = getFileStatus(newWorkingDir);
     if (status.isFile()) {
       throw new FileNotFoundException("Cannot setWD to a file");
diff --git a/src/java/org/apache/hadoop/fs/FilterFs.java b/src/java/org/apache/hadoop/fs/FilterFs.java
index ce1bbd71118..9ca00265f96 100644
--- a/src/java/org/apache/hadoop/fs/FilterFs.java
+++ b/src/java/org/apache/hadoop/fs/FilterFs.java
@@ -56,22 +56,27 @@ protected FilterFs(AbstractFileSystem fs) throws IOException,
   }
 
   @Override
-  protected Statistics getStatistics() {
+  public Statistics getStatistics() {
     return myFs.getStatistics();
   }
+  
+  @Override
+  public Path makeQualified(Path path) {
+    return myFs.makeQualified(path);
+  }
 
   @Override
-  protected Path getInitialWorkingDirectory() {
+  public Path getInitialWorkingDirectory() {
     return myFs.getInitialWorkingDirectory();
   }
   
   @Override
-  protected Path getHomeDirectory() {
+  public Path getHomeDirectory() {
     return myFs.getHomeDirectory();
   }
   
   @Override
-  protected FSDataOutputStream createInternal(Path f,
+  public FSDataOutputStream createInternal(Path f,
     EnumSet<CreateFlag> flag, FsPermission absolutePermission, int bufferSize,
     short replication, long blockSize, Progressable progress,
     int bytesPerChecksum, boolean createParent) 
@@ -82,85 +87,85 @@ protected FSDataOutputStream createInternal(Path f,
   }
 
   @Override
-  protected boolean delete(Path f, boolean recursive) 
+  public boolean delete(Path f, boolean recursive) 
       throws IOException, UnresolvedLinkException {
     checkPath(f);
     return myFs.delete(f, recursive);
   }
 
   @Override
-  protected BlockLocation[] getFileBlockLocations(Path f, long start, long len)
+  public BlockLocation[] getFileBlockLocations(Path f, long start, long len)
       throws IOException, UnresolvedLinkException {
     checkPath(f);
     return myFs.getFileBlockLocations(f, start, len);
   }
 
   @Override
-  protected FileChecksum getFileChecksum(Path f) 
+  public FileChecksum getFileChecksum(Path f) 
       throws IOException, UnresolvedLinkException {
     checkPath(f);
     return myFs.getFileChecksum(f);
   }
 
   @Override
-  protected FileStatus getFileStatus(Path f) 
+  public FileStatus getFileStatus(Path f) 
       throws IOException, UnresolvedLinkException {
     checkPath(f);
     return myFs.getFileStatus(f);
   }
 
   @Override
-  protected FileStatus getFileLinkStatus(final Path f) 
+  public FileStatus getFileLinkStatus(final Path f) 
     throws IOException, UnresolvedLinkException {
     checkPath(f);
     return myFs.getFileLinkStatus(f);
   }
   
   @Override
-  protected FsStatus getFsStatus(final Path f) throws AccessControlException,
+  public FsStatus getFsStatus(final Path f) throws AccessControlException,
     FileNotFoundException, UnresolvedLinkException, IOException {
     return myFs.getFsStatus(f);
   }
 
   @Override
-  protected FsStatus getFsStatus() throws IOException {
+  public FsStatus getFsStatus() throws IOException {
     return myFs.getFsStatus();
   }
 
   @Override
-  protected FsServerDefaults getServerDefaults() throws IOException {
+  public FsServerDefaults getServerDefaults() throws IOException {
     return myFs.getServerDefaults();
   }
 
   @Override
-  protected int getUriDefaultPort() {
+  public int getUriDefaultPort() {
     return myFs.getUriDefaultPort();
   }
 
   @Override
-  protected URI getUri() {
+  public URI getUri() {
     return myFs.getUri();
   }
   
   @Override
-  protected void checkPath(Path path) {
+  public void checkPath(Path path) {
     myFs.checkPath(path);
   }
   
   @Override
-  protected String getUriPath(final Path p) {
+  public String getUriPath(final Path p) {
     return myFs.getUriPath(p);
   }
   
   @Override
-  protected FileStatus[] listStatus(Path f) 
+  public FileStatus[] listStatus(Path f) 
       throws IOException, UnresolvedLinkException {
     checkPath(f);
     return myFs.listStatus(f);
   }
 
   @Override
-  protected void mkdir(Path dir, FsPermission permission, boolean createParent)
+  public void mkdir(Path dir, FsPermission permission, boolean createParent)
     throws IOException, UnresolvedLinkException {
     checkPath(dir);
     myFs.mkdir(dir, permission, createParent);
@@ -168,21 +173,21 @@ protected void mkdir(Path dir, FsPermission permission, boolean createParent)
   }
 
   @Override
-  protected FSDataInputStream open(final Path f) throws AccessControlException,
+  public FSDataInputStream open(final Path f) throws AccessControlException,
     FileNotFoundException, UnresolvedLinkException, IOException {
     checkPath(f);
     return myFs.open(f);
   }
 
   @Override
-  protected FSDataInputStream open(Path f, int bufferSize) 
+  public FSDataInputStream open(Path f, int bufferSize) 
     throws IOException, UnresolvedLinkException {
     checkPath(f);
     return myFs.open(f, bufferSize);
   }
 
   @Override
-  protected void renameInternal(Path src, Path dst) 
+  public void renameInternal(Path src, Path dst) 
     throws IOException, UnresolvedLinkException {
     checkPath(src);
     checkPath(dst);
@@ -190,7 +195,7 @@ protected void renameInternal(Path src, Path dst)
   }
 
   @Override
-  protected void renameInternal(final Path src, final Path dst,
+  public void renameInternal(final Path src, final Path dst,
       boolean overwrite) throws AccessControlException,
       FileAlreadyExistsException, FileNotFoundException,
       ParentNotDirectoryException, UnresolvedLinkException, IOException {
@@ -198,7 +203,7 @@ protected void renameInternal(final Path src, final Path dst,
   }
   
   @Override
-  protected void setOwner(Path f, String username, String groupname)
+  public void setOwner(Path f, String username, String groupname)
     throws IOException, UnresolvedLinkException {
     checkPath(f);
     myFs.setOwner(f, username, groupname);
@@ -206,45 +211,45 @@ protected void setOwner(Path f, String username, String groupname)
   }
 
   @Override
-  protected void setPermission(Path f, FsPermission permission)
+  public void setPermission(Path f, FsPermission permission)
     throws IOException, UnresolvedLinkException {
     checkPath(f);
     myFs.setPermission(f, permission);
   }
 
   @Override
-  protected boolean setReplication(Path f, short replication)
+  public boolean setReplication(Path f, short replication)
     throws IOException, UnresolvedLinkException {
     checkPath(f);
     return myFs.setReplication(f, replication);
   }
 
   @Override
-  protected void setTimes(Path f, long mtime, long atime) 
+  public void setTimes(Path f, long mtime, long atime) 
       throws IOException, UnresolvedLinkException {
     checkPath(f);
     myFs.setTimes(f, mtime, atime);
   }
 
   @Override
-  protected void setVerifyChecksum(boolean verifyChecksum) 
+  public void setVerifyChecksum(boolean verifyChecksum) 
       throws IOException, UnresolvedLinkException {
     myFs.setVerifyChecksum(verifyChecksum);
   }
 
   @Override
-  protected boolean supportsSymlinks() {
+  public boolean supportsSymlinks() {
     return myFs.supportsSymlinks();
   }
 
   @Override
-  protected void createSymlink(Path target, Path link, boolean createParent) 
+  public void createSymlink(Path target, Path link, boolean createParent) 
     throws IOException, UnresolvedLinkException {
     myFs.createSymlink(target, link, createParent);
   }
 
   @Override
-  protected Path getLinkTarget(final Path f) throws IOException {
+  public Path getLinkTarget(final Path f) throws IOException {
     return myFs.getLinkTarget(f);
   }
 }
diff --git a/src/java/org/apache/hadoop/fs/FsConstants.java b/src/java/org/apache/hadoop/fs/FsConstants.java
index b2bf4ba4c77..3ebbac220d9 100644
--- a/src/java/org/apache/hadoop/fs/FsConstants.java
+++ b/src/java/org/apache/hadoop/fs/FsConstants.java
@@ -33,4 +33,11 @@ public interface FsConstants {
   
   // URI scheme for FTP
   public static final String FTP_SCHEME = "ftp";
+  
+  
+  /**
+   * ViewFs: viewFs file system (ie the mount file system on client side)
+   */
+  public static final URI VIEWFS_URI = URI.create("viewfs:///");
+  public static final String VIEWFS_SCHEME = "viewfs";
 }
diff --git a/src/java/org/apache/hadoop/fs/ftp/FtpFs.java b/src/java/org/apache/hadoop/fs/ftp/FtpFs.java
index ef0d5189605..0a97375d9cd 100644
--- a/src/java/org/apache/hadoop/fs/ftp/FtpFs.java
+++ b/src/java/org/apache/hadoop/fs/ftp/FtpFs.java
@@ -52,12 +52,12 @@ public class FtpFs extends DelegateToFileSystem {
   }
   
   @Override
-  protected int getUriDefaultPort() {
+  public int getUriDefaultPort() {
     return FTP.DEFAULT_PORT;
   }
   
   @Override
-  protected FsServerDefaults getServerDefaults() throws IOException {
+  public FsServerDefaults getServerDefaults() throws IOException {
     return FtpConfigKeys.getServerDefaults();
   }
 }
diff --git a/src/java/org/apache/hadoop/fs/local/LocalConfigKeys.java b/src/java/org/apache/hadoop/fs/local/LocalConfigKeys.java
index 962b5292f83..0561151b0dc 100644
--- a/src/java/org/apache/hadoop/fs/local/LocalConfigKeys.java
+++ b/src/java/org/apache/hadoop/fs/local/LocalConfigKeys.java
@@ -44,7 +44,7 @@ public class LocalConfigKeys extends CommonConfigurationKeys {
                                                 "file.client-write-packet-size";
   public static final int CLIENT_WRITE_PACKET_SIZE_DEFAULT = 64*1024;
 
-  protected static FsServerDefaults getServerDefaults() throws IOException {
+  public static FsServerDefaults getServerDefaults() throws IOException {
     return new FsServerDefaults(
         BLOCK_SIZE_DEFAULT,
         BYTES_PER_CHECKSUM_DEFAULT,
diff --git a/src/java/org/apache/hadoop/fs/local/RawLocalFs.java b/src/java/org/apache/hadoop/fs/local/RawLocalFs.java
index 2032d713ad4..ab3e2e18ada 100644
--- a/src/java/org/apache/hadoop/fs/local/RawLocalFs.java
+++ b/src/java/org/apache/hadoop/fs/local/RawLocalFs.java
@@ -62,22 +62,22 @@ public class RawLocalFs extends DelegateToFileSystem {
   }
   
   @Override
-  protected int getUriDefaultPort() {
+  public int getUriDefaultPort() {
     return -1; // No default port for file:///
   }
   
   @Override
-  protected FsServerDefaults getServerDefaults() throws IOException {
+  public FsServerDefaults getServerDefaults() throws IOException {
     return LocalConfigKeys.getServerDefaults();
   }
   
   @Override
-  protected boolean supportsSymlinks() {
+  public boolean supportsSymlinks() {
     return true;
   }  
   
   @Override
-  protected void createSymlink(Path target, Path link, boolean createParent) 
+  public void createSymlink(Path target, Path link, boolean createParent) 
       throws IOException {
     final String targetScheme = target.toUri().getScheme();
     if (targetScheme != null && !"file".equals(targetScheme)) {
@@ -123,7 +123,7 @@ private String readLink(Path p) {
    * the object the link refers to.
    */
   @Override
-  protected FileStatus getFileLinkStatus(final Path f) throws IOException {
+  public FileStatus getFileLinkStatus(final Path f) throws IOException {
     String target = readLink(f);
     try {
       FileStatus fs = getFileStatus(f);
@@ -160,7 +160,7 @@ protected FileStatus getFileLinkStatus(final Path f) throws IOException {
   }
   
   @Override
-  protected Path getLinkTarget(Path f) throws IOException {
+  public Path getLinkTarget(Path f) throws IOException {
     /* We should never get here. Valid local links are resolved transparently
      * by the underlying local file system and accessing a dangling link will 
      * result in an IOException, not an UnresolvedLinkException, so FileContext
diff --git a/src/test/core/org/apache/hadoop/fs/FileContextMainOperationsBaseTest.java b/src/test/core/org/apache/hadoop/fs/FileContextMainOperationsBaseTest.java
index 500e50e2851..1af5a6391bc 100644
--- a/src/test/core/org/apache/hadoop/fs/FileContextMainOperationsBaseTest.java
+++ b/src/test/core/org/apache/hadoop/fs/FileContextMainOperationsBaseTest.java
@@ -91,7 +91,7 @@ public void setUp() throws Exception {
   
   @After
   public void tearDown() throws Exception {
-    fc.delete(getTestRootPath(fc, "test"), true);
+    fc.delete(new Path(getAbsoluteTestRootPath(fc), new Path("test")), true);
     fc.delete(new Path(LOCAL_FS_ROOT_URI), true);
   }
   
@@ -125,7 +125,7 @@ public void testFsStatus() throws Exception {
   public void testWorkingDirectory() throws Exception {
 
     // First we cd to our test root
-    Path workDir = new Path(getTestRootDir(fc), new Path("test"));
+    Path workDir = new Path(getAbsoluteTestRootPath(fc), new Path("test"));
     fc.setWorkingDirectory(workDir);
     Assert.assertEquals(workDir, fc.getWorkingDirectory());
 
@@ -138,7 +138,7 @@ public void testWorkingDirectory() throws Exception {
     // cd using a relative path
 
     // Go back to our test root
-    workDir = getTestRootPath(fc, "test");
+    workDir = new Path(getAbsoluteTestRootPath(fc), new Path("test"));
     fc.setWorkingDirectory(workDir);
     Assert.assertEquals(workDir, fc.getWorkingDirectory());
     
@@ -157,6 +157,11 @@ public void testWorkingDirectory() throws Exception {
     Path absolutePath = new Path(absoluteDir, "foo");
     fc.create(absolutePath, EnumSet.of(CreateFlag.CREATE)).close();
     fc.open(new Path("foo")).close();
+    
+    
+    // Now mkdir relative to the dir we cd'ed to
+    fc.mkdir(new Path("newDir"), FileContext.DEFAULT_PERM, true);
+    Assert.assertTrue(isDir(fc, new Path(absoluteDir, "newDir")));
 
     absoluteDir = getTestRootPath(fc, "nonexistingPath");
     try {
diff --git a/src/test/core/org/apache/hadoop/fs/FileContextPermissionBase.java b/src/test/core/org/apache/hadoop/fs/FileContextPermissionBase.java
index cd326b21ef6..203485eaa03 100644
--- a/src/test/core/org/apache/hadoop/fs/FileContextPermissionBase.java
+++ b/src/test/core/org/apache/hadoop/fs/FileContextPermissionBase.java
@@ -65,7 +65,7 @@ public abstract class FileContextPermissionBase {
     }
   }
   
-  static FileContext fc;
+  protected static FileContext fc;
 
   @Before
   public void setUp() throws Exception {
diff --git a/src/test/core/org/apache/hadoop/fs/FileContextSymlinkBaseTest.java b/src/test/core/org/apache/hadoop/fs/FileContextSymlinkBaseTest.java
index 8c7885e426d..bb0affe3b84 100644
--- a/src/test/core/org/apache/hadoop/fs/FileContextSymlinkBaseTest.java
+++ b/src/test/core/org/apache/hadoop/fs/FileContextSymlinkBaseTest.java
@@ -117,18 +117,13 @@ public void testStatRoot() throws IOException {
   }
   
   @Test
-  /** Test setWorkingDirectory resolves symlinks */
-  public void testSetWDResolvesLinks() throws IOException {
+  /** Test setWorkingDirectory not resolves symlinks */
+  public void testSetWDNotResolvesLinks() throws IOException {
     Path dir       = new Path(testBaseDir1());
     Path linkToDir = new Path(testBaseDir1()+"/link");
     fc.createSymlink(dir, linkToDir, false);
     fc.setWorkingDirectory(linkToDir);
-    // Local file system does not resolve symlinks, others do.
-    if ("file".equals(getScheme())) {
-      assertEquals(linkToDir.getName(), fc.getWorkingDirectory().getName());
-    } else {
-      assertEquals(dir.getName(), fc.getWorkingDirectory().getName());
-    }
+    assertEquals(linkToDir.getName(), fc.getWorkingDirectory().getName());
   }
   
   @Test
diff --git a/src/test/core/org/apache/hadoop/fs/FileContextTestHelper.java b/src/test/core/org/apache/hadoop/fs/FileContextTestHelper.java
index 633dc748bd6..1a817233036 100644
--- a/src/test/core/org/apache/hadoop/fs/FileContextTestHelper.java
+++ b/src/test/core/org/apache/hadoop/fs/FileContextTestHelper.java
@@ -17,23 +17,24 @@
  */
 package org.apache.hadoop.fs;
 
-import java.io.BufferedReader;
+
 import java.io.DataInputStream;
 import java.io.IOException;
 import java.io.FileNotFoundException;
-import java.io.InputStreamReader;
 import java.util.EnumSet;
 
 import org.apache.hadoop.fs.Options.CreateOpts;
 import org.apache.hadoop.fs.Options.CreateOpts.BlockSize;
 import org.apache.hadoop.io.IOUtils;
+import org.junit.Assert;
 
 /**
  * Helper class for unit tests.
  */
 public final class FileContextTestHelper {
-  private static final String TEST_ROOT_DIR = System.getProperty("test.build.data",
-      "build/test/data") + "/test";
+  // The test root is relative to the <wd>/build/test/data by default
+  public static final String TEST_ROOT_DIR = 
+    System.getProperty("test.build.data", "build/test/data") + "/test";
   private static final int DEFAULT_BLOCK_SIZE = 1024;
   private static final int DEFAULT_NUM_BLOCKS = 2;
   private static String absTestRootDir = null;
@@ -61,20 +62,24 @@ public static Path getTestRootPath(FileContext fc, String pathString) {
     return fc.makeQualified(new Path(TEST_ROOT_DIR, pathString));
   }
   
+  
+  // the getAbsolutexxx method is needed because the root test dir
+  // can be messed up by changing the working dir.
+
   public static String getAbsoluteTestRootDir(FileContext fc)
       throws IOException {
     if (absTestRootDir == null) {
       if (TEST_ROOT_DIR.startsWith("/")) {
         absTestRootDir = TEST_ROOT_DIR;
       } else {
-        absTestRootDir = getDefaultWorkingDirectory(fc).toString() + "/"
+        absTestRootDir = fc.getWorkingDirectory().toString() + "/"
             + TEST_ROOT_DIR;
       }
     }
     return absTestRootDir;
   }
   
-  public static Path getTestRootDir(FileContext fc) throws IOException {
+  public static Path getAbsoluteTestRootPath(FileContext fc) throws IOException {
     return fc.makeQualified(new Path(getAbsoluteTestRootDir(fc)));
   }
 
@@ -117,6 +122,11 @@ public static Path createFile(FileContext fc, String name) throws IOException {
     return path;
   }
   
+  public static void createFileNonRecursive(FileContext fc, String name)
+  throws IOException {
+    Path path = getTestRootPath(fc, name);
+    createFileNonRecursive(fc, path);
+  }
   public static void createFileNonRecursive(FileContext fc, Path path)
       throws IOException {
     createFile(fc, path, DEFAULT_NUM_BLOCKS, CreateOpts.donotCreateParent());
@@ -150,18 +160,71 @@ public static boolean isSymlink(FileContext fc, Path p) throws IOException {
     }
   }
   
-  public static void writeFile(FileContext fc, Path path,byte b[]) throws Exception {
+  public static void writeFile(FileContext fc, Path path, byte b[])
+      throws Exception {
     FSDataOutputStream out = 
       fc.create(path,EnumSet.of(CreateFlag.CREATE), CreateOpts.createParent());
     out.write(b);
     out.close();
   }
   
-  public static byte[] readFile(FileContext fc, Path path, int len ) throws Exception {
+  public static byte[] readFile(FileContext fc, Path path, int len)
+      throws Exception {
     DataInputStream dis = fc.open(path);
     byte[] buffer = new byte[len];
     IOUtils.readFully(dis, buffer, 0, len);
     dis.close();
     return buffer;
   }
+  public static FileStatus containsPath(FileContext fc, Path path,
+      FileStatus[] dirList)
+    throws IOException {
+    return containsPath(getTestRootPath(fc, path.toString()), dirList);
+  }
+  
+  public static FileStatus containsPath(Path path,
+      FileStatus[] dirList)
+    throws IOException {
+    for(int i = 0; i < dirList.length; i ++) { 
+      if (path.equals(dirList[i].getPath()))
+        return dirList[i];
+      }
+    return null;
+  }
+  
+  public static FileStatus containsPath(FileContext fc, String path,
+      FileStatus[] dirList)
+     throws IOException {
+    return containsPath(fc, new Path(path), dirList);
+  }
+  
+  public static enum fileType {isDir, isFile, isSymlink};
+  
+  public static void checkFileStatus(FileContext aFc, String path,
+      fileType expectedType) throws IOException {
+    FileStatus s = aFc.getFileStatus(new Path(path));
+    Assert.assertNotNull(s);
+    if (expectedType == fileType.isDir) {
+      Assert.assertTrue(s.isDirectory());
+    } else if (expectedType == fileType.isFile) {
+      Assert.assertTrue(s.isFile());
+    } else if (expectedType == fileType.isSymlink) {
+      Assert.assertTrue(s.isSymlink());
+    }
+    Assert.assertEquals(aFc.makeQualified(new Path(path)), s.getPath());
+  }
+  
+  public static void checkFileLinkStatus(FileContext aFc, String path,
+      fileType expectedType) throws IOException {
+    FileStatus s = aFc.getFileLinkStatus(new Path(path));
+    Assert.assertNotNull(s);
+    if (expectedType == fileType.isDir) {
+      Assert.assertTrue(s.isDirectory());
+    } else if (expectedType == fileType.isFile) {
+      Assert.assertTrue(s.isFile());
+    } else if (expectedType == fileType.isSymlink) {
+      Assert.assertTrue(s.isSymlink());
+    }
+    Assert.assertEquals(aFc.makeQualified(new Path(path)), s.getPath());
+  }
 }