MAPREDUCE-3952. In MR2, when Total input paths to process == 1,CombinefileInputFormat.getSplits() returns 0 split. (Bhallamudi Venkata Siva Kamesh via tgraves)

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

hadoop-mapreduce-project/CHANGES.txt

@@ -27,6 +27,10 @@ Release 0.23.7 - UNRELEASED
     MAPREDUCE-4458. Warn if java.library.path is used for AM or Task
     (Robert Parker via jeagles)
 
+    MAPREDUCE-3952. In MR2, when Total input paths to process == 1,
+    CombinefileInputFormat.getSplits() returns 0 split. (Bhallamudi 
+    Venkata Siva Kamesh via tgraves)
+
 Release 0.23.6 - UNRELEASED
 
   INCOMPATIBLE CHANGES

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/CombineFileInputFormat.java

@@ -498,6 +498,7 @@ public abstract class CombineFileInputFormat<K, V>
       if (locations == null) {
         blocks = new OneBlockInfo[0];
       } else {
+
         if (!isSplitable) {
           // if the file is not splitable, just create the one block with
           // full file length
@@ -515,7 +516,7 @@ public abstract class CombineFileInputFormat<K, V>
             long left = locations[i].getLength();
             long myOffset = locations[i].getOffset();
             long myLength = 0;
-            while (left > 0) {
+            do {
               if (maxSize == 0) {
                 myLength = left;
               } else {
@@ -537,7 +538,7 @@ public abstract class CombineFileInputFormat<K, V>
               myOffset += myLength;
 
               blocksList.add(oneblock);
-            }
+            } while (left > 0);
           }
           blocks = blocksList.toArray(new OneBlockInfo[blocksList.size()]);
         }

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/lib/input/TestCombineFileInputFormat.java

@@ -1147,6 +1147,35 @@ public class TestCombineFileInputFormat extends TestCase {
     }
   }
 
+  
+  /**
+   * Test when the input file's length is 0.
+   */
+  @Test
+  public void testForEmptyFile() throws Exception {
+    Configuration conf = new Configuration();
+    FileSystem fileSys = FileSystem.get(conf);
+    Path file = new Path("test" + "/file");
+    FSDataOutputStream out = fileSys.create(file, true,
+        conf.getInt("io.file.buffer.size", 4096), (short) 1, (long) BLOCKSIZE);
+    out.write(new byte[0]);
+    out.close();
+
+    // split it using a CombinedFile input format
+    DummyInputFormat inFormat = new DummyInputFormat();
+    Job job = Job.getInstance(conf);
+    FileInputFormat.setInputPaths(job, "test");
+    List<InputSplit> splits = inFormat.getSplits(job);
+    assertEquals(1, splits.size());
+    CombineFileSplit fileSplit = (CombineFileSplit) splits.get(0);
+    assertEquals(1, fileSplit.getNumPaths());
+    assertEquals(file.getName(), fileSplit.getPath(0).getName());
+    assertEquals(0, fileSplit.getOffset(0));
+    assertEquals(0, fileSplit.getLength(0));
+
+    fileSys.delete(file.getParent(), true);
+  }
+
   static class TestFilter implements PathFilter {
     private Path p;