svn commit: r658704 - in /hadoop/core/trunk: CHANGES.txt src/java/org/apache/hadoop/mapred/MapTask.java src/test/org/apache/hadoop/mapred/TestMapRed.java

[om...@apache.org]

Author: omalley
Date: Wed May 21 07:31:35 2008
New Revision: 658704

URL: http://svn.apache.org/viewvc?rev=658704&view=rev
Log:
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. 

Modified:
    hadoop/core/trunk/CHANGES.txt
    hadoop/core/trunk/src/java/org/apache/hadoop/mapred/MapTask.java
    hadoop/core/trunk/src/test/org/apache/hadoop/mapred/TestMapRed.java

Modified: hadoop/core/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/CHANGES.txt?rev=658704&r1=658703&r2=658704&view=diff
==============================================================================
--- hadoop/core/trunk/CHANGES.txt (original)
+++ hadoop/core/trunk/CHANGES.txt Wed May 21 07:31:35 2008
@@ -297,6 +297,10 @@
     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

Modified: hadoop/core/trunk/src/java/org/apache/hadoop/mapred/MapTask.java
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/src/java/org/apache/hadoop/mapred/MapTask.java?rev=658704&r1=658703&r2=658704&view=diff
==============================================================================
--- hadoop/core/trunk/src/java/org/apache/hadoop/mapred/MapTask.java (original)
+++ hadoop/core/trunk/src/java/org/apache/hadoop/mapred/MapTask.java Wed May 21 07:31:35 2008
@@ -475,6 +475,10 @@
               : (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);
 

Modified: hadoop/core/trunk/src/test/org/apache/hadoop/mapred/TestMapRed.java
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/src/test/org/apache/hadoop/mapred/TestMapRed.java?rev=658704&r1=658703&r2=658704&view=diff
==============================================================================
--- hadoop/core/trunk/src/test/org/apache/hadoop/mapred/TestMapRed.java (original)
+++ hadoop/core/trunk/src/test/org/apache/hadoop/mapred/TestMapRed.java Wed May 21 07:31:35 2008
@@ -30,6 +30,7 @@
 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 @@
     }
       
   }
+
+  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,