HADOOP-17742. fix distcp fail when copying to ftp filesystem

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

@@ -72,8 +72,6 @@ public class FTPFileSystem extends FileSystem {
   public static final String FS_FTP_DATA_CONNECTION_MODE =
       "fs.ftp.data.connection.mode";
   public static final String FS_FTP_TRANSFER_MODE = "fs.ftp.transfer.mode";
-  public static final String E_SAME_DIRECTORY_ONLY =
-      "only same directory renames are supported";
   public static final String FS_FTP_TIMEOUT = "fs.ftp.timeout";
 
   private URI uri;
@@ -625,8 +623,8 @@ private boolean isFile(FTPClient client, Path file) {
   }
 
   /*
-   * Assuming that parent of both source and destination is the same. Is the
-   * assumption correct or it is suppose to work like 'move' ?
+   * The parent of source and destination can be different. It is suppose to
+   * work like 'move'
    */
   @Override
   public boolean rename(Path src, Path dst) throws IOException {
@@ -685,21 +683,15 @@ private boolean rename(FTPClient client, Path src, Path dst)
       throw new FileAlreadyExistsException("Destination path " + dst
           + " already exists");
     }
-    String parentSrc = absoluteSrc.getParent().toUri().toString();
-    String parentDst = absoluteDst.getParent().toUri().toString();
-    if (isParentOf(absoluteSrc, absoluteDst)) {
-      throw new IOException("Cannot rename " + absoluteSrc + " under itself"
-      + " : "+ absoluteDst);
-    }
 
-    if (!parentSrc.equals(parentDst)) {
-      throw new IOException("Cannot rename source: " + absoluteSrc
-          + " to " + absoluteDst
-          + " -"+ E_SAME_DIRECTORY_ONLY);
+    Path dstParentPath = absoluteDst.getParent();
+    if (!exists(dstParentPath)) {
+      // when the parent path of dst don't exist, the path need to create.
+      mkdirs(dstParentPath);
     }
-    String from = absoluteSrc.getName();
-    String to = absoluteDst.getName();
-    client.changeWorkingDirectory(parentSrc);
+
+    String from = absoluteSrc.toUri().getPath();
+    String to = absoluteDst.toUri().getPath();
     boolean renamed = client.rename(from, to);
     return renamed;
   }

hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/DistCpConstants.java

@@ -192,4 +192,15 @@ private DistCpConstants() {
 
   public static final String CLASS_INSTANTIATION_ERROR_MSG =
           "Unable to instantiate ";
+
+  /**
+   * The prefix of target temp file path.
+   */
+  public static final String TARGET_TEMP_FILE_PREFIX_DOT = ".distcp.tmp.";
+
+  /**
+   * FTP filesystem can't be work well when the path with dot prefix,
+   * so the target tmp file path use the prefix without dot.
+   */
+  public static final String TARGET_TEMP_FILE_PREFIX_FTP = "distcp.tmp.";
 }

hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/CopyCommitter.java

@@ -170,8 +170,10 @@ private void deleteAttemptTempFiles(Path targetWorkPath,
       return;
     }
 
+    String tempFilePrefix = DistCpUtils.getTargetTempFilePrefix(targetWorkPath);
     FileStatus[] tempFiles = targetFS.globStatus(
-        new Path(targetWorkPath, ".distcp.tmp." + jobId.replaceAll("job","attempt") + "*"));
+        new Path(targetWorkPath, tempFilePrefix
+                + jobId.replaceAll("job", "attempt") + "*"));
 
     if (tempFiles != null && tempFiles.length > 0) {
       for (FileStatus file : tempFiles) {

hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/mapred/RetriableFileCopyCommand.java

@@ -129,16 +129,15 @@ private long doCopy(CopyListingFileStatus source, Path target,
       throws IOException {
     LOG.info("Copying {} to {}", source.getPath(), target);
 
+    final Configuration configuration = context.getConfiguration();
+    FileSystem targetFS = target.getFileSystem(configuration);
     final boolean toAppend = action == FileAction.APPEND;
     final boolean useTempTarget = !toAppend && !directWrite;
-    Path targetPath = useTempTarget ? getTempFile(target, context) : target;
+    Path targetPath = useTempTarget ? getTempFile(target, context, targetFS) : target;
 
     LOG.info("Writing to {} target file path {}", useTempTarget ? "temporary"
         : "direct", targetPath);
 
-    final Configuration configuration = context.getConfiguration();
-    FileSystem targetFS = target.getFileSystem(configuration);
-
     try {
       final Path sourcePath = source.getPath();
       final FileSystem sourceFS = sourcePath.getFileSystem(configuration);
@@ -161,6 +160,7 @@ private long doCopy(CopyListingFileStatus source, Path target,
       if (useTempTarget) {
         LOG.info("Renaming temporary target file path {} to {}", targetPath,
             target);
+        target = new Path(target.toUri().getPath());
         promoteTmpToTarget(targetPath, target, targetFS);
       }
       LOG.info("Completed writing {} ({} bytes)", target, bytesRead);
@@ -257,17 +257,18 @@ private void promoteTmpToTarget(Path tmpTarget, Path target, FileSystem fs)
     }
   }
 
-  private Path getTempFile(Path target, Mapper.Context context) {
+  private Path getTempFile(Path target, Mapper.Context context, FileSystem fileSystem) {
     Path targetWorkPath = new Path(context.getConfiguration().
         get(DistCpConstants.CONF_LABEL_TARGET_WORK_PATH));
 
     Path root = target.equals(targetWorkPath) ? targetWorkPath.getParent()
         : targetWorkPath;
-    Path tempFile = new Path(root, ".distcp.tmp." +
+    String tempFilePrefix = DistCpUtils.getTargetTempFilePrefix(target);
+    Path tempFile = new Path(root, tempFilePrefix +
         context.getTaskAttemptID().toString() +
         "." + String.valueOf(System.currentTimeMillis()));
     LOG.info("Creating temp file: {}", tempFile);
-    return tempFile;
+    return new Path(tempFile.toUri().getPath());
   }
 
   @VisibleForTesting

hadoop-tools/hadoop-distcp/src/main/java/org/apache/hadoop/tools/util/DistCpUtils.java

@@ -662,4 +662,21 @@ public static Path getSplitChunkPath(Path targetFile,
         + ".____distcpSplit____" + srcFileStatus.getChunkOffset()
         + "." + srcFileStatus.getChunkLength());
   }
+
+  /**
+   * Return the target temp file prefix.
+   *
+   * The FTPFilesystem can't work well when the file name is starts with dot.
+   *
+   * @param targetPath target path
+   * @return temp file path prefix
+   */
+  public static String getTargetTempFilePrefix(Path targetPath) {
+    String schema = targetPath.toUri().getScheme();
+    if (StringUtils.equalsIgnoreCase("ftp", schema)) {
+      return DistCpConstants.TARGET_TEMP_FILE_PREFIX_FTP;
+    } else {
+      return DistCpConstants.TARGET_TEMP_FILE_PREFIX_DOT;
+    }
+  }
 }