Merge -r 949657:949658 from trunk to 0.21 branch. Fixes: HADOOP-6585

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.21@949660 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -862,6 +862,9 @@ Release 0.21.0 - Unreleased
     HADOOP-6563. Add more symlink tests to cover intermediate symlinks
     in paths. (Eli Collins via suresh)
 
+    HADOOP-6585.  Add FileStatus#isDirectory and isFile.  (Eli Collins via
+    tomwhite)
+
   OPTIMIZATIONS
 
     HADOOP-5595. NameNode does not need to run a replicator to choose a

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

@@ -592,9 +592,6 @@ public abstract class AbstractFileSystem {
       ParentNotDirectoryException, UnresolvedLinkException, IOException {
     // Default implementation deals with overwrite in a non-atomic way
     final FileStatus srcStatus = getFileLinkStatus(src);
-    if (srcStatus == null) {
-      throw new FileNotFoundException("rename source " + src + " not found.");
-    }
 
     FileStatus dstStatus;
     try {
@@ -611,32 +608,29 @@ public abstract class AbstractFileSystem {
         throw new FileAlreadyExistsException(
             "Cannot rename symlink "+src+" to its target "+dst);
       }
-      if (srcStatus.isDir() != dstStatus.isDir()) {
-        throw new IOException("Source " + src + " Destination " + dst
-            + " both should be either file or directory");
+      // It's OK to rename a file to a symlink and vice versa
+      if (srcStatus.isDirectory() != dstStatus.isDirectory()) {
+        throw new IOException("Source " + src + " and destination " + dst
+            + " must both be directories");
       }
       if (!overwrite) {
-        throw new FileAlreadyExistsException("rename destination " + dst
+        throw new FileAlreadyExistsException("Rename destination " + dst
             + " already exists.");
       }
       // Delete the destination that is a file or an empty directory
-      if (dstStatus.isDir()) {
+      if (dstStatus.isDirectory()) {
         Iterator<FileStatus> list = listStatusIterator(dst);
         if (list != null && list.hasNext()) {
           throw new IOException(
-              "rename cannot overwrite non empty destination directory " + dst);
+              "Rename cannot overwrite non empty destination directory " + dst);
         }
       }
       delete(dst, false);
     } else {
       final Path parent = dst.getParent();
-      final FileStatus parentStatus = getFileLinkStatus(parent);
-      if (parentStatus == null) {
-        throw new FileNotFoundException("rename destination parent " + parent
-            + " not found.");
-      }
-      if (!parentStatus.isDir() && !parentStatus.isSymlink()) {
-        throw new ParentNotDirectoryException("rename destination parent "
+      final FileStatus parentStatus = getFileStatus(parent);
+      if (parentStatus.isFile()) {
+        throw new ParentNotDirectoryException("Rename destination parent "
             + parent + " is a file.");
       }
     }

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

@@ -456,7 +456,7 @@ public abstract class ChecksumFileSystem extends FilterFileSystem {
     } catch(FileNotFoundException e) {
       return false;
     }
-    if(fstatus.isDir()) {
+    if (fstatus.isDirectory()) {
       //this works since the crcs are in the same
       //directories and the files. so we just delete
       //everything in the underlying filesystem

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

@@ -388,7 +388,7 @@ public abstract class ChecksumFs extends FilterFs {
   private boolean isDirectory(Path f) 
     throws IOException, UnresolvedLinkException {
     try {
-      return getMyFs().getFileStatus(f).isDir();
+      return getMyFs().getFileStatus(f).isDirectory();
     } catch (FileNotFoundException e) {
       return false;               // f does not exist
     }
@@ -450,7 +450,7 @@ public abstract class ChecksumFs extends FilterFs {
     } catch(FileNotFoundException e) {
       return false;
     }
-    if (fstatus.isDir()) {
+    if (fstatus.isDirectory()) {
       //this works since the crcs are in the same
       //directories and the files. so we just delete
       //everything in the underlying filesystem

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

@@ -73,7 +73,7 @@ public abstract class DelegateToFileSystem extends AbstractFileSystem {
       if (stat == null) {
         throw new FileNotFoundException("Missing parent:" + f);
       }
-      if (!stat.isDir()) {
+      if (!stat.isDirectory()) {
           throw new ParentNotDirectoryException("parent is not a dir:" + f);
       }
       // parent does exist - go ahead with create of file.

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

@@ -450,7 +450,7 @@ public final class FileContext {
      */  
     final Path newWorkingDir = resolve(new Path(workingDir, newWDir));
     FileStatus status = getFileStatus(newWorkingDir);
-    if (!status.isDir()) {
+    if (status.isFile()) {
       throw new FileNotFoundException("Cannot setWD to a file");
     }
     workingDir = newWorkingDir;
@@ -1420,16 +1420,14 @@ public final class FileContext {
         throws AccessControlException, FileNotFoundException,
         UnsupportedFileSystemException, IOException {
       FileStatus status = FileContext.this.getFileStatus(f);
-      if (!status.isDir()) {
-        // f is a file
+      if (status.isFile()) {
         return new ContentSummary(status.getLen(), 1, 0);
       }
-      // f is a directory
       long[] summary = {0, 0, 1};
       Iterator<FileStatus> statusIterator = FileContext.this.listStatus(f);
       while(statusIterator.hasNext()) {
         FileStatus s = statusIterator.next();
-        ContentSummary c = s.isDir() ? getContentSummary(s.getPath()) :
+        ContentSummary c = s.isDirectory() ? getContentSummary(s.getPath()) :
                                        new ContentSummary(s.getLen(), 1, 0);
         summary[0] += c.getLength();
         summary[1] += c.getFileCount();
@@ -1847,7 +1845,7 @@ public final class FileContext {
       Path qDst = makeQualified(dst);
       checkDest(qSrc.getName(), qDst, overwrite);
       FileStatus fs = FileContext.this.getFileStatus(qSrc);
-      if (fs.isDir()) {
+      if (fs.isDirectory()) {
         checkDependencies(qSrc, qDst);
         mkdir(qDst, FsPermission.getDefault(), true);
         FileStatus[] contents = listStatus(qSrc);
@@ -2015,7 +2013,7 @@ public final class FileContext {
       throws AccessControlException, IOException {
     FileStatus dstFs = getFileStatus(dst);
     try {
-      if (dstFs.isDir()) {
+      if (dstFs.isDirectory()) {
         if (null == srcName) {
           throw new IOException("Target " + dst + " is a directory");
         }

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

@@ -91,13 +91,42 @@ public class FileStatus implements Writable, Comparable {
     return length;
   }
 
+  /**
+   * Is this a file?
+   * @return true if this is a file
+   */
+  public boolean isFile() {
+    return !isdir && !isSymlink();
+  }
+
   /**
    * Is this a directory?
    * @return true if this is a directory
    */
+  public boolean isDirectory() {
+    return isdir;
+  }
+  
+  /**
+   * Old interface, instead use the explicit {@link FileStatus#isFile()}, 
+   * {@link FileStatus#isDirectory()}, and {@link FileStatus#isSymlink()} 
+   * @return true if this is a directory.
+   * @deprecated Use {@link FileStatus#isFile()},  
+   * {@link FileStatus#isDirectory()}, and {@link FileStatus#isSymlink()} 
+   * instead.
+   */
+  @Deprecated
   public boolean isDir() {
     return isdir;
   }
+  
+  /**
+   * Is this a symbolic link?
+   * @return true if this is a symbolic link
+   */
+  public boolean isSymlink() {
+    return symlink != null;
+  }
 
   /**
    * Get the block size of the file.
@@ -198,14 +227,6 @@ public class FileStatus implements Writable, Comparable {
     this.group = (group == null) ? "" :  group;
   }
 
-  /**
-   * Is this a symbolic link?
-   * @return true if this is a symbolic link
-   */
-  public boolean isSymlink() {
-    return symlink != null;
-  }
-
   /**
    * @return The contents of the symbolic link.
    */

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

@@ -721,8 +721,8 @@ public abstract class FileSystem extends Configured implements Closeable {
       if (stat == null) {
         throw new FileNotFoundException("Missing parent:" + f);
       }
-      if (!stat.isDir()) {
-          throw new ParentNotDirectoryException("parent is not a dir:" + f);
+      if (!stat.isDirectory()) {
+        throw new ParentNotDirectoryException("parent is not a dir:" + f);
       }
       // parent does exist - go ahead with create of file.
     }
@@ -767,8 +767,8 @@ public abstract class FileSystem extends Configured implements Closeable {
       if (stat == null) {
         throw new FileNotFoundException("Missing parent:" + f);
       }
-      if (!stat.isDir()) {
-          throw new ParentNotDirectoryException("parent is not a dir");
+      if (!stat.isDirectory()) {
+        throw new ParentNotDirectoryException("parent is not a dir");
       }
       // parent does exist - go ahead with mkdir of leaf
     }
@@ -912,7 +912,7 @@ public abstract class FileSystem extends Configured implements Closeable {
       dstStatus = null;
     }
     if (dstStatus != null) {
-      if (srcStatus.isDir() != dstStatus.isDir()) {
+      if (srcStatus.isDirectory() != dstStatus.isDirectory()) {
         throw new IOException("Source " + src + " Destination " + dst
             + " both should be either file or directory");
       }
@@ -921,7 +921,7 @@ public abstract class FileSystem extends Configured implements Closeable {
             + " already exists.");
       }
       // Delete the destination that is a file or an empty directory
-      if (dstStatus.isDir()) {
+      if (dstStatus.isDirectory()) {
         FileStatus[] list = listStatus(dst);
         if (list != null && list.length != 0) {
           throw new IOException(
@@ -936,7 +936,7 @@ public abstract class FileSystem extends Configured implements Closeable {
         throw new FileNotFoundException("rename destination parent " + parent
             + " not found.");
       }
-      if (!parentStatus.isDir()) {
+      if (!parentStatus.isDirectory()) {
         throw new ParentNotDirectoryException("rename destination parent " + parent
             + " is a file.");
       }
@@ -1025,7 +1025,7 @@ public abstract class FileSystem extends Configured implements Closeable {
    */
   public boolean isDirectory(Path f) throws IOException {
     try {
-      return getFileStatus(f).isDir();
+      return getFileStatus(f).isDirectory();
     } catch (FileNotFoundException e) {
       return false;               // f does not exist
     }
@@ -1037,7 +1037,7 @@ public abstract class FileSystem extends Configured implements Closeable {
    */
   public boolean isFile(Path f) throws IOException {
     try {
-      return !getFileStatus(f).isDir();
+      return getFileStatus(f).isFile();
     } catch (FileNotFoundException e) {
       return false;               // f does not exist
     }
@@ -1053,14 +1053,14 @@ public abstract class FileSystem extends Configured implements Closeable {
   /** Return the {@link ContentSummary} of a given {@link Path}. */
   public ContentSummary getContentSummary(Path f) throws IOException {
     FileStatus status = getFileStatus(f);
-    if (!status.isDir()) {
+    if (status.isFile()) {
       // f is a file
       return new ContentSummary(status.getLen(), 1, 0);
     }
     // f is a directory
     long[] summary = {0, 0, 1};
     for(FileStatus s : listStatus(f)) {
-      ContentSummary c = s.isDir() ? getContentSummary(s.getPath()) :
+      ContentSummary c = s.isDirectory() ? getContentSummary(s.getPath()) :
                                      new ContentSummary(s.getLen(), 1, 0);
       summary[0] += c.getLength();
       summary[1] += c.getFileCount();

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

@@ -180,7 +180,7 @@ public class FileUtil {
                             "doest not exist");
     } else {
       FileStatus sdst = dstFS.getFileStatus(dst);
-      if (!sdst.isDir()) 
+      if (!sdst.isDirectory()) 
         throw new IOException("copying multiple files, but last argument `" +
                               dst + "' is not a directory");
     }
@@ -219,7 +219,7 @@ public class FileUtil {
                               Configuration conf) throws IOException {
     Path src = srcStatus.getPath();
     dst = checkDest(src.getName(), dstFS, dst, overwrite);
-    if (srcStatus.isDir()) {
+    if (srcStatus.isDirectory()) {
       checkDependencies(srcFS, src, dstFS, dst);
       if (!dstFS.mkdirs(dst)) {
         return false;
@@ -258,7 +258,7 @@ public class FileUtil {
                                   Configuration conf, String addString) throws IOException {
     dstFile = checkDest(srcDir.getName(), dstFS, dstFile, false);
 
-    if (!srcFS.getFileStatus(srcDir).isDir())
+    if (srcFS.getFileStatus(srcDir).isDirectory())
       return false;
    
     OutputStream out = dstFS.create(dstFile);
@@ -266,7 +266,7 @@ public class FileUtil {
     try {
       FileStatus contents[] = srcFS.listStatus(srcDir);
       for (int i = 0; i < contents.length; i++) {
-        if (!contents[i].isDir()) {
+        if (contents[i].isFile()) {
           InputStream in = srcFS.open(contents[i].getPath());
           try {
             IOUtils.copyBytes(in, out, conf, false);
@@ -342,7 +342,7 @@ public class FileUtil {
                               File dst, boolean deleteSource,
                               Configuration conf) throws IOException {
     Path src = srcStatus.getPath();
-    if (srcStatus.isDir()) {
+    if (srcStatus.isDirectory()) {
       if (!dst.mkdirs()) {
         return false;
       }
@@ -367,7 +367,7 @@ public class FileUtil {
       boolean overwrite) throws IOException {
     if (dstFS.exists(dst)) {
       FileStatus sdst = dstFS.getFileStatus(dst);
-      if (sdst.isDir()) {
+      if (sdst.isDirectory()) {
         if (null == srcName) {
           throw new IOException("Target " + dst + " is a directory");
         }

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

@@ -236,7 +236,7 @@ public class FsShell extends Configured implements Tool {
      */
     
     Path src = srcStatus.getPath();
-    if (!srcStatus.isDir()) {
+    if (srcStatus.isFile()) {
       if (dst.exists()) {
         // match the error message in FileUtil.checkDest():
         throw new IOException("Target " + dst + " already exists");
@@ -266,6 +266,8 @@ public class FsShell extends Configured implements Tool {
         FileStatus status = csfs.getFileStatus(csfs.getChecksumFile(src));
         copyToLocal(fs, status, dstcs, false);
       } 
+    } else if (srcStatus.isSymlink()) {
+      throw new AssertionError("Symlinks unsupported");
     } else {
       // once FileUtil.copy() supports tmp file, we don't need to mkdirs().
       if (!dst.mkdirs()) {
@@ -531,7 +533,7 @@ public class FsShell extends Configured implements Tool {
                               Path src, boolean recursive,
                               List<Path> waitingList)
     throws IOException {
-    if (!srcFs.getFileStatus(src).isDir()) {
+    if (srcFs.getFileStatus(src).isFile()) {
       setFileReplication(src, srcFs, newRep, waitingList);
       return;
     }
@@ -543,8 +545,10 @@ public class FsShell extends Configured implements Tool {
     }
 
     for (int i = 0; i < items.length; i++) {
-      if (!items[i].isDir()) {
+      if (items[i].isFile()) {
         setFileReplication(items[i].getPath(), srcFs, newRep, waitingList);
+      } else if (items[i].isSymlink()) {
+        throw new AssertionError("Symlinks unsupported");
       } else if (recursive) {
         setReplication(newRep, srcFs, items[i].getPath(), recursive, 
                        waitingList);
@@ -633,10 +637,10 @@ public class FsShell extends Configured implements Tool {
         Path cur = stat.getPath();
         String mdate = dateForm.format(new Date(stat.getModificationTime()));
         
-        System.out.print((stat.isDir() ? "d" : "-") + 
+        System.out.print((stat.isDirectory() ? "d" : "-") + 
           stat.getPermission() + " ");
         System.out.printf("%"+ maxReplication + 
-          "s ", (!stat.isDir() ? stat.getReplication() : "-"));
+          "s ", (stat.isFile() ? stat.getReplication() : "-"));
         if (maxOwner > 0)
           System.out.printf("%-"+ maxOwner + "s ", stat.getOwner());
         if (maxGroup > 0)
@@ -644,7 +648,7 @@ public class FsShell extends Configured implements Tool {
         System.out.printf("%"+ maxLen + "d ", stat.getLen());
         System.out.print(mdate + " ");
         System.out.println(cur.toUri().getPath());
-        if (recursive && stat.isDir()) {
+        if (recursive && stat.isDirectory()) {
           numOfErrors += ls(stat,srcFs, recursive, printHeader);
         }
       }
@@ -729,7 +733,7 @@ public class FsShell extends Configured implements Tool {
 
       for (FileStatus stat : statusToPrint) {
         long length;
-        if (summary || stat.isDir()) {
+        if (summary || stat.isDirectory()) {
           length = srcFs.getContentSummary(stat.getPath()).getLength();
         } else {
           length = stat.getLen();
@@ -784,7 +788,7 @@ public class FsShell extends Configured implements Tool {
     FileStatus fstatus = null;
     try {
       fstatus = srcFs.getFileStatus(f);
-      if (fstatus.isDir()) {
+      if (fstatus.isDirectory()) {
         throw new IOException("cannot create directory " 
             + src + ": File exists");
       }
@@ -810,7 +814,7 @@ public class FsShell extends Configured implements Tool {
     FileStatus st;
     if (srcFs.exists(f)) {
       st = srcFs.getFileStatus(f);
-      if (st.isDir()) {
+      if (st.isDirectory()) {
         // TODO: handle this
         throw new IOException(src + " is a directory");
       } else if (st.getLen() != 0)
@@ -835,7 +839,7 @@ public class FsShell extends Configured implements Tool {
       case 'z':
         return srcFs.getFileStatus(f).getLen() == 0 ? 0 : 1;
       case 'd':
-        return srcFs.getFileStatus(f).isDir() ? 0 : 1;
+        return srcFs.getFileStatus(f).isDirectory() ? 0 : 1;
       default:
         throw new IOException("Unknown flag: " + flag);
     }
@@ -869,7 +873,8 @@ public class FsShell extends Configured implements Tool {
               buf.append(f.getLen());
               break;
             case 'F':
-              buf.append(f.isDir() ? "directory" : "regular file");
+              buf.append(f.isDirectory() ? "directory" 
+                                         : (f.isFile() ? "regular file" : "symlink"));
               break;
             case 'n':
               buf.append(f.getPath().getName());
@@ -936,7 +941,7 @@ public class FsShell extends Configured implements Tool {
         } catch(IOException e) {
         }
         if((srcFstatus!= null) && (dstFstatus!= null)) {
-          if (srcFstatus.isDir()  && !dstFstatus.isDir()) {
+          if (srcFstatus.isDirectory()  && !dstFstatus.isDirectory()) {
             throw new IOException("cannot overwrite non directory "
                 + dst + " with directory " + srcs[i]);
           }
@@ -1128,7 +1133,7 @@ public class FsShell extends Configured implements Tool {
           + src + ": No such file or directory.");
     }
     
-    if (fs.isDir() && !recursive) {
+    if (fs.isDirectory() && !recursive) {
       throw new IOException("Cannot remove directory \"" + src +
                             "\", use -rmr instead");
     }
@@ -1192,7 +1197,7 @@ public class FsShell extends Configured implements Tool {
     path = new Path(src);
     FileSystem srcFs = path.getFileSystem(getConf());
     FileStatus fileStatus = srcFs.getFileStatus(path);
-    if (fileStatus.isDir()) {
+    if (fileStatus.isDirectory()) {
       throw new IOException("Source must be a file.");
     }
 
@@ -1244,9 +1249,9 @@ public class FsShell extends Configured implements Tool {
   
   /** helper returns listStatus() */
   private static FileStatus[] shellListStatus(String cmd, 
-                                                   FileSystem srcFs,
-                                                   FileStatus src) {
-    if (!src.isDir()) {
+                                              FileSystem srcFs,
+                                              FileStatus src) {
+    if (src.isFile()) {
       FileStatus[] files = { src };
       return files;
     }
@@ -1275,7 +1280,7 @@ public class FsShell extends Configured implements Tool {
                                    boolean recursive) throws IOException {
     int errors = 0;
     handler.run(stat, srcFs);
-    if (recursive && stat.isDir() && handler.okToContinue()) {
+    if (recursive && stat.isDirectory() && handler.okToContinue()) {
       FileStatus[] files = shellListStatus(handler.getName(), srcFs, stat);
       if (files == null) {
         return 1;

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

@@ -215,7 +215,7 @@ public class RawLocalFileSystem extends FileSystem {
     if (!exists(f)) {
       throw new FileNotFoundException("File " + f + " not found.");
     }
-    if (getFileStatus(f).isDir()) {
+    if (getFileStatus(f).isDirectory()) {
       throw new IOException("Cannot append to a diretory (=" + f + " ).");
     }
     return new FSDataOutputStream(new BufferedOutputStream(

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

@@ -259,7 +259,7 @@ public class Trash extends Configured {
             }
 
             for (FileStatus home : homes) {         // dump each trash
-              if (!home.isDir())
+              if (!home.isDirectory())
                 continue;
               try {
                 Trash trash = new Trash(home.getPath(), conf);

src/java/org/apache/hadoop/fs/ftp/FTPFileSystem.java

@@ -164,7 +164,7 @@ public class FTPFileSystem extends FileSystem {
     Path workDir = new Path(client.printWorkingDirectory());
     Path absolute = makeAbsolute(workDir, file);
     FileStatus fileStat = getFileStatus(client, absolute);
-    if (fileStat.isDir()) {
+    if (fileStat.isDirectory()) {
       disconnect(client);
       throw new IOException("Path " + file + " is a directory.");
     }
@@ -307,7 +307,7 @@ public class FTPFileSystem extends FileSystem {
     Path absolute = makeAbsolute(workDir, file);
     String pathName = absolute.toUri().getPath();
     FileStatus fileStat = getFileStatus(client, absolute);
-    if (!fileStat.isDir()) {
+    if (fileStat.isFile()) {
       return client.deleteFile(pathName);
     }
     FileStatus[] dirEntries = listStatus(client, absolute);
@@ -370,7 +370,7 @@ public class FTPFileSystem extends FileSystem {
     Path workDir = new Path(client.printWorkingDirectory());
     Path absolute = makeAbsolute(workDir, file);
     FileStatus fileStat = getFileStatus(client, absolute);
-    if (!fileStat.isDir()) {
+    if (fileStat.isFile()) {
       return new FileStatus[] { fileStat };
     }
     FTPFile[] ftpFiles = client.listFiles(absolute.toUri().getPath());
@@ -500,7 +500,7 @@ public class FTPFileSystem extends FileSystem {
    */
   private boolean isFile(FTPClient client, Path file) {
     try {
-      return !getFileStatus(client, file).isDir();
+      return getFileStatus(client, file).isFile();
     } catch (FileNotFoundException e) {
       return false; // file does not exist
     } catch (IOException ioe) {

src/java/org/apache/hadoop/fs/permission/ChmodParser.java

@@ -44,7 +44,7 @@ public class ChmodParser extends PermissionParser {
   public short applyNewPermission(FileStatus file) {
     FsPermission perms = file.getPermission();
     int existing = perms.toShort();
-    boolean exeOk = file.isDir() || (existing & 0111) != 0;
+    boolean exeOk = file.isDirectory() || (existing & 0111) != 0;
     
     return (short)combineModes(existing, exeOk);
   }

src/java/org/apache/hadoop/fs/s3native/NativeS3FileSystem.java

@@ -354,7 +354,7 @@ public class NativeS3FileSystem extends FileSystem {
     }
     Path absolutePath = makeAbsolute(f);
     String key = pathToKey(absolutePath);
-    if (status.isDir()) {
+    if (status.isDirectory()) {
       if (!recurse && listStatus(f).length > 0) {
         throw new IOException("Can not delete " + f + " at is a not empty directory and recurse option is false");
       }
@@ -509,7 +509,7 @@ public class NativeS3FileSystem extends FileSystem {
   private boolean mkdir(Path f) throws IOException {
     try {
       FileStatus fileStatus = getFileStatus(f);
-      if (!fileStatus.isDir()) {
+      if (fileStatus.isFile()) {
         throw new IOException(String.format(
             "Can't make directory for path '%s' since it is a file.", f));
 
@@ -525,7 +525,7 @@ public class NativeS3FileSystem extends FileSystem {
   @Override
   public FSDataInputStream open(Path f, int bufferSize) throws IOException {
     FileStatus fs = getFileStatus(f); // will throw if the file doesn't exist
-    if (fs.isDir()) {
+    if (fs.isDirectory()) {
       throw new IOException("'" + f + "' is a directory");
     }
     LOG.info("Opening '" + f + "' for reading");
@@ -563,7 +563,7 @@ public class NativeS3FileSystem extends FileSystem {
     // Figure out the final destination
     String dstKey;
     try {
-      boolean dstIsFile = !getFileStatus(dst).isDir();
+      boolean dstIsFile = getFileStatus(dst).isFile();
       if (dstIsFile) {
         LOG.debug(debugPreamble + "returning false as dst is an already existing file");
         return false;
@@ -575,7 +575,7 @@ public class NativeS3FileSystem extends FileSystem {
       LOG.debug(debugPreamble + "using dst as output destination");
       dstKey = pathToKey(makeAbsolute(dst));
       try {
-        if (!getFileStatus(dst.getParent()).isDir()) {
+        if (getFileStatus(dst.getParent()).isFile()) {
           LOG.debug(debugPreamble + "returning false as dst parent exists and is a file");
           return false;
         }
@@ -587,7 +587,7 @@ public class NativeS3FileSystem extends FileSystem {
 
     boolean srcIsFile;
     try {
-      srcIsFile = !getFileStatus(src).isDir();
+      srcIsFile = getFileStatus(src).isFile();
     } catch (FileNotFoundException e) {
       LOG.debug(debugPreamble + "returning false as src does not exist");
       return false;

src/java/org/apache/hadoop/util/DiskChecker.java

@@ -147,7 +147,7 @@ public class DiskChecker {
     FileStatus stat = localFS.getFileStatus(dir);
     FsPermission actual = stat.getPermission();
 
-    if (!stat.isDir())
+    if (!stat.isDirectory())
       throw new DiskErrorException("not a directory: "+ dir.toString());
 
     FsAction user = actual.getUserAction();

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

@@ -257,8 +257,7 @@ public abstract class FileContextSymlinkBaseTest {
     Path linkToFile = new Path(testBaseDir1()+"/linkToFile");    
     createAndWriteFile(file);
     fc.createSymlink(file, linkToFile, false);
-    // NB: isDir is true since we need !isDir to imply file (HADOOP-6584)
-    //assertTrue(isDir(fc, linkToFile));
+    assertFalse(fc.getFileLinkStatus(linkToFile).isDirectory());
     assertTrue(isSymlink(fc, linkToFile));
     assertTrue(isFile(fc, linkToFile));
     assertFalse(isDir(fc, linkToFile));
@@ -275,8 +274,7 @@ public abstract class FileContextSymlinkBaseTest {
 
     assertFalse(fc.getFileStatus(linkToDir).isSymlink());
     assertTrue(isDir(fc, linkToDir));
-    // NB: isDir is true since we need !isDir to imply file (HADOOP-6584)    
-    //assertTrue(fc.getFileLinkStatus(linkToDir).isDir());
+    assertFalse(fc.getFileLinkStatus(linkToDir).isDirectory());
     assertTrue(fc.getFileLinkStatus(linkToDir).isSymlink());
 
     assertFalse(isFile(fc, linkToDir));
@@ -292,9 +290,7 @@ public abstract class FileContextSymlinkBaseTest {
     Path file = new Path("/noSuchFile");
     Path link = new Path(testBaseDir1()+"/link");    
     fc.createSymlink(file, link, false);
-
-    // NB: isDir is true since we need !isDir to imply file (HADOOP-6584)
-    //assertTrue(fc.getFileLinkStatus(link).isDir());
+    assertFalse(fc.getFileLinkStatus(link).isDirectory());
     assertTrue(fc.getFileLinkStatus(link).isSymlink());
   }
   
@@ -365,15 +361,12 @@ public abstract class FileContextSymlinkBaseTest {
 
     // Check getFileStatus 
     assertFalse(fc.getFileStatus(linkAbs).isSymlink());
-    assertFalse(fc.getFileStatus(linkAbs).isDir());
+    assertFalse(fc.getFileStatus(linkAbs).isDirectory());
     assertEquals(fileSize, fc.getFileStatus(linkAbs).getLen());
-    // NB: These are links to files so ensure !isDir is true
-    assertFalse(fc.getFileStatus(linkAbs).isDir());
 
     // Check getFileLinkStatus
     assertTrue(isSymlink(fc, linkAbs));
-    // NB: isDir is true since we need !isDir to imply file (HADOOP-6584)
-    //assertTrue(fc.getFileLinkStatus(linkAbs).isDir());
+    assertFalse(fc.getFileLinkStatus(linkAbs).isDirectory());
 
     // Check getSymlink always returns a qualified target, except
     // when partially qualified paths are used (see tests below).
@@ -595,7 +588,7 @@ public abstract class FileContextSymlinkBaseTest {
     fc.createSymlink(dir1, linkToDir, false);
     assertFalse(isFile(fc, linkToDir));
     assertTrue(isDir(fc, linkToDir)); 
-    assertTrue(fc.getFileStatus(linkToDir).isDir());
+    assertTrue(fc.getFileStatus(linkToDir).isDirectory());
     assertTrue(fc.getFileLinkStatus(linkToDir).isSymlink());
   }
   
@@ -610,7 +603,7 @@ public abstract class FileContextSymlinkBaseTest {
     assertTrue(isFile(fc, fileViaLink));
     assertFalse(isDir(fc, fileViaLink));
     assertFalse(fc.getFileLinkStatus(fileViaLink).isSymlink());
-    assertFalse(fc.getFileStatus(fileViaLink).isDir());
+    assertFalse(fc.getFileStatus(fileViaLink).isDirectory());
     readFile(fileViaLink);
     fc.delete(fileViaLink, true);
     assertFalse(exists(fc, fileViaLink));
@@ -625,7 +618,7 @@ public abstract class FileContextSymlinkBaseTest {
     Path subDirViaLink = new Path(linkToDir, "subDir");
     fc.createSymlink(dir1, linkToDir, false);
     fc.mkdir(subDirViaLink, FileContext.DEFAULT_PERM, true);
-    assertTrue(fc.getFileStatus(subDirViaLink).isDir());
+    assertTrue(isDir(fc, subDirViaLink));
     fc.delete(subDirViaLink, false);
     assertFalse(exists(fc, subDirViaLink));
     assertFalse(exists(fc, subDir));
@@ -702,7 +695,7 @@ public abstract class FileContextSymlinkBaseTest {
     assertTrue(isFile(fc, fileViaLink));
     assertFalse(isDir(fc, fileViaLink));
     assertFalse(fc.getFileLinkStatus(fileViaLink).isSymlink());
-    assertFalse(fc.getFileStatus(fileViaLink).isDir());
+    assertFalse(fc.getFileStatus(fileViaLink).isDirectory());
     readFile(fileViaLink);
   }
 
@@ -826,7 +819,8 @@ public abstract class FileContextSymlinkBaseTest {
       fc.rename(file, subDir);
       fail("Renamed file to a directory");
     } catch (IOException e) {
-      // Expected. Both should be either a file or directory.
+      // Expected. Both must be directories.
+      assertTrue(unwrapException(e) instanceof IOException);
     }
     assertTrue(exists(fc, file));
   }
@@ -880,7 +874,8 @@ public abstract class FileContextSymlinkBaseTest {
       fc.rename(dir1, linkToDir, Rename.OVERWRITE);
       fail("Renamed directory to a symlink");
     } catch (IOException e) {
-      // Expected. Both should be either a file or directory.
+      // Expected. Both must be directories.
+      assertTrue(unwrapException(e) instanceof IOException);
     }
     assertTrue(exists(fc, dir1));
     assertTrue(exists(fc, linkToDir));
@@ -898,7 +893,8 @@ public abstract class FileContextSymlinkBaseTest {
       fc.rename(dir1, linkToFile, Rename.OVERWRITE);
       fail("Renamed directory to a symlink");
     } catch (IOException e) {
-      // Expected. Both should be either a file or directory.
+      // Expected. Both must be directories.
+      assertTrue(unwrapException(e) instanceof IOException);
     }
     assertTrue(exists(fc, dir1));
     assertTrue(exists(fc, linkToFile));
@@ -914,7 +910,8 @@ public abstract class FileContextSymlinkBaseTest {
       fc.rename(dir, link, Rename.OVERWRITE);
       fail("Renamed directory to a symlink"); 
     } catch (IOException e) {
-      // Expected. Both should be either a file or directory.
+      // Expected. Both must be directories.
+      assertTrue(unwrapException(e) instanceof IOException);
     }
     assertTrue(exists(fc, dir));
     assertTrue(fc.getFileLinkStatus(link) != null);
@@ -934,6 +931,7 @@ public abstract class FileContextSymlinkBaseTest {
       fail("Renamed file to symlink w/o overwrite"); 
     } catch (IOException e) {
       // Expected
+      assertTrue(unwrapException(e) instanceof FileAlreadyExistsException);
     }
     fc.rename(file, link, Rename.OVERWRITE);
     assertFalse(exists(fc, file));
@@ -956,6 +954,7 @@ public abstract class FileContextSymlinkBaseTest {
       fail("Renamed file to symlink w/o overwrite"); 
     } catch (IOException e) {
       // Expected
+      assertTrue(unwrapException(e) instanceof FileAlreadyExistsException);
     }
     fc.rename(file1, link, Rename.OVERWRITE);
     assertFalse(exists(fc, file1));
@@ -1017,6 +1016,7 @@ public abstract class FileContextSymlinkBaseTest {
       fail("Renamed w/o passing overwrite");
     } catch (IOException e) {
       // Expected
+      assertTrue(unwrapException(e) instanceof FileAlreadyExistsException);
     }
     fc.rename(link, file1, Rename.OVERWRITE);
     assertFalse(exists(fc, link));
@@ -1036,13 +1036,15 @@ public abstract class FileContextSymlinkBaseTest {
       fc.rename(link, dir2);
       fail("Renamed link to a directory");
     } catch (IOException e) {
-      // Expected
+      // Expected. Both must be directories.
+      assertTrue(unwrapException(e) instanceof IOException);
     }
     try {
       fc.rename(link, dir2, Rename.OVERWRITE);
       fail("Renamed link to a directory");
     } catch (IOException e) {
-      // Expected
+      // Expected. Both must be directories.
+      assertTrue(unwrapException(e) instanceof IOException);
     }
     // Also fails when dir2 has a sub-directory
     fc.mkdir(subDir, FsPermission.getDefault(), false);
@@ -1050,7 +1052,8 @@ public abstract class FileContextSymlinkBaseTest {
       fc.rename(link, dir2, Rename.OVERWRITE);
       fail("Renamed link to a directory");
     } catch (IOException e) {
-      // Expected
+      // Expected. Both must be directories.
+      assertTrue(unwrapException(e) instanceof IOException);
     }
   }
 
@@ -1082,7 +1085,7 @@ public abstract class FileContextSymlinkBaseTest {
     fc.createSymlink(file, link1, false);
     fc.rename(link1, link2);
     assertTrue(fc.getFileLinkStatus(link2).isSymlink());
-    assertFalse(fc.getFileStatus(link2).isDir());
+    assertFalse(fc.getFileStatus(link2).isDirectory());
     readFile(link2);
     readFile(file);
     try {
@@ -1188,6 +1191,44 @@ public abstract class FileContextSymlinkBaseTest {
     readFile(link);
   }
 
+  @Test
+  /** Test rename a file to path with destination that has symlink parent */
+  public void testRenameFileWithDestParentSymlink() throws IOException {
+    Path link  = new Path(testBaseDir1(), "link");
+    Path file1 = new Path(testBaseDir1(), "file1");    
+    Path file2 = new Path(testBaseDir1(), "file2");    
+    Path file3 = new Path(link, "file3");
+    Path dir2  = new Path(testBaseDir2());
+    
+    // Renaming /dir1/file1 to non-existant file /dir1/link/file3 is OK
+    // if link points to a directory...
+    fc.createSymlink(dir2, link, false);
+    createAndWriteFile(file1);
+    fc.rename(file1, file3);
+    assertFalse(exists(fc, file1));
+    assertTrue(exists(fc, file3));
+    fc.rename(file3, file1);
+    
+    // But fails if link is dangling...
+    fc.delete(link, false);
+    fc.createSymlink(file2, link, false);    
+    try {
+      fc.rename(file1, file3);
+    } catch (IOException e) {
+      // Expected
+      assertTrue(unwrapException(e) instanceof FileNotFoundException);
+    }
+
+    // And if link points to a file...
+    createAndWriteFile(file2);
+    try {
+      fc.rename(file1, file3);
+    } catch (IOException e) {
+      // Expected
+      assertTrue(unwrapException(e) instanceof ParentNotDirectoryException);
+    }    
+  }
+  
   @Test
   /** Operate on a file using a path with an intermediate symlink */  
   public void testAccessFileViaSymlink() throws IOException {
@@ -1230,7 +1271,7 @@ public abstract class FileContextSymlinkBaseTest {
     Path dirViaLink = new Path(linkToDir, "dir");
     fc.createSymlink(baseDir, linkToDir, false);
     fc.mkdir(dirViaLink, FileContext.DEFAULT_PERM, true);
-    assertTrue(fc.getFileStatus(dirViaLink).isDir());
+    assertTrue(fc.getFileStatus(dirViaLink).isDirectory());
     FileStatus[] stats = fc.util().listStatus(dirViaLink);
     assertEquals(0, stats.length);
     Iterator<FileStatus> statsItor = fc.listStatus(dirViaLink);

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

@@ -124,7 +124,7 @@ public final class FileContextTestHelper {
   
   public static boolean isFile(FileContext fc, Path p) throws IOException {
     try {
-      return !fc.getFileStatus(p).isDir();
+      return fc.getFileStatus(p).isFile();
     } catch (FileNotFoundException e) {
       return false;
     }
@@ -132,7 +132,7 @@ public final class FileContextTestHelper {
 
   public static boolean isDir(FileContext fc, Path p) throws IOException {
     try {
-      return fc.getFileStatus(p).isDir();
+      return fc.getFileStatus(p).isDirectory();
     } catch (FileNotFoundException e) {
       return false;
     }

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

@@ -104,8 +104,7 @@ public class TestLocalFSFileContextSymlink extends FileContextSymlinkBaseTest {
     FileStatus fsd = fc.getFileLinkStatus(link);
     assertEquals(fileQual, fsd.getSymlink());
     assertTrue(fsd.isSymlink());
-    // NB: isDir is true since we need !isDir to imply file (HADOOP-6584)
-    //assertTrue(fsd.isDir());
+    assertFalse(fsd.isDirectory());
     assertEquals("", fsd.getOwner());
     assertEquals("", fsd.getGroup());
     assertEquals(link, fsd.getPath());

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

@@ -48,7 +48,7 @@ public class TestTrash extends TestCase {
   protected static Path mkdir(FileSystem fs, Path p) throws IOException {
     assertTrue(fs.mkdirs(p));
     assertTrue(fs.exists(p));
-    assertTrue(fs.getFileStatus(p).isDir());
+    assertTrue(fs.getFileStatus(p).isDirectory());
     return p;
   }
 

src/test/core/org/apache/hadoop/fs/loadGenerator/LoadGenerator.java

@@ -547,7 +547,7 @@ public class LoadGenerator extends Configured implements Tool {
     FileStatus[] stats = fs.listStatus(path);
 
     for (FileStatus stat : stats) {
-      if (stat.isDir()) {
+      if (stat.isDirectory()) {
         dirs.add(stat.getPath().toString());
         initFileDirTables(stat.getPath());
       } else {

src/test/core/org/apache/hadoop/fs/s3native/NativeS3FileSystemContractBaseTest.java

@@ -62,7 +62,7 @@ public abstract class NativeS3FileSystemContractBaseTest
   }
   
   public void testNoTrailingBackslashOnBucket() throws Exception {
-    assertTrue(fs.getFileStatus(new Path(fs.getUri().toString())).isDir());
+    assertTrue(fs.getFileStatus(new Path(fs.getUri().toString())).isDirectory());
   }
 
   private void createTestFiles(String base) throws IOException {
@@ -100,7 +100,7 @@ public abstract class NativeS3FileSystemContractBaseTest
         store.storeEmptyFile(base + "/dir/");
       }
 
-      assertTrue(fs.getFileStatus(path).isDir());
+      assertTrue(fs.getFileStatus(path).isDirectory());
       assertEquals(2, fs.listStatus(path).length);
     }
   }
@@ -114,7 +114,7 @@ public abstract class NativeS3FileSystemContractBaseTest
     fs.delete(path, true);
 
     path = path("/test");
-    assertTrue(fs.getFileStatus(path).isDir());
+    assertTrue(fs.getFileStatus(path).isDirectory());
     assertEquals(0, fs.listStatus(path).length);
   }
 
@@ -127,9 +127,9 @@ public abstract class NativeS3FileSystemContractBaseTest
     fs.rename(path("/" + base), dest);
 
     Path path = path("/test");
-    assertTrue(fs.getFileStatus(path).isDir());
+    assertTrue(fs.getFileStatus(path).isDirectory());
     assertEquals(1, fs.listStatus(path).length);
-    assertTrue(fs.getFileStatus(dest).isDir());
+    assertTrue(fs.getFileStatus(dest).isDirectory());
     assertEquals(2, fs.listStatus(dest).length);
   }
 

src/test/core/org/apache/hadoop/util/TestDiskChecker.java

@@ -107,13 +107,13 @@ public class TestDiskChecker {
         .returning(localDir).from.pathToFile(dir));
     FileStatus stat = make(stub(FileStatus.class)
         .returning(perm).from.getPermission());
-    when(stat.isDir()).thenReturn(isDir);
+    when(stat.isDirectory()).thenReturn(isDir);
     when(fs.getFileStatus(dir)).thenReturn(stat);
 
     try {
       DiskChecker.checkDir(fs, dir, perm);
 
-      verify(stat).isDir();
+      verify(stat).isDirectory();
       verify(fs, times(2)).getFileStatus(dir);
       verify(stat, times(2)).getPermission();
       assertTrue("checkDir success", success);