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

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

@@ -191,4 +191,6 @@ private DistCpConstants() {
 
   public static final String CLASS_INSTANTIATION_ERROR_MSG =
           "Unable to instantiate ";
+
+  public static final String TARGET_TMP_FILE_PREFIX = "distcp.tmp.";
 }

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

@@ -45,10 +45,8 @@
 
 import java.io.FileNotFoundException;
 import java.io.IOException;
-import java.util.ArrayList;
-import java.util.EnumSet;
-import java.util.LinkedList;
-import java.util.List;
+import java.util.*;
+import java.util.stream.Collectors;
 
 import static org.apache.hadoop.tools.DistCpConstants.*;
 
@@ -170,17 +168,38 @@ private void deleteAttemptTempFiles(Path targetWorkPath,
       return;
     }
 
-    FileStatus[] tempFiles = targetFS.globStatus(
-        new Path(targetWorkPath, ".distcp.tmp." + jobId.replaceAll("job","attempt") + "*"));
+    String tmpFilePattern = TARGET_TMP_FILE_PREFIX + jobId.replaceAll("job","attempt") + "*";
+    List<FileStatus> tempFiles = listTmpFilePaths(targetFS, targetWorkPath, tmpFilePattern);
 
-    if (tempFiles != null && tempFiles.length > 0) {
+    if (tempFiles != null && tempFiles.size() > 0) {
       for (FileStatus file : tempFiles) {
         LOG.info("Cleaning up " + file.getPath());
         targetFS.delete(file.getPath(), false);
       }
     }
   }
 
+  private List<FileStatus> listTmpFilePaths(FileSystem fileSystem, Path targetWorkPath, String pattern) throws IOException {
+    List<Path> tmpPaths = new ArrayList<>();
+    tmpPaths.add(targetWorkPath);
+    listChildrenPaths(fileSystem, targetWorkPath, tmpPaths);
+    List<FileStatus> allTmpFiles = new ArrayList<>(tmpPaths.size());
+    for (Path path : tmpPaths) {
+      FileStatus[] tmpFiles = fileSystem.globStatus(new Path(path, pattern));
+      allTmpFiles.addAll(Arrays.asList(tmpFiles));
+    }
+    return allTmpFiles;
+  }
+
+  private void listChildrenPaths(FileSystem fileSystem, Path targetWorkPath, List<Path> paths) throws IOException {
+    List<Path> directoryPaths = Arrays.stream(fileSystem.listStatus(targetWorkPath))
+            .filter(status -> status.isDirectory()).map(status -> status.getPath()).collect(Collectors.toList());
+    paths.addAll(directoryPaths);
+    for (Path path : directoryPaths) {
+      listChildrenPaths(fileSystem, path, paths);
+    }
+  }
+
   /**
    * Cleanup meta folder and other temporary files
    *

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

@@ -263,7 +263,7 @@ private Path getTempFile(Path target, Mapper.Context context) {
 
     Path root = target.equals(targetWorkPath) ? targetWorkPath.getParent()
         : targetWorkPath;
-    Path tempFile = new Path(root, ".distcp.tmp." +
+    Path tempFile = new Path(root, DistCpConstants.TARGET_TMP_FILE_PREFIX +
         context.getTaskAttemptID().toString() +
         "." + String.valueOf(System.currentTimeMillis()));
     LOG.info("Creating temp file: {}", tempFile);