From 6e831b0f5ebaaea0609a8d1195cfccd2aa55a6da Mon Sep 17 00:00:00 2001
From: Alejandro Abdelnur <tucu@apache.org>
Date: Thu, 23 Aug 2012 00:24:36 +0000
Subject: [PATCH] MAPREDUCE-4470. Fix
 TestCombineFileInputFormat.testForEmptyFile (ikatsov via tucu)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1376326 13f79535-47bb-0310-9956-ffa450edef68
---
 hadoop-mapreduce-project/CHANGES.txt          |  2 +
 .../lib/input/CombineFileInputFormat.java     |  5 +++
 .../lib/input/TestFileInputFormat.java        | 39 +++++++++++++++++++
 3 files changed, 46 insertions(+)

diff --git a/hadoop-mapreduce-project/CHANGES.txt b/hadoop-mapreduce-project/CHANGES.txt
index 2cd90288df5..150bb217024 100644
--- a/hadoop-mapreduce-project/CHANGES.txt
+++ b/hadoop-mapreduce-project/CHANGES.txt
@@ -53,6 +53,8 @@ Branch-2 ( Unreleased changes )
     MAPREDUCE-4577. HDFS-3672 broke
     TestCombineFileInputFormat.testMissingBlocks() test (atm)
 
+    MAPREDUCE-4470. Fix TestCombineFileInputFormat.testForEmptyFile (ikatsov via tucu)
+
 Release 2.1.0-alpha - Unreleased 
 
   INCOMPATIBLE CHANGES
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/CombineFileInputFormat.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/CombineFileInputFormat.java
index f991f2a6501..b62c2fb0aa6 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/CombineFileInputFormat.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/lib/input/CombineFileInputFormat.java
@@ -497,6 +497,11 @@ public abstract class CombineFileInputFormat<K, V>
       if (locations == null) {
         blocks = new OneBlockInfo[0];
       } else {
+
+        if(locations.length == 0) {
+          locations = new BlockLocation[] { new BlockLocation() };
+        }
+
         if (!isSplitable) {
           // if the file is not splitable, just create the one block with
           // full file length
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/lib/input/TestFileInputFormat.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/lib/input/TestFileInputFormat.java
index 28359585a2d..22bc960ae68 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/lib/input/TestFileInputFormat.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-jobclient/src/test/java/org/apache/hadoop/mapreduce/lib/input/TestFileInputFormat.java
@@ -33,7 +33,9 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.BlockLocation;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mapreduce.InputSplit;
 import org.apache.hadoop.mapreduce.Job;
 import org.apache.hadoop.mapreduce.JobContext;
@@ -136,6 +138,43 @@ public class TestFileInputFormat {
     }
   }
 
+  /**
+   * 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) 1024);
+      out.write(new byte[0]);
+      out.close();
+
+      // split it using a File 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());
+      FileSplit fileSplit = (FileSplit) splits.get(0);
+      assertEquals(0, fileSplit.getLocations().length);
+      assertEquals(file.getName(), fileSplit.getPath().getName());
+      assertEquals(0, fileSplit.getStart());
+      assertEquals(0, fileSplit.getLength());
+
+      fileSys.delete(file.getParent(), true);
+  }
+
+  /** Dummy class to extend FileInputFormat*/
+  private class DummyInputFormat extends FileInputFormat<Text, Text> {
+    @Override
+    public RecordReader<Text,Text> createRecordReader(InputSplit split,
+        TaskAttemptContext context) throws IOException {
+      return null;
+    }
+  }
+
   private class FileInputFormatForTest<K, V> extends FileInputFormat<K, V> {
 
     long splitSize;