HDFS-2514. svn merge -c 1204370 from trunk

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

hadoop-hdfs-project/hadoop-hdfs/CHANGES.txt

@@ -1831,6 +1831,9 @@ Release 0.22.0 - Unreleased
     HDFS-2002. Incorrect computation of needed blocks in getTurnOffTip().
     (Plamen Jeliazkov via shv)
 
+    HDFS-2514. Link resolution bug for intermediate symlinks with
+    relative targets. (eli)
+
 Release 0.21.1 - Unreleased
 
     HDFS-1466. TestFcHdfsSymlink relies on /tmp/test not existing. (eli)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/ClientProtocol.java

@@ -813,9 +813,9 @@ public void setTimes(String src, long mtime, long atime)
 
   /**
    * Create symlink to a file or directory.
-   * @param target The pathname of the destination that the
+   * @param target The path of the destination that the
    *               link points to.
-   * @param link The pathname of the link being created.
+   * @param link The path of the link being created.
    * @param dirPerm permissions to use when creating parent directories
    * @param createParent - if true then missing parent dirs are created
    *                       if false then parent must exist
@@ -836,14 +836,16 @@ public void createSymlink(String target, String link, FsPermission dirPerm,
       IOException;
 
   /**
-   * Resolve the first symbolic link on the specified path.
+   * Return the target of the given symlink. If there is an intermediate
-   * @param path The pathname that needs to be resolved
+   * symlink in the path (ie a symlink leading up to the final path component)
-   * 
+   * then the given path is returned with this symlink resolved.
-   * @return The pathname after resolving the first symbolic link if any.
+   *
-   * 
+   * @param path The path with a link that needs resolution.
+   * @return The path after resolving the first symbolic link in the path.
    * @throws AccessControlException permission denied
    * @throws FileNotFoundException If <code>path</code> does not exist
-   * @throws IOException If an I/O error occurred
+   * @throws IOException If the given path does not refer to a symlink
+   *           or an I/O error occurred
    */
   public String getLinkTarget(String path) throws AccessControlException,
       FileNotFoundException, IOException; 

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/protocol/UnresolvedPathException.java

