HADOOP-17029. Return correct permission and owner for listing on internal directories in ViewFs. Contributed by Abhishek Das.

(cherry picked from commit e7dd02768b)

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

@@ -1161,13 +1161,26 @@ public class ViewFileSystem extends FileSystem {
         INode<FileSystem> inode = iEntry.getValue();
         if (inode.isLink()) {
           INodeLink<FileSystem> link = (INodeLink<FileSystem>) inode;
-
+          try {
+            String linkedPath = link.getTargetFileSystem().getUri().getPath();
+            FileStatus status =
+                ((ChRootedFileSystem)link.getTargetFileSystem())
+                .getMyFs().getFileStatus(new Path(linkedPath));
+            result[i++] = new FileStatus(status.getLen(), false,
+              status.getReplication(), status.getBlockSize(),
+              status.getModificationTime(), status.getAccessTime(),
+              status.getPermission(), status.getOwner(), status.getGroup(),
+              link.getTargetLink(),
+              new Path(inode.fullPath).makeQualified(
+                  myUri, null));
+          } catch (FileNotFoundException ex) {
             result[i++] = new FileStatus(0, false, 0, 0,
               creationTime, creationTime, PERMISSION_555,
               ugi.getShortUserName(), ugi.getPrimaryGroupName(),
               link.getTargetLink(),
               new Path(inode.fullPath).makeQualified(
                   myUri, null));
+          }
         } else {
           result[i++] = new FileStatus(0, true, 0, 0,
             creationTime, creationTime, PERMISSION_555,

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

@@ -910,11 +910,25 @@ public class ViewFs extends AbstractFileSystem {
       if (inode.isLink()) {
         INodeLink<AbstractFileSystem> inodelink = 
           (INodeLink<AbstractFileSystem>) inode;
+        try {
+          String linkedPath = inodelink.getTargetFileSystem()
+              .getUri().getPath();
+          FileStatus status = ((ChRootedFs)inodelink.getTargetFileSystem())
+              .getMyFs().getFileStatus(new Path(linkedPath));
+          result = new FileStatus(status.getLen(), false,
+            status.getReplication(), status.getBlockSize(),
+            status.getModificationTime(), status.getAccessTime(),
+            status.getPermission(), status.getOwner(), status.getGroup(),
+            inodelink.getTargetLink(),
+            new Path(inode.fullPath).makeQualified(
+                myUri, null));
+        } catch (FileNotFoundException ex) {
           result = new FileStatus(0, false, 0, 0, creationTime, creationTime,
             PERMISSION_555, ugi.getShortUserName(), ugi.getPrimaryGroupName(),
             inodelink.getTargetLink(),
             new Path(inode.fullPath).makeQualified(
                 myUri, null));
+        }
       } else {
         result = new FileStatus(0, true, 0, 0, creationTime, creationTime,
           PERMISSION_555, ugi.getShortUserName(), ugi.getPrimaryGroupName(),
@@ -969,12 +983,25 @@ public class ViewFs extends AbstractFileSystem {
           INodeLink<AbstractFileSystem> link = 
             (INodeLink<AbstractFileSystem>) inode;
 
-          result[i++] = new FileStatus(0, false, 0, 0,
+          try {
-            creationTime, creationTime,
+            String linkedPath = link.getTargetFileSystem().getUri().getPath();
-            PERMISSION_555, ugi.getShortUserName(), ugi.getPrimaryGroupName(),
+            FileStatus status = ((ChRootedFs)link.getTargetFileSystem())
+                .getMyFs().getFileStatus(new Path(linkedPath));
+            result[i++] = new FileStatus(status.getLen(), false,
+              status.getReplication(), status.getBlockSize(),
+              status.getModificationTime(), status.getAccessTime(),
+              status.getPermission(), status.getOwner(), status.getGroup(),
               link.getTargetLink(),
               new Path(inode.fullPath).makeQualified(
                   myUri, null));
+          } catch (FileNotFoundException ex) {
+            result[i++] = new FileStatus(0, false, 0, 0,
+              creationTime, creationTime, PERMISSION_555,
+              ugi.getShortUserName(), ugi.getPrimaryGroupName(),
+              link.getTargetLink(),
+              new Path(inode.fullPath).makeQualified(
+                  myUri, null));
+          }
         } else {
           result[i++] = new FileStatus(0, true, 0, 0,
             creationTime, creationTime,

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

@@ -29,10 +29,13 @@ import org.apache.hadoop.fs.FileUtil;
 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;
+import org.junit.After;
 import org.junit.AfterClass;
+import org.junit.Before;
 import org.junit.Test;
 import org.mockito.Mockito;
 
@@ -48,6 +51,17 @@ public class TestViewfsFileStatus {
   private static final File TEST_DIR = GenericTestUtils.getTestDir(
       TestViewfsFileStatus.class.getSimpleName());
 
+  @Before
+  public void setUp() {
+    FileUtil.fullyDelete(TEST_DIR);
+    assertTrue(TEST_DIR.mkdirs());
+  }
+
+  @After
+  public void tearDown() throws IOException {
+    FileUtil.fullyDelete(TEST_DIR);
+  }
+
   @Test
   public void testFileStatusSerialziation()
       throws IOException, URISyntaxException {
@@ -56,20 +70,14 @@ public class TestViewfsFileStatus {
     File infile = new File(TEST_DIR, testfilename);
     final byte[] content = "dingos".getBytes();
 
-    FileOutputStream fos = null;
+    try (FileOutputStream fos =  new FileOutputStream(infile)) {
-    try {
-      fos = new FileOutputStream(infile);
       fos.write(content);
-    } finally {
-      if (fos != null) {
-        fos.close();
-      }
     }
     assertEquals((long)content.length, infile.length());
 
     Configuration conf = new Configuration();
     ConfigUtil.addLink(conf, "/foo/bar/baz", TEST_DIR.toURI());
-    FileSystem vfs = FileSystem.get(FsConstants.VIEWFS_URI, conf);
+    try (FileSystem vfs = FileSystem.get(FsConstants.VIEWFS_URI, conf)) {
       assertEquals(ViewFileSystem.class, vfs.getClass());
       Path path = new Path("/foo/bar/baz", testfilename);
       FileStatus stat = vfs.getFileStatus(path);
@@ -89,6 +97,64 @@ public class TestViewfsFileStatus {
       assertEquals(content.length, deSer.getLen());
       assertFalse(deSer.isErasureCoded());
     }
+  }
+
+  /**
+   * Tests the ACL returned from getFileStatus for directories and files.
+   * @throws IOException
+   */
+  @Test
+  public void testListStatusACL() throws IOException {
+    String testfilename = "testFileACL";
+    String childDirectoryName = "testDirectoryACL";
+    TEST_DIR.mkdirs();
+    File infile = new File(TEST_DIR, testfilename);
+    final byte[] content = "dingos".getBytes();
+
+    try (FileOutputStream fos =  new FileOutputStream(infile)) {
+      fos.write(content);
+    }
+    assertEquals(content.length, infile.length());
+    File childDir = new File(TEST_DIR, childDirectoryName);
+    childDir.mkdirs();
+
+    Configuration conf = new Configuration();
+    ConfigUtil.addLink(conf, "/file", infile.toURI());
+    ConfigUtil.addLink(conf, "/dir", childDir.toURI());
+
+    try (FileSystem vfs = FileSystem.get(FsConstants.VIEWFS_URI, conf)) {
+      assertEquals(ViewFileSystem.class, vfs.getClass());
+      FileStatus[] statuses = vfs.listStatus(new Path("/"));
+
+      FileSystem localFs = FileSystem.getLocal(conf);
+      FileStatus fileStat = localFs.getFileStatus(new Path(infile.getPath()));
+      FileStatus dirStat = localFs.getFileStatus(new Path(childDir.getPath()));
+
+      for (FileStatus status : statuses) {
+        if (status.getPath().getName().equals("file")) {
+          assertEquals(fileStat.getPermission(), status.getPermission());
+        } else {
+          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());
+        }
+      }
+    }
+  }
 
   // Tests that ViewFileSystem.getFileChecksum calls res.targetFileSystem
   // .getFileChecksum with res.remainingPath and not with f