HADOOP-3424. Values returned by getPartition should be checked to

make sure they are in the range 0 to #reduces - 1. Contributed by Chris 
Douglas. 


git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/trunk@658704 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -297,6 +297,10 @@ Trunk (unreleased changes)
     HADOOP-3375. Lease paths were sometimes not removed from 
     LeaseManager.sortedLeasesByPath. (Tsz Wo (Nicholas), SZE via dhruba)
 
+    HADOOP-3424. Values returned by getPartition should be checked to
+    make sure they are in the range 0 to #reduces - 1 (cdouglas via
+    omalley)
+
 Release 0.17.0 - 2008-05-18
 
   INCOMPATIBLE CHANGES

src/java/org/apache/hadoop/mapred/MapTask.java

@@ -475,6 +475,10 @@ class MapTask extends Task {
               : (bufvoid - keystart) + valend);
         }
         int partition = partitioner.getPartition(key, value, partitions);
+        if (partition < 0 || partition >= partitions) {
+          throw new IOException("Illegal partition for " + key + " (" +
+              partition + ")");
+        }
 
         mapOutputRecordCounter.increment(1);
 

src/test/org/apache/hadoop/mapred/TestMapRed.java

@@ -30,6 +30,7 @@ import junit.framework.TestCase;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
 import org.apache.hadoop.io.SequenceFile;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.WritableComparable;
@@ -296,6 +297,56 @@ public class TestMapRed extends TestCase {
     }
       
   }
+
+  private static class BadPartitioner
+      implements Partitioner<LongWritable,Text> {
+    boolean low;
+    public void configure(JobConf conf) {
+      low = conf.getBoolean("test.testmapred.badpartition", true);
+    }
+    public int getPartition(LongWritable k, Text v, int numPartitions) {
+      return low ? -1 : numPartitions;
+    }
+  }
+
+  public void testPartitioner() throws Exception {
+    JobConf conf = new JobConf(TestMapRed.class);
+    conf.setPartitionerClass(BadPartitioner.class);
+    FileSystem fs = FileSystem.getLocal(conf);
+    Path testdir = new Path(
+        System.getProperty("test.build.data","/tmp")).makeQualified(fs);
+    Path inFile = new Path(testdir, "blah/blah");
+    DataOutputStream f = fs.create(inFile);
+    f.writeBytes("blah blah blah\n");
+    f.close();
+    FileInputFormat.setInputPaths(conf, inFile);
+    FileOutputFormat.setOutputPath(conf, new Path(testdir, "out"));
+    conf.setMapperClass(IdentityMapper.class);
+    conf.setReducerClass(IdentityReducer.class);
+    conf.setOutputKeyClass(LongWritable.class);
+    conf.setOutputValueClass(Text.class);
+
+    // partition too low
+    conf.setBoolean("test.testmapred.badpartition", true);
+    boolean pass = true;
+    RunningJob rj = null;
+    try {
+      rj = JobClient.runJob(conf);
+    } catch (IOException e) {
+      pass = false;
+    }
+    assertFalse("should fail for partition < 0", pass);
+
+    // partition too high
+    conf.setBoolean("test.testmapred.badpartition", false);
+    pass = true;
+    try {
+      rj = JobClient.runJob(conf);
+    } catch (IOException e) {
+      pass = false;
+    }
+    assertFalse("should fail for partition >= numPartitions", pass);
+  }
     
   private void checkCompression(CompressionType mapCompression,
                                 CompressionType redCompression,