HADOOP-3828. Provides a way to write skipped records to DFS. Contributed by Sharad Agarwal.

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

CHANGES.txt

@@ -108,6 +108,9 @@ Trunk (unreleased changes)
 
     HADOOP-3754. Add a thrift interface to access HDFS. (dhruba via omalley)
 
+    HADOOP-3828. Provides a way to write skipped records to DFS.
+    (Sharad Agarwal via ddas)
+
   IMPROVEMENTS
 
     HADOOP-3908. Fuse-dfs: better error message if llibhdfs.so doesn't exist.

src/contrib/streaming/src/test/org/apache/hadoop/streaming/TestStreamingBadRecords.java

@@ -121,13 +121,12 @@ public class TestStreamingBadRecords extends ClusterMapReduceTestCase
   public void testDisableSkip() throws Exception {
     JobConf clusterConf = createJobConf();
     createInput();
-    
+    int attSkip =0;
+    SkipBadRecords.setAttemptsToStartSkipping(clusterConf,attSkip);
     //the no of attempts to successfully complete the task depends 
     //on the no of bad records.
-    int mapperAttempts = SkipBadRecords.getAttemptsToStartSkipping(clusterConf)
-                          +1+MAPPER_BAD_RECORDS.size();
-    int reducerAttempts = SkipBadRecords.getAttemptsToStartSkipping(clusterConf)
-                           +1+REDUCER_BAD_RECORDS.size();
+    int mapperAttempts = attSkip+1+MAPPER_BAD_RECORDS.size();
+    int reducerAttempts = attSkip+1+REDUCER_BAD_RECORDS.size();
     String[] args =  new String[] {
       "-input", (new Path(getInputDir(), "text.txt")).toString(),
       "-output", getOutputDir().toString(),
@@ -135,6 +134,7 @@ public class TestStreamingBadRecords extends ClusterMapReduceTestCase
       "-reducer", badReducer,
       "-verbose",
       "-inputformat", "org.apache.hadoop.mapred.KeyValueTextInputFormat",
+      "-jobconf", "mapred.skip.attempts.to.start.skipping="+attSkip,
       "-jobconf", "mapred.map.max.attempts="+mapperAttempts,
       "-jobconf", "mapred.reduce.max.attempts="+reducerAttempts,
       "-jobconf", "mapred.skip.mode.enabled=false",
@@ -157,13 +157,12 @@ public class TestStreamingBadRecords extends ClusterMapReduceTestCase
   public void testSkip() throws Exception {
     JobConf clusterConf = createJobConf();
     createInput();
-    
+    int attSkip =0;
+    SkipBadRecords.setAttemptsToStartSkipping(clusterConf,attSkip);
     //the no of attempts to successfully complete the task depends 
     //on the no of bad records.
-    int mapperAttempts = SkipBadRecords.getAttemptsToStartSkipping(clusterConf)
-                          +1+MAPPER_BAD_RECORDS.size();
-    int reducerAttempts = SkipBadRecords.getAttemptsToStartSkipping(clusterConf)
-                           +1+REDUCER_BAD_RECORDS.size();
+    int mapperAttempts = attSkip+1+MAPPER_BAD_RECORDS.size();
+    int reducerAttempts = attSkip+1+REDUCER_BAD_RECORDS.size();
     
     String[] args =  new String[] {
       "-input", (new Path(getInputDir(), "text.txt")).toString(),
@@ -172,6 +171,7 @@ public class TestStreamingBadRecords extends ClusterMapReduceTestCase
       "-reducer", badReducer,
       "-verbose",
       "-inputformat", "org.apache.hadoop.mapred.KeyValueTextInputFormat",
+      "-jobconf", "mapred.skip.attempts.to.start.skipping="+attSkip,
       "-jobconf", "mapred.map.max.attempts="+mapperAttempts,
       "-jobconf", "mapred.reduce.max.attempts="+reducerAttempts,
       "-jobconf", "mapred.map.tasks=1",

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

@@ -45,7 +45,9 @@ import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.BytesWritable;
 import org.apache.hadoop.io.DataInputBuffer;
 import org.apache.hadoop.io.RawComparator;
+import org.apache.hadoop.io.SequenceFile;
 import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.SequenceFile.CompressionType;
 import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.hadoop.io.compress.DefaultCodec;
 import org.apache.hadoop.io.serializer.SerializationFactory;
@@ -53,6 +55,7 @@ import org.apache.hadoop.io.serializer.Serializer;
 import org.apache.hadoop.mapred.IFile.Writer;
 import org.apache.hadoop.mapred.IFile.Reader;
 import org.apache.hadoop.mapred.Merger.Segment;
+import org.apache.hadoop.mapred.SortedRanges.SkipRangeIterator;
 import org.apache.hadoop.util.IndexedSortable;
 import org.apache.hadoop.util.IndexedSorter;
 import org.apache.hadoop.util.Progress;
@@ -143,19 +146,14 @@ class MapTask extends Task {
     private RecordReader<K,V> rawIn;
     private Counters.Counter inputByteCounter;
     private Counters.Counter inputRecordCounter;
-    private Iterator<Long> skipFailedRecIndexIterator;
-    private TaskUmbilicalProtocol umbilical;
-    private long recIndex = -1;
     private long beforePos = -1;
     private long afterPos = -1;
     
-    TrackedRecordReader(RecordReader<K,V> raw, Counters counters, 
-        TaskUmbilicalProtocol umbilical) {
+    TrackedRecordReader(RecordReader<K,V> raw, Counters counters) 
+      throws IOException{
       rawIn = raw;
-      this.umbilical = umbilical;
       inputRecordCounter = counters.findCounter(MAP_INPUT_RECORDS);
       inputByteCounter = counters.findCounter(MAP_INPUT_BYTES);
-      skipFailedRecIndexIterator = getFailedRanges().skipRangeIterator();
     }
 
     public K createKey() {
@@ -169,28 +167,22 @@ class MapTask extends Task {
     public synchronized boolean next(K key, V value)
     throws IOException {
       boolean ret = moveToNext(key, value);
-      if(isSkipping() && ret) {
-        long nextRecIndex = skipFailedRecIndexIterator.next();
-        long skip = nextRecIndex - recIndex;
-        for(int i=0;i<skip && ret;i++) {
-          ret = moveToNext(key, value);
-        }
-        getCounters().incrCounter(Counter.MAP_SKIPPED_RECORDS, skip);
-        reportNextRecordRange(umbilical, nextRecIndex);
-      }
       if (ret) {
-        inputRecordCounter.increment(1);
-        inputByteCounter.increment(afterPos - beforePos);
+        incrCounters();
       }
       return ret;
     }
+    
+    protected void incrCounters() {
+      inputRecordCounter.increment(1);
+      inputByteCounter.increment(afterPos - beforePos);
+    }
      
-    private synchronized boolean moveToNext(K key, V value)
+    protected synchronized boolean moveToNext(K key, V value)
       throws IOException {
       setProgress(getProgress());
       beforePos = getPos();
       boolean ret = rawIn.next(key, value);
-      recIndex++;
       afterPos = getPos();
       return ret;
     }
@@ -202,6 +194,68 @@ class MapTask extends Task {
     }
   }
 
+  /**
+   * This class skips the records based on the failed ranges from previous 
+   * attempts.
+   */
+  class SkippingRecordReader<K, V> extends TrackedRecordReader<K,V> {
+    private SkipRangeIterator skipIt;
+    private SequenceFile.Writer skipWriter;
+    private TaskUmbilicalProtocol umbilical;
+    private Counters.Counter skipRecCounter;
+    private long recIndex = -1;
+    
+    SkippingRecordReader(RecordReader<K,V> raw, Counters counters, 
+        TaskUmbilicalProtocol umbilical) throws IOException{
+      super(raw,counters);
+      this.umbilical = umbilical;
+      this.skipRecCounter = counters.findCounter(Counter.MAP_SKIPPED_RECORDS);
+      skipIt = getFailedRanges().skipRangeIterator();
+    }
+    
+    public synchronized boolean next(K key, V value)
+    throws IOException {
+      boolean ret = moveToNext(key, value);
+      long nextRecIndex = skipIt.next();
+      long skip = nextRecIndex - recIndex;
+      for(int i=0;i<skip && ret;i++) {
+      	writeSkippedRec(key, value);
+        ret = moveToNext(key, value);
+      }
+      //close the skip writer once all the ranges are skipped
+      if(skip>0 && skipIt.skippedAllRanges() && skipWriter!=null) {
+        skipWriter.close();
+      }
+      skipRecCounter.increment(skip);
+      reportNextRecordRange(umbilical, nextRecIndex);
+      if (ret) {
+        incrCounters();
+      }
+      return ret;
+    }
+    
+    protected synchronized boolean moveToNext(K key, V value)
+    throws IOException {
+	    recIndex++;
+      return super.moveToNext(key, value);
+    }
+    
+    @SuppressWarnings("unchecked")
+    private void writeSkippedRec(K key, V value) throws IOException{
+      if(skipWriter==null) {
+        Path skipDir = SkipBadRecords.getSkipOutputPath(conf);
+        Path skipFile = new Path(skipDir, getTaskID().toString());
+        skipWriter = 
+          SequenceFile.createWriter(
+              skipFile.getFileSystem(conf), conf, skipFile,
+              (Class<K>) createKey().getClass(),
+              (Class<V>) createValue().getClass(), 
+              CompressionType.BLOCK, getReporter(umbilical));
+      }
+      skipWriter.append(key, value);
+    }
+  }
+
   @Override
   @SuppressWarnings("unchecked")
   public void run(final JobConf job, final TaskUmbilicalProtocol umbilical)
@@ -244,7 +298,9 @@ class MapTask extends Task {
       
     RecordReader rawIn =                  // open input
       job.getInputFormat().getRecordReader(instantiatedSplit, job, reporter);
-    RecordReader in = new TrackedRecordReader(rawIn, getCounters(), umbilical);
+    RecordReader in = isSkipping() ? 
+        new SkippingRecordReader(rawIn, getCounters(), umbilical) :
+        new TrackedRecordReader(rawIn, getCounters());
     job.setBoolean("mapred.skip.on", isSkipping());
 
     MapRunnable runner =

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

@@ -56,16 +56,20 @@ import org.apache.hadoop.io.DataInputBuffer;
 import org.apache.hadoop.io.IOUtils;
 import org.apache.hadoop.io.IntWritable;
 import org.apache.hadoop.io.RawComparator;
+import org.apache.hadoop.io.SequenceFile;
 import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.io.WritableFactories;
 import org.apache.hadoop.io.WritableFactory;
 import org.apache.hadoop.io.WritableUtils;
+import org.apache.hadoop.io.SequenceFile.CompressionType;
 import org.apache.hadoop.io.compress.CodecPool;
 import org.apache.hadoop.io.compress.CompressionCodec;
 import org.apache.hadoop.io.compress.Decompressor;
 import org.apache.hadoop.io.compress.DefaultCodec;
 import org.apache.hadoop.mapred.IFile.*;
 import org.apache.hadoop.mapred.Merger.Segment;
+import org.apache.hadoop.mapred.SortedRanges.SkipRangeIterator;
+import org.apache.hadoop.mapred.Task.Counter;
 import org.apache.hadoop.metrics.MetricsContext;
 import org.apache.hadoop.metrics.MetricsRecord;
 import org.apache.hadoop.metrics.MetricsUtil;
@@ -232,6 +236,10 @@ class ReduceTask extends Task {
     @Override
     public VALUE next() {
       reduceInputValueCounter.increment(1);
+      return moveToNext();
+    }
+    
+    protected VALUE moveToNext() {
       return super.next();
     }
     
@@ -243,9 +251,14 @@ class ReduceTask extends Task {
 
   private class SkippingReduceValuesIterator<KEY,VALUE> 
      extends ReduceValuesIterator<KEY,VALUE> {
-     private Iterator<Long> skipFailedRecIndexIterator;
+     private SkipRangeIterator skipIt;
      private TaskUmbilicalProtocol umbilical;
+     private Counters.Counter skipGroupCounter;
+     private Counters.Counter skipRecCounter;
      private long recIndex = -1;
+     private Class<KEY> keyClass;
+     private Class<VALUE> valClass;
+     private SequenceFile.Writer skipWriter;
      
      public SkippingReduceValuesIterator(RawKeyValueIterator in,
          RawComparator<KEY> comparator, Class<KEY> keyClass,
@@ -253,7 +266,13 @@ class ReduceTask extends Task {
          TaskUmbilicalProtocol umbilical) throws IOException {
        super(in, comparator, keyClass, valClass, conf, reporter);
        this.umbilical = umbilical;
-       skipFailedRecIndexIterator = getFailedRanges().skipRangeIterator();
+       this.skipGroupCounter = 
+         getCounters().findCounter(Counter.REDUCE_SKIPPED_GROUPS);
+       this.skipRecCounter = 
+         getCounters().findCounter(Counter.REDUCE_SKIPPED_RECORDS);
+       this.keyClass = keyClass;
+       this.valClass = valClass;
+       skipIt = getFailedRanges().skipRangeIterator();
        mayBeSkip();
      }
      
@@ -264,15 +283,38 @@ class ReduceTask extends Task {
      
      private void mayBeSkip() throws IOException {
        recIndex++;
-       long nextRecIndex = skipFailedRecIndexIterator.next();
+       long nextRecIndex = skipIt.next();
        long skip = nextRecIndex - recIndex;
+       long skipRec = 0;
        for(int i=0;i<skip && super.more();i++) {
+         while (hasNext()) {
+           writeSkippedRec(getKey(), moveToNext());
+           skipRec++;
+         }
          super.nextKey();
          recIndex++;
        }
-       getCounters().incrCounter(Counter.REDUCE_SKIPPED_RECORDS, skip);
+       //close the skip writer once all the ranges are skipped
+       if(skip>0 && skipIt.skippedAllRanges() && skipWriter!=null) {
+         skipWriter.close();
+       }
+       skipGroupCounter.increment(skip);
+       skipRecCounter.increment(skipRec);
        reportNextRecordRange(umbilical, nextRecIndex);
      }
+     
+     @SuppressWarnings("unchecked")
+     private void writeSkippedRec(KEY key, VALUE value) throws IOException{
+       if(skipWriter==null) {
+         Path skipDir = SkipBadRecords.getSkipOutputPath(conf);
+         Path skipFile = new Path(skipDir, getTaskID().toString());
+         skipWriter = SequenceFile.createWriter(
+               skipFile.getFileSystem(conf), conf, skipFile,
+               keyClass, valClass, 
+               CompressionType.BLOCK, getReporter(umbilical));
+       }
+       skipWriter.append(key, value);
+     }
   }
 
   @Override

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

@@ -19,6 +19,7 @@
 package org.apache.hadoop.mapred;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
 
 /**
  * Utility class for skip bad records functionality. It contains various 
@@ -33,6 +34,7 @@ public class SkipBadRecords {
     "mapred.skip.map.auto.incr.proc.count";
   private static final String AUTO_INCR_REDUCE_PROC_COUNT = 
     "mapred.skip.reduce.auto.incr.proc.count";
+  private static final String OUT_PATH = "mapred.skip.out.dir";
   
   /**
    * Is skipping of bad records enabled. If it is enabled 
@@ -158,4 +160,31 @@ public class SkipBadRecords {
     conf.setBoolean(AUTO_INCR_REDUCE_PROC_COUNT, autoIncr);
   }
   
+  /**
+   * Get the directory to which skipped records are written. By default it is 
+   * the sub directory of the output _logs directory.
+   * @param conf the configuration.
+   * @return path skip output directory. Null is returned if this is not set 
+   * and output directory is also not set.
+   */
+  public static Path getSkipOutputPath(Configuration conf) {
+    String name =  conf.get(OUT_PATH);
+    if(name!=null) {
+      return new Path(name);
+    }
+    Path outPath = FileOutputFormat.getOutputPath(new JobConf(conf));
+    return outPath==null ? null : new Path(outPath, 
+        "_logs"+Path.SEPARATOR+"skip");
+  }
+  
+  /**
+   * Set the directory to which skipped records are written. By default it is 
+   * the sub directory of the output _logs directory.
+   * @param conf the configuration.
+   * @param path skip output directory path
+   */
+  public static void setSkipOutputPath(JobConf conf, Path path) {
+    conf.set(OUT_PATH, path.toString());
+  }
+  
 }

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

@@ -35,28 +35,28 @@ import org.apache.hadoop.io.Writable;
  * Provides the SkipRangeIterator, which skips the Ranges 
  * stored in this object.
  */
-public class SortedRanges implements Writable{
+class SortedRanges implements Writable{
   
   private static final Log LOG = 
     LogFactory.getLog(SortedRanges.class);
   
-  private SortedSet<Range> ranges = new TreeSet<Range>();
-  private int indicesCount;
+  private TreeSet<Range> ranges = new TreeSet<Range>();
+  private long indicesCount;
   
   /**
    * Get Iterator which skips the stored ranges.
    * The Iterator.next() call return the index starting from 0.
-   * @return Iterator<Long>
+   * @return SkipRangeIterator
    */
-  public Iterator<Long> skipRangeIterator(){
-    return new SkipRangeIterator();
+  synchronized SkipRangeIterator skipRangeIterator(){
+    return new SkipRangeIterator(ranges.iterator());
   }
   
   /**
    * Get the no of indices stored in the ranges.
    * @return indices count
    */
-  public synchronized int getIndicesCount() {
+  synchronized long getIndicesCount() {
     return indicesCount;
   }
   
@@ -69,7 +69,7 @@ public class SortedRanges implements Writable{
    * If the range is of 0 length, doesn't do anything.
    * @param range Range to be added.
    */
-  public synchronized void add(Range range){
+  synchronized void add(Range range){
     if(range.isEmpty()) {
       return;
     }
@@ -123,7 +123,7 @@ public class SortedRanges implements Writable{
    * If range is of 0 length, doesn't do anything.
    * @param range Range to be removed.
    */
-  public synchronized void remove(Range range) {
+  synchronized void remove(Range range) {
     if(range.isEmpty()) {
       return;
     }
@@ -177,7 +177,8 @@ public class SortedRanges implements Writable{
     }
   }
   
-  public void readFields(DataInput in) throws IOException {
+  public synchronized void readFields(DataInput in) throws IOException {
+    indicesCount = in.readLong();
     ranges = new TreeSet<Range>();
     int size = in.readInt();
     for(int i=0;i<size;i++) {
@@ -187,7 +188,8 @@ public class SortedRanges implements Writable{
     }
   }
 
-  public void write(DataOutput out) throws IOException {
+  public synchronized void write(DataOutput out) throws IOException {
+    out.writeLong(indicesCount);
     out.writeInt(ranges.size());
     Iterator<Range> it = ranges.iterator();
     while(it.hasNext()) {
@@ -215,7 +217,7 @@ public class SortedRanges implements Writable{
     private long startIndex;
     private long length;
         
-    public Range(long startIndex, long length) {
+    Range(long startIndex, long length) {
       if(length<0) {
         throw new RuntimeException("length can't be negative");
       }
@@ -223,7 +225,7 @@ public class SortedRanges implements Writable{
       this.length = length;
     }
     
-    public Range() {
+    Range() {
       this(0,0);
     }
     
@@ -231,7 +233,7 @@ public class SortedRanges implements Writable{
      * Get the start index. Start index in inclusive.
      * @return startIndex. 
      */
-    public long getStartIndex() {
+    long getStartIndex() {
       return startIndex;
     }
     
@@ -239,7 +241,7 @@ public class SortedRanges implements Writable{
      * Get the end index. End index is exclusive.
      * @return endIndex.
      */
-    public long getEndIndex() {
+    long getEndIndex() {
       return startIndex + length;
     }
     
@@ -247,7 +249,7 @@ public class SortedRanges implements Writable{
     * Get Length.
     * @return length
     */
-    public long getLength() {
+    long getLength() {
       return length;
     }
     
@@ -256,7 +258,7 @@ public class SortedRanges implements Writable{
      * @return <code>true</code> if empty
      *         <code>false</code> otherwise.
      */
-    public boolean isEmpty() {
+    boolean isEmpty() {
       return length==0;
     }
     
@@ -299,17 +301,25 @@ public class SortedRanges implements Writable{
   /**
    * Index Iterator which skips the stored ranges.
    */
-  private class SkipRangeIterator implements Iterator<Long> {
-    Iterator<Range> rangeIterator = ranges.iterator();
+  static class SkipRangeIterator implements Iterator<Long> {
+    Iterator<Range> rangeIterator;
     Range range = new Range();
     long currentIndex = -1;
     
+    /**
+     * Constructor
+     * @param rangeIterator the iterator which gives the ranges.
+     */
+    SkipRangeIterator(Iterator<Range> rangeIterator) {
+      this.rangeIterator = rangeIterator;
+    }
+    
     /**
      * Returns true till the index reaches Long.MAX_VALUE.
      * @return <code>true</code> next index exists.
      *         <code>false</code> otherwise.
      */
-    public boolean hasNext() {
+    public synchronized boolean hasNext() {
       return currentIndex<Long.MAX_VALUE;
     }
     
@@ -338,6 +348,15 @@ public class SortedRanges implements Writable{
       }
     }
     
+    /**
+     * Get whether all the ranges have been skipped.
+     * @return <code>true</code> if all ranges have been skipped.
+     *         <code>false</code> otherwise.
+     */
+    synchronized boolean skippedAllRanges() {
+      return !rangeIterator.hasNext() && currentIndex>=range.getEndIndex();
+    }
+    
     /**
      * Remove is not supported. Doesn't apply.
      */

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

@@ -73,6 +73,7 @@ abstract class Task implements Writable, Configurable {
     REDUCE_INPUT_GROUPS,
     REDUCE_INPUT_RECORDS,
     REDUCE_OUTPUT_RECORDS,
+    REDUCE_SKIPPED_GROUPS,
     REDUCE_SKIPPED_RECORDS
   }
   

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

@@ -179,6 +179,7 @@ class TaskInProgress {
   void init(JobID jobId) {
     this.startTime = System.currentTimeMillis();
     this.id = new TaskID(jobId, isMapTask(), partition);
+    this.skipping = startSkipping();
   }
 
   ////////////////////////////////////
@@ -489,10 +490,7 @@ class TaskInProgress {
       machinesWhereFailed.add(trackerHostName);
       LOG.debug("TaskInProgress adding" + status.getNextRecordRange());
       failedRanges.add(status.getNextRecordRange());
-      if(SkipBadRecords.getEnabled(conf) && 
-          numTaskFailures>=SkipBadRecords.getAttemptsToStartSkipping(conf)) {
-        skipping = true;
-      }
+      skipping = startSkipping();
     } else {
       numKilledTasks++;
     }
@@ -502,6 +500,17 @@ class TaskInProgress {
       kill();
     }
   }
+  
+  /**
+   * Get whether to start skipping mode. 
+   */
+  private boolean startSkipping() {
+    if(SkipBadRecords.getEnabled(conf) && 
+        numTaskFailures>=SkipBadRecords.getAttemptsToStartSkipping(conf)) {
+      return true;
+    }
+    return false;
+  }
 
   /**
    * Finalize the <b>completed</b> task; note that this might not be the first 

src/mapred/org/apache/hadoop/mapred/Task_Counter.properties

@@ -13,5 +13,6 @@ REDUCE_INPUT_GROUPS.name=      Reduce input groups
 REDUCE_INPUT_RECORDS.name=     Reduce input records
 REDUCE_OUTPUT_RECORDS.name=    Reduce output records
 REDUCE_SKIPPED_RECORDS.name=   Reduce skipped records
+REDUCE_SKIPPED_GROUPS.name=    Reduce skipped groups
 
 

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

@@ -35,7 +35,9 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.fs.FileUtil;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.SequenceFile;
 import org.apache.hadoop.io.Text;
+import org.apache.hadoop.util.ReflectionUtils;
 
 public class TestBadRecords extends ClusterMapReduceTestCase {
   
@@ -71,6 +73,7 @@ public class TestBadRecords extends ClusterMapReduceTestCase {
     conf.setNumReduceTasks(1);
     conf.setInt("mapred.task.timeout", 30*1000);
     
+    SkipBadRecords.setAttemptsToStartSkipping(conf,0);
     //the no of attempts to successfully complete the task depends 
     //on the no of bad records.
     conf.setMaxMapAttempts(SkipBadRecords.getAttemptsToStartSkipping(conf)+1+
@@ -105,6 +108,59 @@ public class TestBadRecords extends ClusterMapReduceTestCase {
     throws Exception{
     LOG.info(runningJob.getCounters().toString());
     assertTrue(runningJob.isSuccessful());
+    
+    //validate counters
+    Counters counters = runningJob.getCounters();
+    assertEquals(counters.findCounter(Task.Counter.MAP_SKIPPED_RECORDS).
+        getCounter(),mapperBadRecords.size());
+    
+    int mapRecs = input.size() - mapperBadRecords.size();
+    assertEquals(counters.findCounter(Task.Counter.MAP_INPUT_RECORDS).
+        getCounter(),mapRecs);
+    assertEquals(counters.findCounter(Task.Counter.MAP_OUTPUT_RECORDS).
+        getCounter(),mapRecs);
+    
+    int redRecs = mapRecs - redBadRecords.size();
+    assertEquals(counters.findCounter(Task.Counter.REDUCE_SKIPPED_RECORDS).
+        getCounter(),redBadRecords.size());
+    assertEquals(counters.findCounter(Task.Counter.REDUCE_SKIPPED_GROUPS).
+        getCounter(),redBadRecords.size());
+    assertEquals(counters.findCounter(Task.Counter.REDUCE_INPUT_GROUPS).
+        getCounter(),redRecs);
+    assertEquals(counters.findCounter(Task.Counter.REDUCE_INPUT_RECORDS).
+        getCounter(),redRecs);
+    assertEquals(counters.findCounter(Task.Counter.REDUCE_OUTPUT_RECORDS).
+        getCounter(),redRecs);
+    
+    //validate skipped records
+    Path skipDir = SkipBadRecords.getSkipOutputPath(conf);
+    Path[] skips = FileUtil.stat2Paths(getFileSystem().listStatus(skipDir));
+    List<String> mapSkipped = new ArrayList<String>();
+    List<String> redSkipped = new ArrayList<String>();
+    for(Path skipPath : skips) {
+      LOG.info("skipPath: " + skipPath);
+      
+      SequenceFile.Reader reader = new SequenceFile.Reader(
+          getFileSystem(), skipPath, conf);
+      Object key = ReflectionUtils.newInstance(reader.getKeyClass(), conf);
+      Object value = ReflectionUtils.newInstance(reader.getValueClass(), 
+          conf);
+      key = reader.next(key);
+      while(key!=null) {
+        value = reader.getCurrentValue(value);
+        LOG.debug("key:"+key+" value:"+value.toString());
+        if(skipPath.getName().contains("_r_")) {
+          redSkipped.add(value.toString());
+        } else {
+          mapSkipped.add(value.toString());
+        }
+        key = reader.next(key);
+      }
+      reader.close();
+    }
+    assertTrue(mapSkipped.containsAll(mapperBadRecords));
+    assertTrue(redSkipped.containsAll(redBadRecords));
+    
     Path[] outputFiles = FileUtil.stat2Paths(
         getFileSystem().listStatus(getOutputDir(),
         new OutputLogFilter()));