@@ -32,10 +32,10 @@
 @InterfaceStability.Evolving
 public final class UnresolvedPathException extends UnresolvedLinkException {
   private static final long serialVersionUID = 1L;
-  private String originalPath;  // The original path containing the link
+  private String path;        // The path containing the link
-  private String linkTarget;    // The target of the link 
+  private String preceding;   // The path part preceding the link
-  private String remainingPath; // The path part following the link
+  private String remainder;   // The path part following the link
-  
+  private String linkTarget;  // The link's target
 
   /**
    * Used by RemoteException to instantiate an UnresolvedPathException.
@@ -44,22 +44,30 @@ public UnresolvedPathException(String msg) {
     super(msg);
   }
   
-  public UnresolvedPathException(String originalPath, String remainingPath, 
+  public UnresolvedPathException(String path, String preceding,
-      String linkTarget) {
+      String remainder, String linkTarget) {
-    this.originalPath  = originalPath;
+    this.path = path;
-    this.remainingPath = remainingPath;
+    this.preceding = preceding;
-    this.linkTarget    = linkTarget;
+    this.remainder = remainder;
+    this.linkTarget = linkTarget;
   }
 
-  public Path getUnresolvedPath() throws IOException {
+  /**
-    return new Path(originalPath);
+   * Return a path with the link resolved with the target.
-  }
+   */
-  
   public Path getResolvedPath() throws IOException {
-    if (remainingPath == null || "".equals(remainingPath)) {
+    // If the path is absolute we cam throw out the preceding part and
-      return new Path(linkTarget);
+    // just append the remainder to the target, otherwise append each
+    // piece to resolve the link in path.
+    boolean noRemainder = (remainder == null || "".equals(remainder));
+    Path target = new Path(linkTarget);
+    if (target.isUriPathAbsolute()) {
+      return noRemainder ? target : new Path(target, remainder);
+    } else {
+      return noRemainder
+        ? new Path(preceding, target)
+        : new Path(new Path(preceding, linkTarget), remainder);
     }
-    return new Path(linkTarget, remainingPath);
   }
 
   @Override
@@ -68,7 +76,7 @@ public String getMessage() {
     if (msg != null) {
       return msg;
     }
-    String myMsg = "Unresolved path " + originalPath;
+    String myMsg = "Unresolved path " + path;
     try {
       return getResolvedPath().toString();
     } catch (IOException e) {

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -1976,10 +1976,12 @@ void removePathAndBlocks(String src, List<Block> blocks) {
     }
   }
 
-  /** Get the file info for a specific file.
+  /**
+   * Get the file info for a specific file.
+   *
    * @param src The string representation of the path to the file
    * @param resolveLink whether to throw UnresolvedLinkException 
-   *        if src refers to a symlinks
+   *        if src refers to a symlink
    *
    * @throws AccessControlException if access is denied
    * @throws UnresolvedLinkException if a symlink is encountered.

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INode.java

@@ -372,14 +372,16 @@ static String[] getPathNames(String path) {
 
   /**
    * Given some components, create a path name.
-   * @param components
+   * @param components The path components
+   * @param start index
+   * @param end index
    * @return concatenated path
    */
-  static String constructPath(byte[][] components, int start) {
+  static String constructPath(byte[][] components, int start, int end) {
     StringBuilder buf = new StringBuilder();
-    for (int i = start; i < components.length; i++) {
+    for (int i = start; i < end; i++) {
       buf.append(DFSUtil.bytes2String(components[i]));
-      if (i < components.length - 1) {
+      if (i < end - 1) {
         buf.append(Path.SEPARATOR);
       }
     }

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/INodeDirectory.java

@@ -191,18 +191,19 @@ assert compareBytes(this.name, components[0]) == 0 :
         existing[index] = curNode;
       }
       if (curNode.isLink() && (!lastComp || (lastComp && resolveLink))) {
-        if(NameNode.stateChangeLog.isDebugEnabled()) {
+        final String path = constructPath(components, 0, components.length);
+        final String preceding = constructPath(components, 0, count);
+        final String remainder =
+          constructPath(components, count + 1, components.length);
+        final String link = DFSUtil.bytes2String(components[count]);
+        final String target = ((INodeSymlink)curNode).getLinkValue();
+        if (NameNode.stateChangeLog.isDebugEnabled()) {
           NameNode.stateChangeLog.debug("UnresolvedPathException " +
-              " count: " + count +
+            " path: " + path + " preceding: " + preceding +
-              " componenent: " + DFSUtil.bytes2String(components[count]) +
+            " count: " + count + " link: " + link + " target: " + target +
-              " full path: " + constructPath(components, 0) +
+            " remainder: " + remainder);
-              " remaining path: " + constructPath(components, count+1) +
-              " symlink: " + ((INodeSymlink)curNode).getLinkValue());
         }
-        final String linkTarget = ((INodeSymlink)curNode).getLinkValue();
+        throw new UnresolvedPathException(path, preceding, remainder, target);
-        throw new UnresolvedPathException(constructPath(components, 0),
-                                          constructPath(components, count+1),
-                                          linkTarget);
       }
       if (lastComp || !curNode.isDirectory()) {
         break;

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/NameNodeRpcServer.java

@@ -738,10 +738,6 @@ public void createSymlink(String target, String link, FsPermission dirPerms,
   @Override // ClientProtocol
   public String getLinkTarget(String path) throws IOException {
     metrics.incrGetLinkTargetOps();
-    /* Resolves the first symlink in the given path, returning a
-     * new path consisting of the target of the symlink and any 
-     * remaining path components from the original path.
-     */
     try {
       HdfsFileStatus stat = namesystem.getFileInfo(path, false);
       if (stat != null) {

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/fs/TestFcHdfsSymlink.java

@@ -20,6 +20,9 @@
 import java.io.*;
 import java.net.URI;
 
+import org.apache.commons.logging.impl.Log4JLogger;
+import org.apache.log4j.Level;
+
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileContext;
@@ -28,9 +31,11 @@
 import org.apache.hadoop.hdfs.DFSConfigKeys;
 import org.apache.hadoop.hdfs.HdfsConfiguration;
 import org.apache.hadoop.hdfs.MiniDFSCluster;
+import org.apache.hadoop.hdfs.server.namenode.NameNode;
 import org.apache.hadoop.hdfs.protocol.HdfsConstants;
 import static org.apache.hadoop.fs.FileContextTestHelper.*;
 import org.apache.hadoop.ipc.RemoteException;
+
 import static org.junit.Assert.*;
 import org.junit.Test;
 import org.junit.BeforeClass;
@@ -41,6 +46,10 @@
  */
 public class TestFcHdfsSymlink extends FileContextSymlinkBaseTest {
 
+  {
+    ((Log4JLogger)NameNode.stateChangeLog).getLogger().setLevel(Level.ALL);
+  }
+
   private static MiniDFSCluster cluster;
   
   protected String getScheme() {
@@ -250,8 +259,8 @@ public void testLinkOwner() throws IOException {
     Path link = new Path(testBaseDir1(), "symlinkToFile");
     createAndWriteFile(file);
     fc.createSymlink(file, link, false);
-    FileStatus stat_file = fc.getFileStatus(file);
+    FileStatus statFile = fc.getFileStatus(file);
-    FileStatus stat_link = fc.getFileStatus(link);
+    FileStatus statLink = fc.getFileStatus(link);
-    assertEquals(stat_link.getOwner(), stat_file.getOwner());
+    assertEquals(statLink.getOwner(), statFile.getOwner());
   }
 }