HADOOP-17032. Fix getContentSummary in ViewFileSystem to handle multiple children mountpoints pointing to different filesystems (#2060). Contributed by Abhishek Das.

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

@@ -1328,6 +1328,43 @@ public class ViewFileSystem extends FileSystem {
       return new FileStatus[0];
     }
 
+    @Override
+    public ContentSummary getContentSummary(Path f) throws IOException {
+      long[] summary = {0, 0, 1};
+      for (FileStatus status : listStatus(f)) {
+        Path targetPath =
+            Path.getPathWithoutSchemeAndAuthority(status.getPath());
+        InodeTree.ResolveResult<FileSystem> res =
+            fsState.resolve(targetPath.toString(), true);
+        ContentSummary child =
+            res.targetFileSystem.getContentSummary(res.remainingPath);
+        summary[0] += child.getLength();
+        summary[1] += child.getFileCount();
+        summary[2] += child.getDirectoryCount();
+      }
+      return new ContentSummary.Builder()
+          .length(summary[0])
+          .fileCount(summary[1])
+          .directoryCount(summary[2])
+          .build();
+    }
+
+    @Override
+    public FsStatus getStatus(Path p) throws IOException {
+      long[] summary = {0, 0, 0};
+      for (FileStatus status : listStatus(p)) {
+        Path targetPath =
+            Path.getPathWithoutSchemeAndAuthority(status.getPath());
+        InodeTree.ResolveResult<FileSystem> res =
+            fsState.resolve(targetPath.toString(), true);
+        FsStatus child = res.targetFileSystem.getStatus(res.remainingPath);
+        summary[0] += child.getCapacity();
+        summary[1] += child.getUsed();
+        summary[2] += child.getRemaining();
+      }
+      return new FsStatus(summary[0], summary[1], summary[2]);
+    }
+
     @Override
     public boolean mkdirs(Path dir, FsPermission permission)
         throws AccessControlException, FileAlreadyExistsException {

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

@@ -17,7 +17,9 @@
  */
 package org.apache.hadoop.fs.viewfs;
 
+import java.io.File;
 import java.io.FileNotFoundException;
+import java.io.FileOutputStream;
 import java.io.IOException;
 import java.net.URI;
 import java.security.PrivilegedExceptionAction;
@@ -32,6 +34,8 @@ import java.util.Random;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.BlockStoragePolicySpi;
+import org.apache.hadoop.fs.ContentSummary;
+import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.FileSystemTestHelper;
@@ -57,6 +61,8 @@ import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.test.GenericTestUtils;
 import org.junit.Assume;
+import org.junit.Rule;
+import org.junit.rules.TemporaryFolder;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import static org.apache.hadoop.fs.FileSystemTestHelper.*;
@@ -109,6 +115,9 @@ abstract public class ViewFileSystemBaseTest {
     return new FileSystemTestHelper();
   }
 
+  @Rule
+  public TemporaryFolder temporaryFolder = new TemporaryFolder();
+
   @Before
   public void setUp() throws Exception {
     initializeTargetTestRoot();
@@ -1369,4 +1378,56 @@ abstract public class ViewFileSystemBaseTest {
     viewFs.close();
     assertFalse(fsTarget.exists(realTestPath));
   }
+
+  @Test
+  public void testGetContentSummary() throws IOException {
+    ContentSummary summaryBefore =
+        fsView.getContentSummary(new Path("/internalDir"));
+    String expected = "GET CONTENT SUMMARY";
+    Path filePath =
+        new Path("/internalDir/internalDir2/linkToDir3", "foo");
+
+    try (FSDataOutputStream outputStream = fsView.create(filePath)) {
+      outputStream.write(expected.getBytes());
+    }
+
+    Path newDirPath = new Path("/internalDir/linkToDir2", "bar");
+    fsView.mkdirs(newDirPath);
+
+    ContentSummary summaryAfter =
+        fsView.getContentSummary(new Path("/internalDir"));
+    assertEquals("The file count didn't match",
+        summaryBefore.getFileCount() + 1,
+        summaryAfter.getFileCount());
+    assertEquals("The size didn't match",
+        summaryBefore.getLength() + expected.length(),
+        summaryAfter.getLength());
+    assertEquals("The directory count didn't match",
+        summaryBefore.getDirectoryCount() + 1,
+        summaryAfter.getDirectoryCount());
+  }
+
+  @Test
+  public void testGetContentSummaryWithFileInLocalFS() throws Exception {
+    ContentSummary summaryBefore =
+        fsView.getContentSummary(new Path("/internalDir"));
+    String expected = "GET CONTENT SUMMARY";
+    File localFile = temporaryFolder.newFile("localFile");
+    try (FileOutputStream fos = new FileOutputStream(localFile)) {
+      fos.write(expected.getBytes());
+    }
+    ConfigUtil.addLink(conf,
+        "/internalDir/internalDir2/linkToLocalFile", localFile.toURI());
+
+    try (FileSystem fs = FileSystem.get(FsConstants.VIEWFS_URI, conf)) {
+      ContentSummary summaryAfter =
+          fs.getContentSummary(new Path("/internalDir"));
+      assertEquals("The file count didn't match",
+          summaryBefore.getFileCount() + 1,
+          summaryAfter.getFileCount());
+      assertEquals("The directory count didn't match",
+          summaryBefore.getLength() + expected.length(),
+          summaryAfter.getLength());
+    }
+  }
 }