HADOOP-17029. Addressed code review comments.

Author: abhishekdas99

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

@@ -1200,14 +1200,9 @@ public FileStatus[] listStatus(Path f) throws AccessControlException,
         INode<FileSystem> inode = iEntry.getValue();
         if (inode.isLink()) {
           INodeLink<FileSystem> link = (INodeLink<FileSystem>) inode;
-          // For MERGE or NFLY links, the first target link is considered
-          // for fetching the FileStatus with an assumption that the permission
-          // and the owner will be the same for all the target directories.
-          Path linkedPath = new Path(link.targetDirLinkList[0].toString());
-          ChRootedFileSystem linkedFs = (ChRootedFileSystem)
-              link.getTargetFileSystem();
           try {
-            FileStatus status = linkedFs.getMyFs().getFileStatus(linkedPath);
+            FileStatus status = link.getTargetFileSystem()
+                .getFileStatus(new Path("/"));
             result[i++] = new FileStatus(status.getLen(), false,
               status.getReplication(), status.getBlockSize(),
               status.getModificationTime(), status.getAccessTime(),

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

@@ -917,10 +917,9 @@ public FileStatus getFileLinkStatus(final Path f)
       if (inode.isLink()) {
         INodeLink<AbstractFileSystem> inodelink = 
           (INodeLink<AbstractFileSystem>) inode;
-        Path linkedPath = new Path(inodelink.targetDirLinkList[0].toString());
-        ChRootedFs linkedFs = (ChRootedFs) inodelink.getTargetFileSystem();
         try {
-          FileStatus status = linkedFs.getMyFs().getFileStatus(linkedPath);
+          FileStatus status = inodelink.getTargetFileSystem()
+              .getFileStatus(new Path("/"));
           result = new FileStatus(status.getLen(), false,
             status.getReplication(), status.getBlockSize(),
             status.getModificationTime(), status.getAccessTime(),
@@ -989,10 +988,9 @@ public FileStatus[] listStatus(final Path f) throws AccessControlException,
           INodeLink<AbstractFileSystem> link = 
             (INodeLink<AbstractFileSystem>) inode;
 
-          Path linkedPath = new Path(link.targetDirLinkList[0].toString());
-          ChRootedFs linkedFs = (ChRootedFs) link.getTargetFileSystem();
           try {
-            FileStatus status = linkedFs.getMyFs().getFileStatus(linkedPath);
+            FileStatus status = link.getTargetFileSystem()
+                .getFileStatus(new Path("/"));
             result[i++] = new FileStatus(status.getLen(), false,
               status.getReplication(), status.getBlockSize(),
               status.getModificationTime(), status.getAccessTime(),

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

@@ -29,6 +29,7 @@
 import org.apache.hadoop.fs.FsConstants;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.contract.ContractTestUtils;
+import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.io.DataInputBuffer;
 import org.apache.hadoop.io.DataOutputBuffer;
 import org.apache.hadoop.test.GenericTestUtils;
@@ -98,9 +99,12 @@ public void testFileStatusSerialziation()
     }
   }
 
+  /**
+   * Tests the ACL returned from getFileStatus for directories and files.
+   * @throws IOException
+   */
   @Test
-  public void testListStatusACL()
-      throws IOException, URISyntaxException {
+  public void testListStatusACL() throws IOException {
     String testfilename = "testFileACL";
     String childDirectoryName = "testDirectoryACL";
     TEST_DIR.mkdirs();
@@ -110,7 +114,7 @@ public void testListStatusACL()
     try (FileOutputStream fos =  new FileOutputStream(infile)) {
       fos.write(content);
     }
-    assertEquals((long)content.length, infile.length());
+    assertEquals(content.length, infile.length());
     File childDir = new File(TEST_DIR, childDirectoryName);
     childDir.mkdirs();
 
@@ -133,6 +137,22 @@ public void testListStatusACL()
           assertEquals(dirStat.getPermission(), status.getPermission());
         }
       }
+
+      localFs.setPermission(new Path(infile.getPath()),
+          FsPermission.valueOf("-rwxr--r--"));
+      localFs.setPermission(new Path(childDir.getPath()),
+          FsPermission.valueOf("-r--rwxr--"));
+
+      statuses = vfs.listStatus(new Path("/"));
+      for (FileStatus status : statuses) {
+        if (status.getPath().getName().equals("file")) {
+          assertEquals(FsPermission.valueOf("-rwxr--r--"),
+              status.getPermission());
+        } else {
+          assertEquals(FsPermission.valueOf("-r--rwxr--"),
+              status.getPermission());
+        }
+      }
     }
   }