HADOOP-19130. FTPFileSystem rename with full qualified path broken (#6678). Contributed by shawn

Signed-off-by: Ayush Saxena <ayushsaxena@apache.org>

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/ftp/FTPFileSystem.java

@@ -685,21 +685,21 @@ public class FTPFileSystem extends FileSystem {
       throw new FileAlreadyExistsException("Destination path " + dst
           + " already exists");
     }
-    String parentSrc = absoluteSrc.getParent().toUri().toString();
-    String parentDst = absoluteDst.getParent().toUri().toString();
+    URI parentSrc = absoluteSrc.getParent().toUri();
+    URI parentDst = absoluteDst.getParent().toUri();
     if (isParentOf(absoluteSrc, absoluteDst)) {
       throw new IOException("Cannot rename " + absoluteSrc + " under itself"
       + " : "+ absoluteDst);
     }
 
-    if (!parentSrc.equals(parentDst)) {
+    if (!parentSrc.toString().equals(parentDst.toString())) {
       throw new IOException("Cannot rename source: " + absoluteSrc
           + " to " + absoluteDst
           + " -"+ E_SAME_DIRECTORY_ONLY);
     }
     String from = absoluteSrc.getName();
     String to = absoluteDst.getName();
-    client.changeWorkingDirectory(parentSrc);
+    client.changeWorkingDirectory(parentSrc.getPath());
     boolean renamed = client.rename(from, to);
     return renamed;
   }

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/ftp/TestFTPFileSystem.java

@@ -48,6 +48,7 @@ import org.junit.rules.Timeout;
 import static org.hamcrest.CoreMatchers.equalTo;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertTrue;
 
 /**
  * Test basic @{link FTPFileSystem} class methods. Contract tests are in
@@ -236,4 +237,50 @@ public class TestFTPFileSystem {
     ftp.setTimeout(client, conf);
     assertEquals(client.getControlKeepAliveTimeout(), timeout);
   }
-}
+
+  private static void touch(FileSystem fs, Path filePath)
+          throws IOException {
+    touch(fs, filePath, null);
+  }
+
+  private static void touch(FileSystem fs, Path path, byte[] data)
+          throws IOException {
+    FSDataOutputStream out = null;
+    try {
+      out = fs.create(path);
+      if (data != null) {
+        out.write(data);
+      }
+    } finally {
+      if (out != null) {
+        out.close();
+      }
+    }
+  }
+
+  /**
+   * Test renaming a file.
+   *
+   * @throws Exception
+   */
+  @Test
+  public void testRenameFileWithFullQualifiedPath() throws Exception {
+    BaseUser user = server.addUser("test", "password", new WritePermission());
+    Configuration configuration = new Configuration();
+    configuration.set("fs.defaultFS", "ftp:///");
+    configuration.set("fs.ftp.host", "localhost");
+    configuration.setInt("fs.ftp.host.port", server.getPort());
+    configuration.set("fs.ftp.user.localhost", user.getName());
+    configuration.set("fs.ftp.password.localhost", user.getPassword());
+    configuration.setBoolean("fs.ftp.impl.disable.cache", true);
+
+    FileSystem fs = FileSystem.get(configuration);
+
+
+    Path ftpDir = fs.makeQualified(new Path(testDir.toAbsolutePath().toString()));
+    Path file1 = fs.makeQualified(new Path(ftpDir, "renamefile" + "1"));
+    Path file2 = fs.makeQualified(new Path(ftpDir, "renamefile" + "2"));
+    touch(fs, file1);
+    assertTrue(fs.rename(file1, file2));
+  }
+}