Author: szetszwo
Date: Wed Jun  3 22:08:31 2009
New Revision: 781602

URL: http://svn.apache.org/viewvc?rev=781602&view=rev
Log:
HADOOP-5762. Fix a problem that DistCp does not copy empty directory.  Contributed by Rodrigo Schmidt 

Modified:
    hadoop/core/trunk/CHANGES.txt
    hadoop/core/trunk/src/test/hdfs-with-mr/org/apache/hadoop/fs/TestCopyFiles.java
    hadoop/core/trunk/src/tools/org/apache/hadoop/tools/DistCp.java

Modified: hadoop/core/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/CHANGES.txt?rev=781602&r1=781601&r2=781602&view=diff
==============================================================================
--- hadoop/core/trunk/CHANGES.txt (original)
+++ hadoop/core/trunk/CHANGES.txt Wed Jun  3 22:08:31 2009
@@ -774,6 +774,9 @@
 
     HADOOP-5861. s3n files are not getting split by default. (tomwhite)
 
+    HADOOP-5762. Fix a problem that DistCp does not copy empty directory.
+    (Rodrigo Schmidt via szetszwo)
+
 Release 0.20.1 - Unreleased
 
   INCOMPATIBLE CHANGES

Modified: hadoop/core/trunk/src/test/hdfs-with-mr/org/apache/hadoop/fs/TestCopyFiles.java
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/src/test/hdfs-with-mr/org/apache/hadoop/fs/TestCopyFiles.java?rev=781602&r1=781601&r2=781602&view=diff
==============================================================================
--- hadoop/core/trunk/src/test/hdfs-with-mr/org/apache/hadoop/fs/TestCopyFiles.java (original)
+++ hadoop/core/trunk/src/test/hdfs-with-mr/org/apache/hadoop/fs/TestCopyFiles.java Wed Jun  3 22:08:31 2009
@@ -295,6 +295,37 @@
       if (cluster != null) { cluster.shutdown(); }
     }
   }
+
+  /** copy empty directory on dfs file system */
+  public void testEmptyDir() throws Exception {
+    String namenode = null;
+    MiniDFSCluster cluster = null;
+    try {
+      Configuration conf = new Configuration();
+      cluster = new MiniDFSCluster(conf, 2, true, null);
+      final FileSystem hdfs = cluster.getFileSystem();
+      namenode = FileSystem.getDefaultUri(conf).toString();
+      if (namenode.startsWith("hdfs://")) {
+        
+        FileSystem fs = FileSystem.get(URI.create(namenode), new Configuration());
+        fs.mkdirs(new Path("/empty"));
+
+        ToolRunner.run(new DistCp(conf), new String[] {
+                                         "-log",
+                                         namenode+"/logs",
+                                         namenode+"/empty",
+                                         namenode+"/dest"});
+        fs = FileSystem.get(URI.create(namenode+"/destdat"), conf);
+        assertTrue("Destination directory does not exist.",
+                   fs.exists(new Path(namenode+"/dest")));
+        deldir(hdfs, "/dest");
+        deldir(hdfs, "/empty");
+        deldir(hdfs, "/logs");
+      }
+    } finally {
+      if (cluster != null) { cluster.shutdown(); }
+    }
+  }
   
   /** copy files from local file system to dfs file system */
   public void testCopyFromLocalToDfs() throws Exception {

Modified: hadoop/core/trunk/src/tools/org/apache/hadoop/tools/DistCp.java
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/src/tools/org/apache/hadoop/tools/DistCp.java?rev=781602&r1=781601&r2=781602&view=diff
==============================================================================
--- hadoop/core/trunk/src/tools/org/apache/hadoop/tools/DistCp.java (original)
+++ hadoop/core/trunk/src/tools/org/apache/hadoop/tools/DistCp.java Wed Jun  3 22:08:31 2009
@@ -1054,7 +1054,7 @@
     final boolean special =
       (args.srcs.size() == 1 && !dstExists) || update || overwrite;
     int srcCount = 0, cnsyncf = 0, dirsyn = 0;
-    long fileCount = 0L, byteCount = 0L, cbsyncs = 0L;
+    long fileCount = 0L, dirCount = 0L, byteCount = 0L, cbsyncs = 0L;
     try {
       for(Iterator<Path> srcItr = args.srcs.iterator(); srcItr.hasNext(); ) {
         final Path src = srcItr.next();
@@ -1063,6 +1063,10 @@
         Path root = special && srcfilestat.isDir()? src: src.getParent();
         if (srcfilestat.isDir()) {
           ++srcCount;
+          ++dirCount;
+          final String dst = makeRelative(root,src);
+          src_writer.append(new LongWritable(0), new FilePair(srcfilestat, dst));
+          dst_writer.append(new Text(dst), new Text(src.toString()));
         }
 
         Stack<FileStatus> pathstack = new Stack<FileStatus>();
@@ -1077,6 +1081,7 @@
 
             if (child.isDir()) {
               pathstack.push(child);
+              ++dirCount;
             }
             else {
               //skip file if the src and the dst files are the same.
@@ -1161,7 +1166,7 @@
     jobConf.setInt(SRC_COUNT_LABEL, srcCount);
     jobConf.setLong(TOTAL_SIZE_LABEL, byteCount);
     setMapCount(byteCount, jobConf);
-    return fileCount > 0;
+    return (fileCount + dirCount) > 0;
   }
 
   /**