HADOOP-18129: Change URI to String in INodeLink to reduce memory footprint of ViewFileSystem

Fixes #3996
Fixes #4083

(cherry picked from commit da9970dd69)
Signed-off-by: Owen O'Malley <oomalley@linkedin.com>

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

@@ -230,7 +230,7 @@ abstract class InodeTree<T> {
    * is changed later it is then ignored (a dir with null entries)
    */
   static class INodeLink<T> extends INode<T> {
-    final URI[] targetDirLinkList;
+    final String[] targetDirLinkList;
     private T targetFileSystem;   // file system object created from the link.
     // Function to initialize file system. Only applicable for simple links
     private Function<URI, T> fileSystemInitMethod;
@@ -240,7 +240,7 @@ abstract class InodeTree<T> {
      * Construct a mergeLink or nfly.
      */
     INodeLink(final String pathToNode, final UserGroupInformation aUgi,
-        final T targetMergeFs, final URI[] aTargetDirLinkList) {
+        final T targetMergeFs, final String[] aTargetDirLinkList) {
       super(pathToNode, aUgi);
       targetFileSystem = targetMergeFs;
       targetDirLinkList = aTargetDirLinkList;
@@ -251,11 +251,11 @@ abstract class InodeTree<T> {
      */
     INodeLink(final String pathToNode, final UserGroupInformation aUgi,
         Function<URI, T> createFileSystemMethod,
-        final URI aTargetDirLink) {
+        final String aTargetDirLink) throws URISyntaxException {
       super(pathToNode, aUgi);
       targetFileSystem = null;
-      targetDirLinkList = new URI[1];
-      targetDirLinkList[0] = aTargetDirLink;
+      targetDirLinkList = new String[1];
+      targetDirLinkList[0] = new URI(aTargetDirLink).toString();
       this.fileSystemInitMethod = createFileSystemMethod;
     }
 
@@ -293,7 +293,8 @@ abstract class InodeTree<T> {
           if (targetFileSystem != null) {
             return targetFileSystem;
           }
-          targetFileSystem = fileSystemInitMethod.apply(targetDirLinkList[0]);
+          targetFileSystem =
+              fileSystemInitMethod.apply(URI.create(targetDirLinkList[0]));
           if (targetFileSystem == null) {
             throw new IOException(
                 "Could not initialize target File System for URI : " +
@@ -361,7 +362,7 @@ abstract class InodeTree<T> {
     switch (linkType) {
     case SINGLE:
       newLink = new INodeLink<T>(fullPath, aUgi,
-          initAndGetTargetFs(), new URI(target));
+          initAndGetTargetFs(), target);
       break;
     case SINGLE_FALLBACK:
     case MERGE_SLASH:
@@ -370,10 +371,10 @@ abstract class InodeTree<T> {
       throw new IllegalArgumentException("Unexpected linkType: " + linkType);
     case MERGE:
     case NFLY:
-      final URI[] targetUris = StringUtils.stringToURI(
-          StringUtils.getStrings(target));
+      final String[] targetUris = StringUtils.getStrings(target);
       newLink = new INodeLink<T>(fullPath, aUgi,
-            getTargetFileSystem(settings, targetUris), targetUris);
+          getTargetFileSystem(settings, StringUtils.stringToURI(targetUris)),
+          targetUris);
       break;
     default:
       throw new IllegalArgumentException(linkType + ": Infeasible linkType");
@@ -582,8 +583,7 @@ abstract class InodeTree<T> {
     if (isMergeSlashConfigured) {
       Preconditions.checkNotNull(mergeSlashTarget);
       root = new INodeLink<T>(mountTableName, ugi,
-          initAndGetTargetFs(),
-          new URI(mergeSlashTarget));
+          initAndGetTargetFs(), mergeSlashTarget);
       mountPoints.add(new MountPoint<T>("/", (INodeLink<T>) root));
       rootFallbackLink = null;
     } else {
@@ -600,8 +600,7 @@ abstract class InodeTree<T> {
                 + "not allowed.");
           }
           fallbackLink = new INodeLink<T>(mountTableName, ugi,
-              initAndGetTargetFs(),
-              new URI(le.getTarget()));
+              initAndGetTargetFs(), le.getTarget());
         } else {
           createLink(le.getSrc(), le.getTarget(), le.getLinkType(),
               le.getSettings(), le.getUgi(), le.getConfig());

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

@@ -172,16 +172,25 @@ public class ViewFileSystem extends FileSystem {
      * One or more targets of the mount.
      * Multiple targets imply MergeMount.
      */
-    private URI[] targets;
+    private String[] targets;
 
-    MountPoint(Path srcPath, URI[] targetURIs) {
+    MountPoint(Path srcPath, String[] targetURIs) {
       src = srcPath;
       targets = targetURIs;
     }
     Path getSrc() {
       return src;
     }
+
     URI[] getTargets() {
+      URI[] targetUris = new URI[targets.length];
+      for (int i = 0; i < targets.length; i++) {
+        targetUris[i] = URI.create(targets[i]);
+      }
+      return targetUris;
+    }
+
+    public String[] getTargetFileSystemPaths() {
       return targets;
     }
   }

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

@@ -178,16 +178,18 @@ public class ViewFs extends AbstractFileSystem {
   
   
   static public class MountPoint {
-    private Path src;       // the src of the mount
-    private URI[] targets; //  target of the mount; Multiple targets imply mergeMount
-    MountPoint(Path srcPath, URI[] targetURIs) {
+    // the src of the mount
+    private Path src;
+    // Target of the mount; Multiple targets imply mergeMount
+    private String[] targets;
+    MountPoint(Path srcPath, String[] targetURIs) {
       src = srcPath;
       targets = targetURIs;
     }
     Path getSrc() {
       return src;
     }
-    URI[] getTargets() {
+    String[] getTargets() {
       return targets;
     }
   }

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

@@ -1392,4 +1392,22 @@ abstract public class ViewFileSystemBaseTest {
     // viewfs inner cache is disabled
     assertEquals(cacheSize + 1, TestFileUtil.getCacheSize());
   }
+
+  @Test
+  public void testInvalidMountPoints() throws Exception {
+    final String clusterName = "cluster" + new Random().nextInt();
+    Configuration config = new Configuration(conf);
+    config.set(ConfigUtil.getConfigViewFsPrefix(clusterName) + "." +
+        Constants.CONFIG_VIEWFS_LINK + "." + "/invalidPath",
+        "othermockfs:|mockauth/mockpath");
+
+    try {
+      FileSystem viewFs = FileSystem.get(
+          new URI("viewfs://" + clusterName + "/"), config);
+      fail("FileSystem should not initialize. Should fail with IOException");
+    } catch (IOException ex) {
+      assertTrue("Should get URISyntax Exception",
+          ex.getMessage().startsWith("URISyntax exception"));
+    }
+  }
 }