MAPREDUCE-5698. Backport MAPREDUCE-1285 to branch-1 (Yongjun Zhang via Sandy Ryza)

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

CHANGES.txt

@@ -175,6 +175,9 @@ Release 1.3.0 - unreleased
     MAPREDUCE-5702. TaskLogServlet.printTaskLog() has spurious HTML closing 
     tags (rkanter via kasha)
 
+    MAPREDUCE-5698. Backport MAPREDUCE-1285 to branch-1 (Yongjun Zhang via
+    Sandy Ryza)
+
 Release 1.2.2 - unreleased
 
   INCOMPATIBLE CHANGES

src/test/org/apache/hadoop/fs/TestCopyFiles.java

@@ -909,6 +909,40 @@ public class TestCopyFiles extends TestCase {
     }
   }
 
+  /**
+   * verify that -delete option works for other {@link FileSystem}
+   * implementations. See MAPREDUCE-1285
+   */
+  public void testDeleteLocal() throws Exception {
+    MiniDFSCluster cluster = null;
+    try {
+      Configuration conf = new Configuration();
+      final FileSystem localfs = FileSystem.get(LOCAL_FS, conf);
+      cluster = new MiniDFSCluster(conf, 1, true, null);
+      final FileSystem hdfs = cluster.getFileSystem();
+      final String namenode = FileSystem.getDefaultUri(conf).toString();
+      if (namenode.startsWith("hdfs://")) {
+        MyFile[] files = createFiles(URI.create(namenode), "/srcdat");
+        String destdir = TEST_ROOT_DIR + "/destdat";
+        MyFile[] localFiles = createFiles(localfs, destdir);
+        ToolRunner.run(new DistCp(conf), new String[] { "-delete", "-update",
+            "-log", "/logs", namenode + "/srcdat",
+            "file:///" + TEST_ROOT_DIR + "/destdat" });
+        assertTrue("Source and destination directories do not match.",
+            checkFiles(localfs, destdir, files));
+        assertTrue("Log directory does not exist.",
+            hdfs.exists(new Path("/logs")));
+        deldir(localfs, destdir);
+        deldir(hdfs, "/logs");
+        deldir(hdfs, "/srcdat");
+      }
+    } finally {
+      if (cluster != null) {
+        cluster.shutdown();
+      }
+    }
+  }
+
   static void create(FileSystem fs, Path f) throws IOException {
     FSDataOutputStream out = fs.create(f);
     try {

src/tools/org/apache/hadoop/tools/DistCp.java

@@ -1275,7 +1275,7 @@ public class DistCp implements Tool {
     //write dst lsr results
     final Path dstlsr = new Path(jobdir, "_distcp_dst_lsr");
     final SequenceFile.Writer writer = SequenceFile.createWriter(jobfs, jobconf,
-        dstlsr, Text.class, FileStatus.class,
+        dstlsr, Text.class, dstroot.getClass(),
         SequenceFile.CompressionType.NONE);
     try {
       //do lsr to get all file statuses in dstroot