diff --git a/src/main/java/org/apache/hadoop/mrunit/MapDriverBase.java b/src/main/java/org/apache/hadoop/mrunit/MapDriverBase.java
index 2691648..73c1293 100644
--- a/src/main/java/org/apache/hadoop/mrunit/MapDriverBase.java
+++ b/src/main/java/org/apache/hadoop/mrunit/MapDriverBase.java
@@ -29,14 +29,14 @@
 import org.apache.hadoop.mrunit.types.Pair;
 
 /**
- * Harness that allows you to test a Mapper instance. You provide the input
- * (k, v)* pairs that should be sent to the Mapper, and outputs you expect to be
+ * Harness that allows you to test a Mapper instance. You provide the input (k,
+ * v)* pairs that should be sent to the Mapper, and outputs you expect to be
  * sent by the Mapper to the collector for those inputs. By calling runTest(),
  * the harness will deliver the input to the Mapper and will check its outputs
  * against the expected results.
  */
 public abstract class MapDriverBase<K1, V1, K2, V2, T extends MapDriverBase<K1, V1, K2, V2, T>>
-    extends TestDriver<K1, V1, K2, V2, T> {
+    extends TestDriver<K2, V2, T> {
 
   public static final Log LOG = LogFactory.getLog(MapDriverBase.class);
 
@@ -53,16 +53,17 @@
 
   /**
    * Sets the input key to send to the mapper
-   *
+   * 
    * @param key
    * @deprecated MRUNIT-64. Moved to list implementation to support multiple
-   *             input (k, v)*. Replaced by {@link #setInput},
-   *             {@link #addInput}, and {@link #addAll}
+   *             input (k, v)*. Replaced by {@link #setInput}, {@link #addInput}
+   *             , and {@link #addAll}
    */
   @Deprecated
   public void setInputKey(final K1 key) {
     inputKey = copy(key);
   }
+
   @Deprecated
   public K1 getInputKey() {
     return inputKey;
@@ -70,16 +71,17 @@
 
   /**
    * Sets the input value to send to the mapper
-   *
+   * 
    * @param val
    * @deprecated MRUNIT-64. Moved to list implementation to support multiple
-   *             input (k, v)*. Replaced by {@link #setInput},
-   *             {@link #addInput}, and {@link #addAll}
+   *             input (k, v)*. Replaced by {@link #setInput}, {@link #addInput}
+   *             , and {@link #addAll}
    */
   @Deprecated
   public void setInputValue(final V1 val) {
     inputVal = copy(val);
   }
+
   @Deprecated
   public V1 getInputValue() {
     return inputVal;
@@ -87,15 +89,15 @@
 
   /**
    * Sets the input to send to the mapper
-   *
+   * 
    */
   public void setInput(final K1 key, final V1 val) {
-  	setInput(new Pair<K1, V1>(key, val));
+    setInput(new Pair<K1, V1>(key, val));
   }
 
   /**
    * Sets the input to send to the mapper
-   *
+   * 
    * @param inputRecord
    *          a (key, val) pair
    */
@@ -110,7 +112,7 @@
 
   /**
    * Adds an input to send to the mapper
-   *
+   * 
    * @param key
    * @param val
    */
@@ -120,7 +122,7 @@
 
   /**
    * Adds an input to send to the mapper
-   *
+   * 
    * @param input
    *          a (K, V) pair
    */
@@ -130,7 +132,7 @@
 
   /**
    * Adds list of inputs to send to the mapper
-   *
+   * 
    * @param inputs
    *          list of (K, V) pairs
    */
@@ -150,7 +152,7 @@
   /**
    * Expects an input of the form "key \t val" Forces the Mapper input types to
    * Text.
-   *
+   * 
    * @param input
    *          A string of the form "key \t val".
    * @deprecated No replacement due to lack of type safety and incompatibility
@@ -171,7 +173,7 @@
 
   /**
    * Identical to setInputKey() but with fluent programming style
-   *
+   * 
    * @return this
    * @deprecated MRUNIT-64. Moved to list implementation to support multiple
    *             input (k, v)*. Replaced by {@link #withInput} and
@@ -185,7 +187,7 @@
 
   /**
    * Identical to setInputValue() but with fluent programming style
-   *
+   * 
    * @param val
    * @return this
    * @deprecated MRUNIT-64. Moved to list implementation to support multiple
@@ -199,9 +201,9 @@
   }
 
   /**
-   * Similar to setInput() but uses addInput() instead so accumulates values, and returns
-   * the class instance, conforming to the fluent programming style
-   *
+   * Similar to setInput() but uses addInput() instead so accumulates values,
+   * and returns the class instance, conforming to the fluent programming style
+   * 
    * @return this
    */
   public T withInput(final K1 key, final V1 val) {
@@ -211,7 +213,7 @@
 
   /**
    * Identical to setInput() but returns self for fluent programming style
-   *
+   * 
    * @param inputRecord
    * @return this
    */
@@ -222,7 +224,7 @@
 
   /**
    * Identical to setInputFromString, but with a fluent programming style
-   *
+   * 
    * @param input
    *          A string of the form "key \t val". Trims any whitespace.
    * @return this
@@ -237,7 +239,7 @@
 
   /**
    * Identical to addAll() but returns self for fluent programming style
-   *
+   * 
    * @param inputRecords
    * @return this
    */
@@ -254,7 +256,8 @@
   }
 
   /**
-   * @param mapInputPath Path which is to be passed to the mappers InputSplit
+   * @param mapInputPath
+   *          Path which is to be passed to the mappers InputSplit
    */
   public void setMapInputPath(Path mapInputPath) {
     this.mapInputPath = mapInputPath;
@@ -262,7 +265,7 @@
 
   /**
    * @param mapInputPath
-   *       The Path object which will be given to the mapper
+   *          The Path object which will be given to the mapper
    * @return this
    */
   public final T withMapInputPath(Path mapInputPath) {
@@ -289,9 +292,8 @@
 
     if (driverReused()) {
       throw new IllegalStateException("Driver reuse not allowed");
-    }
-    else {
-     setUsedOnceStatus();
+    } else {
+      setUsedOnceStatus();
     }
   }
 
@@ -301,7 +303,8 @@
   @Override
   protected void printPreTestDebugLog() {
     for (Pair<K1, V1> input : inputs) {
-      LOG.debug("Mapping input (" + input.getFirst() + ", " + input.getSecond() + ")");
+      LOG.debug("Mapping input (" + input.getFirst() + ", " + input.getSecond()
+          + ")");
     }
   }
 
diff --git a/src/main/java/org/apache/hadoop/mrunit/MapOutputShuffler.java b/src/main/java/org/apache/hadoop/mrunit/MapOutputShuffler.java
new file mode 100644
index 0000000..fec593f
--- /dev/null
+++ b/src/main/java/org/apache/hadoop/mrunit/MapOutputShuffler.java
@@ -0,0 +1,113 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mrunit;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mrunit.types.Pair;
+
+import java.util.*;
+
+public class MapOutputShuffler<K, V> {
+  private final Configuration configuration;
+  private final Comparator<K> outputKeyComparator;
+  private final Comparator<K> outputValueGroupingComparator;
+
+  public MapOutputShuffler(final Configuration configuration,
+      final Comparator<K> outputKeyComparator,
+      final Comparator<K> outputValueGroupingComparator) {
+    this.configuration = configuration;
+    this.outputKeyComparator = outputKeyComparator;
+    this.outputValueGroupingComparator = outputValueGroupingComparator;
+  }
+
+  public List<Pair<K, List<V>>> shuffle(final List<Pair<K, V>> mapOutputs) {
+
+    final Comparator<K> keyOrderComparator;
+    final Comparator<K> keyGroupComparator;
+
+    if (mapOutputs.isEmpty()) {
+      return Collections.emptyList();
+    }
+
+    // JobConf needs the map output key class to work out the
+    // comparator to use
+
+    JobConf conf = new JobConf(configuration != null ? configuration
+        : new Configuration());
+    K firstKey = mapOutputs.get(0).getFirst();
+    conf.setMapOutputKeyClass(firstKey.getClass());
+
+    // get the ordering comparator or work out from conf
+    if (outputKeyComparator == null) {
+      keyOrderComparator = conf.getOutputKeyComparator();
+    } else {
+      keyOrderComparator = outputKeyComparator;
+    }
+
+    // get the grouping comparator or work out from conf
+    if (outputValueGroupingComparator == null) {
+      keyGroupComparator = conf.getOutputValueGroupingComparator();
+    } else {
+      keyGroupComparator = outputValueGroupingComparator;
+    }
+
+    // sort the map outputs according to their keys
+    Collections.sort(mapOutputs, new Comparator<Pair<K, V>>() {
+      public int compare(final Pair<K, V> o1, final Pair<K, V> o2) {
+        return keyOrderComparator.compare(o1.getFirst(), o2.getFirst());
+      }
+    });
+
+    // apply grouping comparator to create groups
+    final Map<K, List<Pair<K, V>>> groupedByKey = new LinkedHashMap<K, List<Pair<K, V>>>();
+
+    List<Pair<K, V>> groupedKeyList = null;
+    Pair<K, V> previous = null;
+
+    for (final Pair<K, V> mapOutput : mapOutputs) {
+      if (previous == null
+          || keyGroupComparator.compare(previous.getFirst(),
+              mapOutput.getFirst()) != 0) {
+        groupedKeyList = new ArrayList<Pair<K, V>>();
+        groupedByKey.put(mapOutput.getFirst(), groupedKeyList);
+      }
+      groupedKeyList.add(mapOutput);
+      previous = mapOutput;
+    }
+
+    // populate output list
+    final List<Pair<K, List<V>>> outputKeyValuesList = new ArrayList<Pair<K, List<V>>>();
+    for (final Map.Entry<K, List<Pair<K, V>>> groupedByKeyEntry : groupedByKey
+        .entrySet()) {
+
+      // create list to hold values for the grouped key
+      final List<V> valuesList = new ArrayList<V>();
+      for (final Pair<K, V> pair : groupedByKeyEntry.getValue()) {
+        valuesList.add(pair.getSecond());
+      }
+
+      // add key and values to output list
+      outputKeyValuesList.add(new Pair<K, List<V>>(groupedByKeyEntry.getKey(),
+          valuesList));
+    }
+
+    return outputKeyValuesList;
+  }
+}
diff --git a/src/main/java/org/apache/hadoop/mrunit/MapReduceDriver.java b/src/main/java/org/apache/hadoop/mrunit/MapReduceDriver.java
index e9f76c5..518584c 100644
--- a/src/main/java/org/apache/hadoop/mrunit/MapReduceDriver.java
+++ b/src/main/java/org/apache/hadoop/mrunit/MapReduceDriver.java
@@ -17,21 +17,17 @@
  */
 package org.apache.hadoop.mrunit;
 
-import static org.apache.hadoop.mrunit.internal.util.ArgumentChecker.returnNonNull;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.mapred.*;
+import org.apache.hadoop.mrunit.internal.counters.CounterWrapper;
+import org.apache.hadoop.mrunit.types.Pair;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.mapred.Counters;
-import org.apache.hadoop.mapred.InputFormat;
-import org.apache.hadoop.mapred.Mapper;
-import org.apache.hadoop.mapred.OutputFormat;
-import org.apache.hadoop.mapred.Reducer;
-import org.apache.hadoop.mrunit.internal.counters.CounterWrapper;
-import org.apache.hadoop.mrunit.types.Pair;
+import static org.apache.hadoop.mrunit.internal.util.ArgumentChecker.returnNonNull;
 
 /**
  * Harness that allows you to test a Mapper and a Reducer instance together
@@ -42,7 +38,7 @@
  * Reducer (without checking them), and will check the Reducer's outputs against
  * the expected results. This is designed to handle the (k, v)* -> (k, v)* case
  * from the Mapper/Reducer pair, representing a single unit test.
- *
+ * 
  * If a combiner is specified, then it will be run exactly once after the Mapper
  * and before the Reducer.
  */
@@ -86,7 +82,7 @@
 
   /**
    * Sets the counters object to use for this test.
-   *
+   * 
    * @param ctrs
    *          The counters object to use.
    */
@@ -104,7 +100,7 @@
 
   /**
    * Set the Mapper instance to use with this test driver
-   *
+   * 
    * @param m
    *          the Mapper instance to use
    */
@@ -128,7 +124,7 @@
 
   /**
    * Sets the reducer object to use for this test
-   *
+   * 
    * @param r
    *          The reducer object to use
    */
@@ -138,7 +134,7 @@
 
   /**
    * Identical to setReducer(), but with fluent programming style
-   *
+   * 
    * @param r
    *          The Reducer to use
    * @return this
@@ -158,7 +154,7 @@
 
   /**
    * Sets the reducer object to use as a combiner for this test
-   *
+   * 
    * @param c
    *          The combiner object to use
    */
@@ -168,7 +164,7 @@
 
   /**
    * Identical to setCombiner(), but with fluent programming style
-   *
+   * 
    * @param c
    *          The Combiner to use
    * @return this
@@ -189,7 +185,7 @@
   /**
    * Configure {@link Reducer} to output with a real {@link OutputFormat}. Set
    * {@link InputFormat} to read output back in for use with run* methods
-   *
+   * 
    * @param outputFormatClass
    * @param inputFormatClass
    * @return this for fluent style
@@ -203,47 +199,6 @@
     return this;
   }
 
-  /**
-   * The private class to manage starting the reduce phase is used for type
-   * genericity reasons. This class is used in the run() method.
-   */
-  private class ReducePhaseRunner<OUTKEY, OUTVAL> {
-    private List<Pair<OUTKEY, OUTVAL>> runReduce(
-        final List<Pair<K2, List<V2>>> inputs,
-        final Reducer<K2, V2, OUTKEY, OUTVAL> reducer) throws IOException {
-
-      final List<Pair<OUTKEY, OUTVAL>> reduceOutputs = new ArrayList<Pair<OUTKEY, OUTVAL>>();
-
-      if (!inputs.isEmpty()) {
-        if (LOG.isDebugEnabled()) {
-          final StringBuilder sb = new StringBuilder();
-          for (Pair<K2, List<V2>> input : inputs) {
-            formatValueList(input.getSecond(), sb);
-            LOG.debug("Reducing input (" + input.getFirst() + ", " + sb + ")");
-            sb.delete(0, sb.length());
-          }
-        }
-
-        final ReduceDriver<K2, V2, OUTKEY, OUTVAL> reduceDriver = ReduceDriver
-            .newReduceDriver(reducer).withCounters(getCounters())
-            .withConfiguration(getConfiguration()).withAll(inputs);
-
-        if (getOutputSerializationConfiguration() != null) {
-          reduceDriver
-              .withOutputSerializationConfiguration(getOutputSerializationConfiguration());
-        }
-
-        if (outputFormatClass != null) {
-          reduceDriver.withOutputFormat(outputFormatClass, inputFormatClass);
-        }
-
-        reduceOutputs.addAll(reduceDriver.run());
-      }
-
-      return reduceOutputs;
-    }
-  }
-
   @Override
   public List<Pair<K3, V3>> run() throws IOException {
     try {
@@ -251,6 +206,9 @@
       initDistributedCache();
       List<Pair<K2, V2>> mapOutputs = new ArrayList<Pair<K2, V2>>();
 
+      MapOutputShuffler<K2, V2> shuffler = new MapOutputShuffler<K2, V2>(
+          getConfiguration(), keyValueOrderComparator, keyGroupComparator);
+
       // run map component
       LOG.debug("Starting map phase with mapper: " + myMapper);
       mapOutputs.addAll(MapDriver.newMapDriver(myMapper)
@@ -258,18 +216,22 @@
           .withAll(inputList).withMapInputPath(getMapInputPath()).run());
 
       if (myCombiner != null) {
-        // User has specified a combiner. Run this and replace the mapper outputs
+        // User has specified a combiner. Run this and replace the mapper
+        // outputs
         // with the result of the combiner.
         LOG.debug("Starting combine phase with combiner: " + myCombiner);
-        mapOutputs = new ReducePhaseRunner<K2, V2>().runReduce(
-            shuffle(mapOutputs), myCombiner);
+        mapOutputs = new ReducePhaseRunner<K2, V2, K2, V2>(inputFormatClass,
+            getConfiguration(), counters,
+            getOutputSerializationConfiguration(), outputFormatClass)
+            .runReduce(shuffler.shuffle(mapOutputs), myCombiner);
       }
 
       // Run the reduce phase.
       LOG.debug("Starting reduce phase with reducer: " + myReducer);
 
-      return new ReducePhaseRunner<K3, V3>()
-          .runReduce(shuffle(mapOutputs),myReducer);
+      return new ReducePhaseRunner<K2, V2, K3, V3>(inputFormatClass,
+          getConfiguration(), counters, getOutputSerializationConfiguration(),
+          outputFormatClass).runReduce(shuffler.shuffle(mapOutputs), myReducer);
     } finally {
       cleanupDistributedCache();
     }
@@ -283,7 +245,7 @@
   /**
    * Returns a new MapReduceDriver without having to specify the generic types
    * on the right hand side of the object create statement.
-   *
+   * 
    * @return new MapReduceDriver
    */
   public static <K1, V1, K2, V2, K3, V3> MapReduceDriver<K1, V1, K2, V2, K3, V3> newMapReduceDriver() {
@@ -293,7 +255,7 @@
   /**
    * Returns a new MapReduceDriver without having to specify the generic types
    * on the right hand side of the object create statement.
-   *
+   * 
    * @param mapper
    *          passed to MapReduceDriver constructor
    * @param reducer
@@ -308,7 +270,7 @@
   /**
    * Returns a new MapReduceDriver without having to specify the generic types
    * on the right hand side of the object create statement.
-   *
+   * 
    * @param mapper
    *          passed to MapReduceDriver constructor
    * @param reducer
diff --git a/src/main/java/org/apache/hadoop/mrunit/MapReduceDriverBase.java b/src/main/java/org/apache/hadoop/mrunit/MapReduceDriverBase.java
index 25cc274..66545cc 100644
--- a/src/main/java/org/apache/hadoop/mrunit/MapReduceDriverBase.java
+++ b/src/main/java/org/apache/hadoop/mrunit/MapReduceDriverBase.java
@@ -17,15 +17,6 @@
  */
 package org.apache.hadoop.mrunit;
 
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.Comparator;
-import java.util.LinkedHashMap;
-import java.util.List;
-import java.util.Map;
-import java.util.Map.Entry;
-
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.fs.Path;
@@ -35,6 +26,11 @@
 import org.apache.hadoop.mrunit.types.Pair;
 import org.apache.hadoop.util.ReflectionUtils;
 
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Comparator;
+import java.util.List;
+
 /**
  * Harness that allows you to test a Mapper and a Reducer instance together You
  * provide the input key and value that should be sent to the Mapper, and
@@ -46,7 +42,7 @@
  * pair, representing a single unit test.
  */
 public abstract class MapReduceDriverBase<K1, V1, K2, V2, K3, V3, T extends MapReduceDriverBase<K1, V1, K2, V2, K3, V3, T>>
-    extends TestDriver<K1, V1, K3, V3, T> {
+    extends TestDriver<K3, V3, T> {
 
   public static final Log LOG = LogFactory.getLog(MapReduceDriverBase.class);
 
@@ -62,7 +58,7 @@
 
   /**
    * Adds an input to send to the mapper
-   *
+   * 
    * @param key
    * @param val
    */
@@ -72,7 +68,7 @@
 
   /**
    * Adds an input to send to the Mapper
-   *
+   * 
    * @param input
    *          The (k, v) pair to add to the input list.
    */
@@ -82,7 +78,7 @@
 
   /**
    * Adds input to send to the mapper
-   *
+   * 
    * @param inputs
    *          List of (k, v) pairs to add to the input list
    */
@@ -95,7 +91,7 @@
   /**
    * Expects an input of the form "key \t val" Forces the Mapper input types to
    * Text.
-   *
+   * 
    * @param input
    *          A string of the form "key \t val". Trims any whitespace.
    * @deprecated No replacement due to lack of type safety and incompatibility
@@ -114,33 +110,31 @@
 
   /**
    * Identical to addInput() but returns self for fluent programming style
-   *
+   * 
    * @param key
    * @param val
    * @return this
    */
-  public T withInput(final K1 key,
-      final V1 val) {
+  public T withInput(final K1 key, final V1 val) {
     addInput(key, val);
     return thisAsMapReduceDriver();
   }
 
   /**
    * Identical to addInput() but returns self for fluent programming style
-   *
+   * 
    * @param input
    *          The (k, v) pair to add
    * @return this
    */
-  public T withInput(
-      final Pair<K1, V1> input) {
+  public T withInput(final Pair<K1, V1> input) {
     addInput(input);
     return thisAsMapReduceDriver();
   }
 
   /**
    * Identical to addInputFromString, but with a fluent programming style
-   *
+   * 
    * @param input
    *          A string of the form "key \t val". Trims any whitespace.
    * @return this
@@ -148,21 +142,19 @@
    *             with non Text Writables
    */
   @Deprecated
-  public T withInputFromString(
-      final String input) {
+  public T withInputFromString(final String input) {
     addInputFromString(input);
     return thisAsMapReduceDriver();
   }
 
   /**
    * Identical to addAll() but returns self for fluent programming style
-   *
+   * 
    * @param inputs
    *          List of (k, v) pairs to add
    * @return this
    */
-  public T withAll(
-      final List<Pair<K1, V1>> inputs) {
+  public T withAll(final List<Pair<K1, V1>> inputs) {
     addAll(inputs);
     return thisAsMapReduceDriver();
   }
@@ -175,7 +167,8 @@
   }
 
   /**
-   * @param mapInputPath Path which is to be passed to the mappers InputSplit
+   * @param mapInputPath
+   *          Path which is to be passed to the mappers InputSplit
    */
   public void setMapInputPath(Path mapInputPath) {
     this.mapInputPath = mapInputPath;
@@ -183,7 +176,7 @@
 
   /**
    * @param mapInputPath
-   *       The Path object which will be given to the mapper
+   *          The Path object which will be given to the mapper
    * @return this
    */
   public final T withMapInputPath(Path mapInputPath) {
@@ -203,8 +196,7 @@
     }
     if (driverReused()) {
       throw new IllegalStateException("Driver reuse not allowed");
-    }
-    else {
+    } else {
       setUsedOnceStatus();
     }
   }
@@ -213,93 +205,13 @@
   public abstract List<Pair<K3, V3>> run() throws IOException;
 
   /**
-   * Take the outputs from the Mapper, combine all values for the same key, and
-   * sort them by key.
-   *
-   * @param mapOutputs
-   *          An unordered list of (key, val) pairs from the mapper
-   * @return the sorted list of (key, list(val))'s to present to the reducer
-   */
-  public List<Pair<K2, List<V2>>> shuffle(final List<Pair<K2, V2>> mapOutputs) {
-
-    final Comparator<K2> keyOrderComparator;
-    final Comparator<K2> keyGroupComparator;
-
-    if (mapOutputs.isEmpty()) {
-      return Collections.emptyList();
-    }
-
-    // JobConf needs the map output key class to work out the
-    // comparator to use
-    JobConf conf = new JobConf(getConfiguration());
-    K2 firstKey = mapOutputs.get(0).getFirst();
-    conf.setMapOutputKeyClass(firstKey.getClass());
-
-    // get the ordering comparator or work out from conf
-    if (keyValueOrderComparator == null) {
-      keyOrderComparator = conf.getOutputKeyComparator();
-    } else {
-      keyOrderComparator = this.keyValueOrderComparator;
-    }
-
-    // get the grouping comparator or work out from conf
-    if (this.keyGroupComparator == null) {
-      keyGroupComparator = conf.getOutputValueGroupingComparator();
-    } else {
-      keyGroupComparator = this.keyGroupComparator;
-    }
-
-    // sort the map outputs according to their keys
-    Collections.sort(mapOutputs, new Comparator<Pair<K2, V2>>() {
-      public int compare(final Pair<K2, V2> o1, final Pair<K2, V2> o2) {
-        return keyOrderComparator.compare(o1.getFirst(), o2.getFirst());
-      }
-    });
-
-    // apply grouping comparator to create groups
-    final Map<K2, List<Pair<K2, V2>>> groupedByKey =
-        new LinkedHashMap<K2, List<Pair<K2, V2>>>();
-
-    List<Pair<K2, V2>> groupedKeyList = null;
-    Pair<K2,V2> previous = null;
-
-    for (final Pair<K2, V2> mapOutput : mapOutputs) {
-      if (previous == null || keyGroupComparator
-          .compare(previous.getFirst(), mapOutput.getFirst()) != 0) {
-        groupedKeyList = new ArrayList<Pair<K2, V2>>();
-        groupedByKey.put(mapOutput.getFirst(), groupedKeyList);
-      }
-      groupedKeyList.add(mapOutput);
-      previous = mapOutput;
-    }
-
-    // populate output list
-    final List<Pair<K2, List<V2>>> outputKeyValuesList = new ArrayList<Pair<K2, List<V2>>>();
-    for (final Entry<K2, List<Pair<K2, V2>>> groupedByKeyEntry :
-            groupedByKey.entrySet()) {
-
-      // create list to hold values for the grouped key
-      final List<V2> valuesList = new ArrayList<V2>();
-      for (final Pair<K2, V2> pair : groupedByKeyEntry.getValue()) {
-        valuesList.add(pair.getSecond());
-      }
-
-      // add key and values to output list
-      outputKeyValuesList.add(new Pair<K2, List<V2>>(
-          groupedByKeyEntry.getKey(), valuesList));
-    }
-
-    return outputKeyValuesList;
-  }
-
-  /**
    * Set the key grouping comparator, similar to calling the following API calls
    * but passing a real instance rather than just the class:
    * <UL>
    * <LI>pre 0.20.1 API: {@link JobConf#setOutputValueGroupingComparator(Class)}
    * <LI>0.20.1+ API: {@link Job#setGroupingComparatorClass(Class)}
    * </UL>
-   *
+   * 
    * @param groupingComparator
    */
   public void setKeyGroupingComparator(
@@ -315,7 +227,7 @@
    * <LI>pre 0.20.1 API: {@link JobConf#setOutputKeyComparatorClass(Class)}
    * <LI>0.20.1+ API: {@link Job#setSortComparatorClass(Class)}
    * </UL>
-   *
+   * 
    * @param orderComparator
    */
   public void setKeyOrderComparator(final RawComparator<K2> orderComparator) {
@@ -326,7 +238,7 @@
   /**
    * Identical to {@link #setKeyGroupingComparator(RawComparator)}, but with a
    * fluent programming style
-   *
+   * 
    * @param groupingComparator
    *          Comparator to use in the shuffle stage for key grouping
    * @return this
@@ -339,7 +251,7 @@
   /**
    * Identical to {@link #setKeyOrderComparator(RawComparator)}, but with a
    * fluent programming style
-   *
+   * 
    * @param orderComparator
    *          Comparator to use in the shuffle stage for key value ordering
    * @return this
diff --git a/src/main/java/org/apache/hadoop/mrunit/MultipleInputsMapReduceDriver.java b/src/main/java/org/apache/hadoop/mrunit/MultipleInputsMapReduceDriver.java
new file mode 100644
index 0000000..e155e08
--- /dev/null
+++ b/src/main/java/org/apache/hadoop/mrunit/MultipleInputsMapReduceDriver.java
@@ -0,0 +1,482 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mrunit;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.mapred.*;
+import org.apache.hadoop.mrunit.internal.counters.CounterWrapper;
+import org.apache.hadoop.mrunit.internal.driver.MultipleInputsMapReduceDriverBase;
+import org.apache.hadoop.mrunit.types.Pair;
+
+import java.io.IOException;
+import java.util.*;
+
+import static org.apache.hadoop.mrunit.internal.util.ArgumentChecker.returnNonNull;
+
+/**
+ * Harness that allows you to test multiple Mappers and a Reducer instance
+ * together (along with an optional combiner). You provide the input keys and
+ * values that should be sent to each Mapper, and outputs you expect to be sent
+ * by the Reducer to the collector for those inputs. By calling runTest(), the
+ * harness will deliver the inputs to the respective Mappers, feed the
+ * intermediate results to the Reducer (without checking them), and will check
+ * the Reducer's outputs against the expected results.
+ * 
+ * If a combiner is specified, it will run exactly once after all the Mappers
+ * and before the Reducer
+ * 
+ * @param <K1>
+ *          The common map output key type
+ * @param <V1>
+ *          The common map output value type
+ * @param <K2>
+ *          The reduce output key type
+ * @param <V2>
+ *          The reduce output value type
+ */
+public class MultipleInputsMapReduceDriver<K1, V1, K2, V2>
+    extends
+        MultipleInputsMapReduceDriverBase<Mapper, K1, V1, K2, V2, MultipleInputsMapReduceDriver<K1, V1, K2, V2>> {
+  public static final Log LOG = LogFactory
+      .getLog(MultipleInputsMapReduceDriver.class);
+
+  private Set<Mapper> mappers = new HashSet<Mapper>();
+
+  /**
+   * Add a mapper to use with this test driver
+   * 
+   * @param mapper
+   *          The mapper instance to add
+   * @param <K>
+   *          The input key type to the mapper
+   * @param <V>
+   *          The input value type to the mapper
+   */
+  public <K, V> void addMapper(final Mapper<K, V, K1, V1> mapper) {
+    this.mappers.add(returnNonNull(mapper));
+  }
+
+  /**
+   * Identical to addMapper but supports a fluent programming style
+   * 
+   * @param mapper
+   *          The mapper instance to add
+   * @param <K>
+   *          The input key type to the mapper
+   * @param <V>
+   *          The input value type to the mapper
+   * @return this
+   */
+  public <K, V> MultipleInputsMapReduceDriver<K1, V1, K2, V2> withMapper(
+      final Mapper<K, V, K1, V1> mapper) {
+    addMapper(mapper);
+    return this;
+  }
+
+  /**
+   * @return The Mapper instances being used by this test
+   */
+  public Collection<Mapper> getMappers() {
+    return mappers;
+  }
+
+  private Reducer<K1, V1, K1, V1> combiner;
+
+  /**
+   * Set the combiner to use with this test driver
+   * 
+   * @param combiner
+   *          The combiner instance to use
+   */
+  public void setCombiner(final Reducer<K1, V1, K1, V1> combiner) {
+    this.combiner = returnNonNull(combiner);
+  }
+
+  /**
+   * Identical to setCombiner but supports a fluent programming style
+   * 
+   * @param combiner
+   *          The combiner instance to use
+   * @return this
+   */
+  public MultipleInputsMapReduceDriver<K1, V1, K2, V2> withCombiner(
+      final Reducer<K1, V1, K1, V1> combiner) {
+    setCombiner(combiner);
+    return this;
+  }
+
+  /**
+   * @return The combiner instance being used by this test
+   */
+  public Reducer<K1, V1, K1, V1> getCombiner() {
+    return combiner;
+  }
+
+  private Reducer<K1, V1, K2, V2> reducer;
+
+  /**
+   * Set the reducer to use with this test driver
+   * 
+   * @param reducer
+   *          The reducer instance to use
+   */
+  public void setReducer(final Reducer<K1, V1, K2, V2> reducer) {
+    this.reducer = returnNonNull(reducer);
+  }
+
+  /**
+   * Identical to setReducer but supports a fluent programming style
+   * 
+   * @param reducer
+   *          The reducer instance to use
+   * @return this
+   */
+  public MultipleInputsMapReduceDriver<K1, V1, K2, V2> withReducer(
+      final Reducer<K1, V1, K2, V2> reducer) {
+    setReducer(reducer);
+    return this;
+  }
+
+  /**
+   * @return Get the reducer instance being used by this test
+   */
+  public Reducer<K1, V1, K2, V2> getReducer() {
+    return reducer;
+  }
+
+  private Counters counters;
+
+  /**
+   * @return The counters used in this test
+   */
+  public Counters getCounters() {
+    return counters;
+  }
+
+  /**
+   * Sets the counters object to use for this test
+   * 
+   * @param counters
+   *          The counters object to use
+   */
+  public void setCounters(Counters counters) {
+    this.counters = counters;
+    counterWrapper = new CounterWrapper(counters);
+  }
+
+  /**
+   * Identical to setCounters but supports a fluent programming style
+   * 
+   * @param counters
+   *          The counters object to use
+   * @return this
+   */
+  public MultipleInputsMapReduceDriver<K1, V1, K2, V2> withCounter(
+      Counters counters) {
+    setCounters(counters);
+    return this;
+  }
+
+  private Class<? extends OutputFormat> outputFormatClass;
+
+  /**
+   * Configure {@link Reducer} to output with a real {@link OutputFormat}.
+   * 
+   * @param outputFormatClass
+   *          The OutputFormat class
+   * @return this
+   */
+  public MultipleInputsMapReduceDriver<K1, V1, K2, V2> withOutputFormat(
+      final Class<? extends OutputFormat> outputFormatClass) {
+    this.outputFormatClass = returnNonNull(outputFormatClass);
+    return this;
+  }
+
+  private Class<? extends InputFormat> inputFormatClass;
+
+  /**
+   * Set the InputFormat
+   * 
+   * @param inputFormatClass
+   *          The InputFormat class
+   * @return this
+   */
+  public MultipleInputsMapReduceDriver<K1, V1, K2, V2> withInputFormat(
+      final Class<? extends InputFormat> inputFormatClass) {
+    this.inputFormatClass = returnNonNull(inputFormatClass);
+    return this;
+  }
+
+  /**
+   * Construct a driver with the specified Reducer. Note that a Combiner can be
+   * set separately.
+   * 
+   * @param reducer
+   *          The reducer to use
+   */
+  public MultipleInputsMapReduceDriver(Reducer<K1, V1, K2, V2> reducer) {
+    this();
+    this.reducer = reducer;
+  }
+
+  /**
+   * Construct a driver with the specified Combiner and Reducers
+   * 
+   * @param combiner
+   *          The combiner to use
+   * @param reducer
+   *          The reducer to use
+   */
+  public MultipleInputsMapReduceDriver(Reducer<K1, V1, K1, V1> combiner,
+                                       Reducer<K1, V1, K2, V2> reducer) {
+    this(reducer);
+    this.combiner = combiner;
+  }
+
+  /**
+   * Construct a driver without specifying a Combiner nor a Reducer. Note that
+   * these can be set with the appropriate set methods and that at least the
+   * Reducer must be set.
+   */
+  public MultipleInputsMapReduceDriver() {
+    setCounters(new Counters());
+  }
+
+  /**
+   * Static factory-style method to construct a driver instance with the
+   * specified Combiner and Reducer
+   * 
+   * @param combiner
+   *          The combiner to use
+   * @param reducer
+   *          The reducer to use
+   * @param <K1>
+   *          The common output key type of the mappers
+   * @param <V1>
+   *          The common output value type of the mappers
+   * @param <K2>
+   *          The output key type of the reducer
+   * @param <V2>
+   *          The output value type of the reducer
+   * @return this to support fluent programming style
+   */
+  public static <K1, V1, K2, V2> MultipleInputsMapReduceDriver<K1, V1, K2, V2> newMultipleInputMapReduceDriver(
+      final Reducer<K1, V1, K1, V1> combiner,
+      final Reducer<K1, V1, K2, V2> reducer) {
+    return new MultipleInputsMapReduceDriver<K1, V1, K2, V2>(combiner, reducer);
+  }
+
+  /**
+   * Static factory-style method to construct a driver instance with the
+   * specified Reducer
+   * 
+   * @param reducer
+   *          The reducer to use
+   * @param <K1>
+   *          The common output key type of the mappers
+   * @param <V1>
+   *          The common output value type of the mappers
+   * @param <K2>
+   *          The output key type of the reducer
+   * @param <V2>
+   *          The output value type of the reducer
+   * @return this to support fluent programming style
+   */
+  public static <K1, V1, K2, V2> MultipleInputsMapReduceDriver<K1, V1, K2, V2> newMultipleInputMapReduceDriver(
+      final Reducer<K1, V1, K2, V2> reducer) {
+    return new MultipleInputsMapReduceDriver<K1, V1, K2, V2>(reducer);
+  }
+
+  /**
+   * Static factory-style method to construct a driver instance without
+   * specifying a Combiner nor a Reducer. Note that these can be set separately
+   * by using the appropriate set (or with) methods and that at least a Reducer
+   * must be set
+   * 
+   * @param <K1>
+   *          The common output key type of the mappers
+   * @param <V1>
+   *          The common output value type of the mappers
+   * @param <K2>
+   *          The output key type of the reducer
+   * @param <V2>
+   *          The output value type of the reducer
+   * @return this to support fluent programming style
+   */
+  public static <K1, V1, K2, V2> MultipleInputsMapReduceDriver<K1, V1, K2, V2> newMultipleInputMapReduceDriver() {
+    return new MultipleInputsMapReduceDriver<K1, V1, K2, V2>();
+  }
+
+  /**
+   * Add the specified (key, val) pair to the specified mapper
+   * 
+   * @param mapper
+   *          The mapper to add the input pair to
+   * @param key
+   *          The key
+   * @param val
+   *          The value
+   * @param <K>
+   *          The type of the key
+   * @param <V>
+   *          The type of the value
+   */
+  public <K, V> void addInput(final Mapper<K, V, K1, V1> mapper, final K key,
+      final V val) {
+    super.addInput(mapper, key, val);
+  }
+
+  /**
+   * Add the specified input pair to the specified mapper
+   * 
+   * @param mapper
+   *          The mapper to add the input pair to
+   * @param input
+   *          The (k,v) pair to add
+   * @param <K>
+   *          The type of the key
+   * @param <V>
+   *          The type of the value
+   */
+  public <K, V> void addInput(final Mapper<K, V, K1, V1> mapper,
+      final Pair<K, V> input) {
+    super.addInput(mapper, input);
+  }
+
+  /**
+   * Add the specified input pairs to the specified mapper
+   * 
+   * @param mapper
+   *          The mapper to add the input pairs to
+   * @param inputs
+   *          The (k, v) pairs to add
+   * @param <K>
+   *          The type of the key
+   * @param <V>
+   *          The type of the value
+   */
+  public <K, V> void addAll(final Mapper<K, V, K1, V1> mapper,
+      final List<Pair<K, V>> inputs) {
+    super.addAll(mapper, inputs);
+  }
+
+  /**
+   * Identical to addInput but supports fluent programming style
+   * 
+   * @param mapper
+   *          The mapper to add the input pair to
+   * @param key
+   *          The key
+   * @param val
+   *          The value
+   * @param <K>
+   *          The type of the key
+   * @param <V>
+   *          The type of the value
+   * @return this
+   */
+  public <K, V> MultipleInputsMapReduceDriver<K1, V1, K2, V2> withInput(
+      final Mapper<K, V, K1, V1> mapper, final K key, final V val) {
+    return super.withInput(mapper, key, val);
+  }
+
+  /**
+   * Identical to addInput but supports fluent programming style
+   * 
+   * @param mapper
+   *          The mapper to add the input pairs to
+   * @param inputs
+   *          The (k, v) pairs to add
+   * @param <K>
+   *          The type of the key
+   * @param <V>
+   *          The type of the value
+   * @return this
+   */
+  public <K, V> MultipleInputsMapReduceDriver<K1, V1, K2, V2> withInput(
+      final Mapper<K, V, K1, V1> mapper, final Pair<K, V> input) {
+    return super.withInput(mapper, input);
+  }
+
+  /**
+   * Identical to addInput but supports fluent programming style
+   * 
+   * @param mapper
+   *          The mapper to add the input pairs to
+   * @param inputs
+   *          The (k, v) pairs to add
+   * @param <K>
+   *          The type of the key
+   * @param <V>
+   *          The type of the value
+   * @return this
+   */
+  public <K, V> MultipleInputsMapReduceDriver<K1, V1, K2, V2> withAll(
+      final Mapper<K, V, K1, V1> mapper, final List<Pair<K, V>> inputs) {
+    return super.withAll(mapper, inputs);
+  }
+
+  @Override
+  protected void preRunChecks(Set<Mapper> mappers, Object reducer) {
+    if (mappers.isEmpty()) {
+      throw new IllegalStateException("No mappers were provided");
+    }
+    super.preRunChecks(mappers, reducer);
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  public List<Pair<K2, V2>> run() throws IOException {
+    try {
+      preRunChecks(mappers, reducer);
+      initDistributedCache();
+
+      List<Pair<K1, V1>> outputs = new ArrayList<Pair<K1, V1>>();
+
+      MapOutputShuffler<K1, V1> shuffler = new MapOutputShuffler<K1, V1>(
+          getConfiguration(), keyValueOrderComparator, keyGroupComparator);
+
+      for (Mapper mapper : mappers) {
+        MapDriver mapDriver = MapDriver.newMapDriver(mapper);
+        mapDriver.setCounters(counters);
+        mapDriver.setConfiguration(getConfiguration());
+        mapDriver.addAll(inputs.get(mapper));
+        mapDriver.withMapInputPath(getMapInputPath(mapper));
+        outputs.addAll(mapDriver.run());
+      }
+
+      if (combiner != null) {
+        LOG.debug("Starting combine phase with combiner: " + combiner);
+        outputs = new ReducePhaseRunner<K1, V1, K1, V1>(inputFormatClass,
+            getConfiguration(), counters,
+            getOutputSerializationConfiguration(), outputFormatClass)
+            .runReduce(shuffler.shuffle(outputs), combiner);
+      }
+
+      LOG.debug("Starting reduce phase with reducer: " + reducer);
+
+      return new ReducePhaseRunner<K1, V1, K2, V2>(inputFormatClass,
+          getConfiguration(), counters, getOutputSerializationConfiguration(),
+          outputFormatClass).runReduce(shuffler.shuffle(outputs), reducer);
+    } finally {
+      cleanupDistributedCache();
+    }
+  }
+}
diff --git a/src/main/java/org/apache/hadoop/mrunit/PipelineMapReduceDriver.java b/src/main/java/org/apache/hadoop/mrunit/PipelineMapReduceDriver.java
index 9cf4e42..3c9d297 100644
--- a/src/main/java/org/apache/hadoop/mrunit/PipelineMapReduceDriver.java
+++ b/src/main/java/org/apache/hadoop/mrunit/PipelineMapReduceDriver.java
@@ -38,23 +38,23 @@
  * workflow, as well as a set of (key, value) pairs to pass in to the first
  * Mapper. You can also specify the outputs you expect to be sent to the final
  * Reducer in the pipeline.
- *
+ * 
  * By calling runTest(), the harness will deliver the input to the first Mapper,
  * feed the intermediate results to the first Reducer (without checking them),
  * and proceed to forward this data along to subsequent Mapper/Reducer jobs in
  * the pipeline until the final Reducer. The last Reducer's outputs are checked
  * against the expected results.
- *
+ * 
  * This is designed for slightly more complicated integration tests than the
  * MapReduceDriver, which is for smaller unit tests.
- *
+ * 
  * (K1, V1) in the type signature refer to the types associated with the inputs
  * to the first Mapper. (K2, V2) refer to the types associated with the final
  * Reducer's output. No intermediate types are specified.
  */
 @SuppressWarnings("rawtypes")
 public class PipelineMapReduceDriver<K1, V1, K2, V2> extends
-    TestDriver<K1, V1, K2, V2, PipelineMapReduceDriver<K1, V1, K2, V2>> {
+    TestDriver<K2, V2, PipelineMapReduceDriver<K1, V1, K2, V2>> {
 
   public static final Log LOG = LogFactory
       .getLog(PipelineMapReduceDriver.class);
@@ -83,7 +83,7 @@
 
   /**
    * Sets the counters object to use for this test.
-   *
+   * 
    * @param ctrs
    *          The counters object to use.
    */
@@ -102,7 +102,7 @@
   /**
    * Add a Mapper and Reducer instance to the pipeline to use with this test
    * driver
-   *
+   * 
    * @param m
    *          The Mapper instance to add to the pipeline
    * @param r
@@ -115,7 +115,7 @@
   /**
    * Add a Mapper and Reducer instance to the pipeline to use with this test
    * driver
-   *
+   * 
    * @param p
    *          The Mapper and Reducer instances to add to the pipeline
    */
@@ -126,7 +126,7 @@
   /**
    * Add a Mapper and Reducer instance to the pipeline to use with this test
    * driver using fluent style
-   *
+   * 
    * @param m
    *          The Mapper instance to use
    * @param r
@@ -141,7 +141,7 @@
   /**
    * Add a Mapper and Reducer instance to the pipeline to use with this test
    * driver using fluent style
-   *
+   * 
    * @param p
    *          The Mapper and Reducer instances to add to the pipeline
    */
@@ -160,7 +160,7 @@
 
   /**
    * Adds an input to send to the mapper
-   *
+   * 
    * @param key
    * @param val
    */
@@ -170,7 +170,7 @@
 
   /**
    * Adds list of inputs to send to the mapper
-   *
+   * 
    * @param inputs
    *          list of (K, V) pairs
    */
@@ -182,7 +182,7 @@
 
   /**
    * Identical to addInput() but returns self for fluent programming style
-   *
+   * 
    * @param key
    * @param val
    * @return this
@@ -195,7 +195,7 @@
 
   /**
    * Adds an input to send to the Mapper
-   *
+   * 
    * @param input
    *          The (k, v) pair to add to the input list.
    */
@@ -205,7 +205,7 @@
 
   /**
    * Identical to addInput() but returns self for fluent programming style
-   *
+   * 
    * @param input
    *          The (k, v) pair to add
    * @return this
@@ -219,7 +219,7 @@
   /**
    * Expects an input of the form "key \t val" Forces the Mapper input types to
    * Text.
-   *
+   * 
    * @param input
    *          A string of the form "key \t val". Trims any whitespace.
    * @deprecated No replacement due to lack of type safety and incompatibility
@@ -233,7 +233,7 @@
 
   /**
    * Identical to addInputFromString, but with a fluent programming style
-   *
+   * 
    * @param input
    *          A string of the form "key \t val". Trims any whitespace.
    * @return this
@@ -249,8 +249,9 @@
 
   /**
    * Identical to addAll() but returns self for fluent programming style
-   *
-   * @param inputRecords input key/value pairs
+   * 
+   * @param inputRecords
+   *          input key/value pairs
    * @return this
    */
   public PipelineMapReduceDriver<K1, V1, K2, V2> withAll(
@@ -267,7 +268,8 @@
   }
 
   /**
-   * @param mapInputPath Path which is to be passed to the mappers InputSplit
+   * @param mapInputPath
+   *          Path which is to be passed to the mappers InputSplit
    */
   public void setMapInputPath(Path mapInputPath) {
     this.mapInputPath = mapInputPath;
@@ -275,10 +277,11 @@
 
   /**
    * @param mapInputPath
-   *       The Path object which will be given to the mapper
+   *          The Path object which will be given to the mapper
    * @return this
    */
-  public final PipelineMapReduceDriver<K1, V1, K2, V2> withMapInputPath(Path mapInputPath) {
+  public final PipelineMapReduceDriver<K1, V1, K2, V2> withMapInputPath(
+      Path mapInputPath) {
     setMapInputPath(mapInputPath);
     return this;
   }
@@ -297,8 +300,7 @@
     }
     if (driverReused()) {
       throw new IllegalStateException("Driver reuse not allowed");
-    }
-    else {
+    } else {
       setUsedOnceStatus();
     }
 
@@ -331,7 +333,7 @@
   /**
    * Returns a new PipelineMapReduceDriver without having to specify the generic
    * types on the right hand side of the object create statement.
-   *
+   * 
    * @return new PipelineMapReduceDriver
    */
   public static <K1, V1, K2, V2> PipelineMapReduceDriver<K1, V1, K2, V2> newPipelineMapReduceDriver() {
@@ -341,7 +343,7 @@
   /**
    * Returns a new PipelineMapReduceDriver without having to specify the generic
    * types on the right hand side of the object create statement.
-   *
+   * 
    * @param pipeline
    *          passed to PipelineMapReduceDriver constructor
    * @return new PipelineMapReduceDriver
diff --git a/src/main/java/org/apache/hadoop/mrunit/ReduceDriverBase.java b/src/main/java/org/apache/hadoop/mrunit/ReduceDriverBase.java
index c66087f..e24b6dd 100644
--- a/src/main/java/org/apache/hadoop/mrunit/ReduceDriverBase.java
+++ b/src/main/java/org/apache/hadoop/mrunit/ReduceDriverBase.java
@@ -17,17 +17,17 @@
  */
 package org.apache.hadoop.mrunit;
 
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.mrunit.internal.io.Serialization;
 import org.apache.hadoop.mrunit.internal.output.MockOutputCreator;
 import org.apache.hadoop.mrunit.types.Pair;
 
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
 /**
  * Harness that allows you to test a Reducer instance. You provide a key and a
  * set of intermediate values for that key that represent inputs that should be
@@ -37,7 +37,7 @@
  * expected results.
  */
 public abstract class ReduceDriverBase<K1, V1, K2, V2, T extends ReduceDriverBase<K1, V1, K2, V2, T>>
-    extends TestDriver<K1, V1, K2, V2, T> {
+    extends TestDriver<K2, V2, T> {
 
   protected List<Pair<K1, List<V1>>> inputs = new ArrayList<Pair<K1, List<V1>>>();
   @Deprecated
@@ -53,7 +53,7 @@
 
   /**
    * Returns a list of values.
-   *
+   * 
    * @return List of values
    * @deprecated MRUNIT-64. Moved to list implementation to support multiple
    *             input (k, v*)*. Replaced by {@link #getInputValues(Object)}
@@ -65,7 +65,7 @@
 
   /**
    * Returns a list of values for the given key
-   *
+   * 
    * @param key
    * @return List for the given key, or null if key does not exist
    */
@@ -80,7 +80,7 @@
 
   /**
    * Sets the input key to send to the Reducer
-   *
+   * 
    * @deprecated MRUNIT-64. Moved to list implementation to support multiple
    *             input (k, v*)*. Replaced by {@link #setInput},
    *             {@link #addInput}, and {@link #addAll}
@@ -92,7 +92,7 @@
 
   /**
    * adds an input value to send to the reducer
-   *
+   * 
    * @param val
    * @deprecated MRUNIT-64. Moved to list implementation to support multiple
    *             input (k, v*)*. Replaced by {@link #setInput},
@@ -105,7 +105,7 @@
 
   /**
    * Sets the input values to send to the reducer; overwrites existing ones
-   *
+   * 
    * @param values
    * @deprecated MRUNIT-64. Moved to list implementation to support multiple
    *             input (k, v*)*. Replaced by {@link #setInput},
@@ -119,7 +119,7 @@
 
   /**
    * Adds a set of input values to send to the reducer
-   *
+   * 
    * @param values
    * @deprecated MRUNIT-64. Moved to list implementation to support multiple
    *             input (k, v*)*. Replaced by {@link #setInput},
@@ -134,7 +134,7 @@
 
   /**
    * Sets the input to send to the reducer
-   *
+   * 
    * @param key
    * @param values
    */
@@ -155,7 +155,7 @@
 
   /**
    * Add input (K, V*) to send to the Reducer
-   *
+   * 
    * @param key
    *          The key too add
    * @param values
@@ -173,7 +173,7 @@
 
   /**
    * Add input (K, V*) to send to the Reducer
-   *
+   * 
    * @param input
    *          input pair
    */
@@ -183,7 +183,7 @@
 
   /**
    * Adds input to send to the Reducer
-   *
+   * 
    * @param inputs
    *          list of (K, V*) pairs
    */
@@ -196,7 +196,7 @@
   /**
    * Expects an input of the form "key \t val, val, val..." Forces the Reducer
    * input types to Text.
-   *
+   * 
    * @param input
    *          A string of the form "key \t val,val,val". Trims any whitespace.
    * @deprecated No replacement due to lack of type safety and incompatibility
@@ -218,7 +218,7 @@
 
   /**
    * Identical to setInputKey() but with fluent programming style
-   *
+   * 
    * @return this
    * @deprecated MRUNIT-64. Moved to list implementation to support multiple
    *             input (k, v*)*. Replaced by {@link #withInput(Object, List)},
@@ -232,7 +232,7 @@
 
   /**
    * Identical to addInputValue() but with fluent programming style
-   *
+   * 
    * @param val
    * @return this
    * @deprecated MRUNIT-64. Moved to list implementation to support multiple
@@ -247,7 +247,7 @@
 
   /**
    * Identical to addInputValues() but with fluent programming style
-   *
+   * 
    * @param values
    * @return this
    * @deprecated MRUNIT-64. Moved to list implementation to support multiple
@@ -262,18 +262,17 @@
 
   /**
    * Identical to setInput() but returns self for fluent programming style
-   *
+   * 
    * @return this
    */
-  public T withInput(final K1 key,
-      final List<V1> values) {
+  public T withInput(final K1 key, final List<V1> values) {
     addInput(key, values);
     return thisAsReduceDriver();
   }
 
   /**
    * Identical to setInput, but with a fluent programming style
-   *
+   * 
    * @param input
    *          A string of the form "key \t val". Trims any whitespace.
    * @return this
@@ -288,7 +287,7 @@
 
   /**
    * Identical to addInput() but returns self for fluent programming style
-   *
+   * 
    * @param input
    * @return this
    */
@@ -299,12 +298,11 @@
 
   /**
    * Identical to addAll() but returns self for fluent programming style
-   *
+   * 
    * @param inputs
    * @return this
    */
-  public T withAll(
-      final List<Pair<K1, List<V1>>> inputs) {
+  public T withAll(final List<Pair<K1, List<V1>>> inputs) {
     addAll(inputs);
     return thisAsReduceDriver();
   }
@@ -327,8 +325,7 @@
     }
     if (driverReused()) {
       throw new IllegalStateException("Driver reuse not allowed");
-    }
-    else {
+    } else {
       setUsedOnceStatus();
     }
   }
diff --git a/src/main/java/org/apache/hadoop/mrunit/ReducePhaseRunner.java b/src/main/java/org/apache/hadoop/mrunit/ReducePhaseRunner.java
new file mode 100644
index 0000000..76da743
--- /dev/null
+++ b/src/main/java/org/apache/hadoop/mrunit/ReducePhaseRunner.java
@@ -0,0 +1,92 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mrunit;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapred.Counters;
+import org.apache.hadoop.mapred.InputFormat;
+import org.apache.hadoop.mapred.OutputFormat;
+import org.apache.hadoop.mapred.Reducer;
+import org.apache.hadoop.mrunit.types.Pair;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * The class to manage starting the reduce phase is used for type genericity
+ * reasons. This class is used in the run() method.
+ */
+class ReducePhaseRunner<INKEY, INVAL, OUTKEY, OUTVAL> {
+  public static final Log LOG = LogFactory.getLog(ReducePhaseRunner.class);
+
+  private final Configuration configuration;
+  private final Counters counters;
+  private Configuration outputSerializationConfiguration;
+  private Class<? extends OutputFormat> outputFormatClass;
+  private Class<? extends InputFormat> inputFormatClass;
+
+  ReducePhaseRunner(Class<? extends InputFormat> inputFormatClass,
+      Configuration configuration, Counters counters,
+      Configuration outputSerializationConfiguration,
+      Class<? extends OutputFormat> outputFormatClass) {
+    this.inputFormatClass = inputFormatClass;
+    this.configuration = configuration;
+    this.counters = counters;
+    this.outputSerializationConfiguration = outputSerializationConfiguration;
+    this.outputFormatClass = outputFormatClass;
+  }
+
+  public List<Pair<OUTKEY, OUTVAL>> runReduce(
+      final List<Pair<INKEY, List<INVAL>>> inputs,
+      final Reducer<INKEY, INVAL, OUTKEY, OUTVAL> reducer) throws IOException {
+
+    final List<Pair<OUTKEY, OUTVAL>> reduceOutputs = new ArrayList<Pair<OUTKEY, OUTVAL>>();
+
+    if (!inputs.isEmpty()) {
+      if (LOG.isDebugEnabled()) {
+        final StringBuilder sb = new StringBuilder();
+        for (Pair<INKEY, List<INVAL>> input : inputs) {
+          TestDriver.formatValueList(input.getSecond(), sb);
+          LOG.debug("Reducing input (" + input.getFirst() + ", " + sb + ")");
+          sb.delete(0, sb.length());
+        }
+      }
+
+      final ReduceDriver<INKEY, INVAL, OUTKEY, OUTVAL> reduceDriver = ReduceDriver
+          .newReduceDriver(reducer).withCounters(counters)
+          .withConfiguration(configuration).withAll(inputs);
+
+      if (outputSerializationConfiguration != null) {
+        reduceDriver
+            .withOutputSerializationConfiguration(outputSerializationConfiguration);
+      }
+
+      if (outputFormatClass != null) {
+        reduceDriver.withOutputFormat(outputFormatClass, inputFormatClass);
+      }
+
+      reduceOutputs.addAll(reduceDriver.run());
+    }
+
+    return reduceOutputs;
+  }
+}
\ No newline at end of file
diff --git a/src/main/java/org/apache/hadoop/mrunit/TestDriver.java b/src/main/java/org/apache/hadoop/mrunit/TestDriver.java
index be4d67e..a2d9536 100644
--- a/src/main/java/org/apache/hadoop/mrunit/TestDriver.java
+++ b/src/main/java/org/apache/hadoop/mrunit/TestDriver.java
@@ -17,14 +17,6 @@
  */
 package org.apache.hadoop.mrunit;
 
-import static org.apache.hadoop.mrunit.internal.util.ArgumentChecker.returnNonNull;
-
-import java.io.File;
-import java.io.IOException;
-import java.net.URI;
-import java.util.*;
-
-import com.google.common.collect.Lists;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
@@ -41,31 +33,38 @@
 import org.apache.hadoop.mrunit.internal.util.StringUtils;
 import org.apache.hadoop.mrunit.types.Pair;
 
-public abstract class TestDriver<K1, V1, K2, V2, T extends TestDriver<K1, V1, K2, V2, T>> {
+import java.io.File;
+import java.io.IOException;
+import java.net.URI;
+import java.util.*;
+
+import static org.apache.hadoop.mrunit.internal.util.ArgumentChecker.returnNonNull;
+
+public abstract class TestDriver<K, V, T extends TestDriver<K, V, T>> {
 
   public static final Log LOG = LogFactory.getLog(TestDriver.class);
 
-  protected List<Pair<K2, V2>> expectedOutputs;
+  protected List<Pair<K, V>> expectedOutputs;
 
   private boolean strictCountersChecking = false;
   protected List<Pair<Enum<?>, Long>> expectedEnumCounters;
   protected List<Pair<Pair<String, String>, Long>> expectedStringCounters;
   /**
-   * Configuration object, do not use directly, always use the
-   * the getter as it lazily creates the object in the case
-   * the setConfiguration() method will be used by the user.
+   * Configuration object, do not use directly, always use the the getter as it
+   * lazily creates the object in the case the setConfiguration() method will be
+   * used by the user.
    */
   private Configuration configuration;
   /**
-   * Serialization object, do not use directly, always use the
-   * the getter as it lazily creates the object in the case
-   * the setConfiguration() method will be used by the user.
+   * Serialization object, do not use directly, always use the the getter as it
+   * lazily creates the object in the case the setConfiguration() method will be
+   * used by the user.
    */
   private Serialization serialization;
 
   private Configuration outputSerializationConfiguration;
-  private Comparator<K2> keyComparator;
-  private Comparator<V2> valueComparator;
+  private Comparator<K> keyComparator;
+  private Comparator<V> valueComparator;
   private File tmpDistCacheDir;
   protected CounterWrapper counterWrapper;
   protected MockMultipleOutputs mos;
@@ -73,18 +72,18 @@
   protected Map<String, List<Pair<?, ?>>> expectedPathOutputs;
   private boolean hasRun = false;
 
-
   public TestDriver() {
-    expectedOutputs = new ArrayList<Pair<K2, V2>>();
+    expectedOutputs = new ArrayList<Pair<K, V>>();
     expectedEnumCounters = new ArrayList<Pair<Enum<?>, Long>>();
     expectedStringCounters = new ArrayList<Pair<Pair<String, String>, Long>>();
-    expectedMultipleOutputs = new HashMap<String, List<Pair<?, ? >>>();
-	expectedPathOutputs = new HashMap<String, List<Pair<?, ?>>>();
+    expectedMultipleOutputs = new HashMap<String, List<Pair<?, ?>>>();
+    expectedPathOutputs = new HashMap<String, List<Pair<?, ?>>>();
   }
 
   /**
    * Check to see if this driver is being reused
-   * @return  boolean - true if run() has been called more than once
+   * 
+   * @return boolean - true if run() has been called more than once
    */
   protected boolean driverReused() {
     return this.hasRun;
@@ -96,74 +95,76 @@
   protected void setUsedOnceStatus() {
     this.hasRun = true;
   }
+
   /**
    * Adds output (k, v)* pairs we expect
-   *
+   * 
    * @param outputRecords
    *          The (k, v)* pairs to add
    */
-  public void addAllOutput(final List<Pair<K2, V2>> outputRecords) {
-    for (Pair<K2, V2> output : outputRecords) {
+  public void addAllOutput(final List<Pair<K, V>> outputRecords) {
+    for (Pair<K, V> output : outputRecords) {
       addOutput(output);
     }
   }
 
   /**
    * Functions like addAllOutput() but returns self for fluent programming style
-   *
+   * 
    * @param outputRecords
    * @return this
    */
-  public T withAllOutput(
-      final List<Pair<K2, V2>> outputRecords) {
+  public T withAllOutput(final List<Pair<K, V>> outputRecords) {
     addAllOutput(outputRecords);
     return thisAsTestDriver();
   }
 
   /**
    * Adds an output (k, v) pair we expect
-   *
+   * 
    * @param outputRecord
    *          The (k, v) pair to add
    */
-  public void addOutput(final Pair<K2, V2> outputRecord) {
+  public void addOutput(final Pair<K, V> outputRecord) {
     addOutput(outputRecord.getFirst(), outputRecord.getSecond());
   }
 
   /**
    * Adds a (k, v) pair we expect as output
-   * @param key the key
-   * @param val the value
+   * 
+   * @param key
+   *          the key
+   * @param val
+   *          the value
    */
-  public void addOutput(final K2 key, final V2 val) {
+  public void addOutput(final K key, final V val) {
     expectedOutputs.add(copyPair(key, val));
   }
 
   /**
    * Works like addOutput(), but returns self for fluent style
-   *
+   * 
    * @param outputRecord
    * @return this
    */
-  public T withOutput(final Pair<K2, V2> outputRecord) {
+  public T withOutput(final Pair<K, V> outputRecord) {
     addOutput(outputRecord);
     return thisAsTestDriver();
   }
 
   /**
    * Works like addOutput() but returns self for fluent programming style
-   *
+   * 
    * @return this
    */
-  public T withOutput(final K2 key, final V2 val) {
+  public T withOutput(final K key, final V val) {
     addOutput(key, val);
     return thisAsTestDriver();
   }
 
   /**
-   * Expects an input of the form "key \t val" Forces the output types to
-   * Text.
-   *
+   * Expects an input of the form "key \t val" Forces the output types to Text.
+   * 
    * @param output
    *          A string of the form "key \t val". Trims any whitespace.
    * @deprecated No replacement due to lack of type safety and incompatibility
@@ -172,12 +173,12 @@
   @Deprecated
   @SuppressWarnings("unchecked")
   public void addOutputFromString(final String output) {
-    addOutput((Pair<K2, V2>) parseTabbedPair(output));
+    addOutput((Pair<K, V>) parseTabbedPair(output));
   }
 
   /**
    * Identical to addOutputFromString, but with a fluent programming style
-   *
+   * 
    * @param output
    *          A string of the form "key \t val". Trims any whitespace.
    * @return this
@@ -193,7 +194,7 @@
   /**
    * @return the list of (k, v) pairs expected as output from this driver
    */
-  public List<Pair<K2, V2>> getExpectedOutputs() {
+  public List<Pair<K, V>> getExpectedOutputs() {
     return expectedOutputs;
   }
 
@@ -233,22 +234,21 @@
 
   /**
    * Register expected enumeration based counter value
-   *
+   * 
    * @param e
    *          Enumeration based counter
    * @param expectedValue
    *          Expected value
    * @return this
    */
-  public T withCounter(final Enum<?> e,
-      final long expectedValue) {
+  public T withCounter(final Enum<?> e, final long expectedValue) {
     expectedEnumCounters.add(new Pair<Enum<?>, Long>(e, expectedValue));
     return thisAsTestDriver();
   }
 
   /**
    * Register expected name based counter value
-   *
+   * 
    * @param group
    *          Counter group
    * @param name
@@ -257,8 +257,8 @@
    *          Expected value
    * @return this
    */
-  public T withCounter(final String group,
-      final String name, final long expectedValue) {
+  public T withCounter(final String group, final String name,
+      final long expectedValue) {
     expectedStringCounters.add(new Pair<Pair<String, String>, Long>(
         new Pair<String, String>(group, name), expectedValue));
     return thisAsTestDriver();
@@ -268,7 +268,7 @@
    * Change counter checking. After this method is called, the test will fail if
    * an actual counter is not matched by an expected counter. By default, the
    * test only check that every expected counter is there.
-   *
+   * 
    * This mode allows you to ensure that no unexpected counters has been
    * declared.
    */
@@ -282,7 +282,7 @@
    *         reducer associated with the driver
    */
   public Configuration getConfiguration() {
-    if(configuration == null) {
+    if (configuration == null) {
       configuration = new Configuration();
     }
     return configuration;
@@ -291,14 +291,14 @@
   /**
    * @return The comparator for output keys or null of none has been set
    */
-  public Comparator<K2> getKeyComparator() {
+  public Comparator<K> getKeyComparator() {
     return this.keyComparator;
   }
 
   /**
    * @return The comparator for output values or null of none has been set
    */
-  public Comparator<V2> getValueComparator() {
+  public Comparator<V> getValueComparator() {
     return this.valueComparator;
   }
 
@@ -306,11 +306,11 @@
    * @param configuration
    *          The configuration object that will given to the mapper and/or
    *          reducer associated with the driver. This method should only be
-   *          called directly after the constructor as the internal state
-   *          of the driver depends on the configuration object
-   * @deprecated
-   *          Use getConfiguration() to set configuration items as opposed to
-   *          overriding the entire configuration object as it's used internally.
+   *          called directly after the constructor as the internal state of the
+   *          driver depends on the configuration object
+   * @deprecated Use getConfiguration() to set configuration items as opposed to
+   *             overriding the entire configuration object as it's used
+   *             internally.
    */
   @Deprecated
   public void setConfiguration(final Configuration configuration) {
@@ -323,14 +323,13 @@
    *          with the driver. This method should only be called directly after
    *          the constructor as the internal state of the driver depends on the
    *          configuration object
-   * @deprecated
-   *          Use getConfiguration() to set configuration items as opposed to
-   *          overriding the entire configuration object as it's used internally.
+   * @deprecated Use getConfiguration() to set configuration items as opposed to
+   *             overriding the entire configuration object as it's used
+   *             internally.
    * @return this object for fluent coding
    */
   @Deprecated
-  public T withConfiguration(
-      final Configuration configuration) {
+  public T withConfiguration(final Configuration configuration) {
     setConfiguration(configuration);
     return thisAsTestDriver();
   }
@@ -339,7 +338,7 @@
    * Get the {@link Configuration} to use when copying output for use with run*
    * methods or for the InputFormat when reading output back in when setting a
    * real OutputFormat.
-   *
+   * 
    * @return outputSerializationConfiguration, null when no
    *         outputSerializationConfiguration is set
    */
@@ -353,7 +352,7 @@
    * real OutputFormat. When this configuration is not set, MRUnit will use the
    * configuration set with {@link #withConfiguration(Configuration)} or
    * {@link #setConfiguration(Configuration)}
-   *
+   * 
    * @param configuration
    */
   public void setOutputSerializationConfiguration(
@@ -367,22 +366,21 @@
    * real OutputFormat. When this configuration is not set, MRUnit will use the
    * configuration set with {@link #withConfiguration(Configuration)} or
    * {@link #setConfiguration(Configuration)}
-   *
+   * 
    * @param configuration
    * @return this for fluent style
    */
-  public T withOutputSerializationConfiguration(
-      Configuration configuration) {
+  public T withOutputSerializationConfiguration(Configuration configuration) {
     setOutputSerializationConfiguration(configuration);
     return thisAsTestDriver();
   }
 
   /**
-   * Adds a file to be put on the distributed cache.
-   * The path may be relative and will try to be resolved from
-   * the classpath of the test.
-   *
-   * @param path path to the file
+   * Adds a file to be put on the distributed cache. The path may be relative
+   * and will try to be resolved from the classpath of the test.
+   * 
+   * @param path
+   *          path to the file
    */
   public void addCacheFile(String path) {
     addCacheFile(DistCacheUtils.findResource(path));
@@ -390,7 +388,9 @@
 
   /**
    * Adds a file to be put on the distributed cache.
-   * @param uri uri of the file
+   * 
+   * @param uri
+   *          uri of the file
    */
   public void addCacheFile(URI uri) {
     DistributedCache.addCacheFile(uri, getConfiguration());
@@ -398,7 +398,9 @@
 
   /**
    * Set the list of files to put on the distributed cache
-   * @param files list of URIs
+   * 
+   * @param files
+   *          list of URIs
    */
   public void setCacheFiles(URI[] files) {
     DistributedCache.setCacheFiles(files, getConfiguration());
@@ -406,26 +408,30 @@
 
   /**
    * Set the output key comparator
-   * @param keyComparator the key comparator
+   * 
+   * @param keyComparator
+   *          the key comparator
    */
-  public void setKeyComparator(Comparator<K2> keyComparator) {
+  public void setKeyComparator(Comparator<K> keyComparator) {
     this.keyComparator = keyComparator;
   }
 
   /**
    * Set the output value comparator
-   * @param valueComparator the value comparator
+   * 
+   * @param valueComparator
+   *          the value comparator
    */
-  public void setValueComparator(Comparator<V2> valueComparator) {
+  public void setValueComparator(Comparator<V> valueComparator) {
     this.valueComparator = valueComparator;
   }
 
   /**
-   * Adds an archive to be put on the distributed cache.
-   * The path may be relative and will try to be resolved from
-   * the classpath of the test.
-   *
-   * @param path path to the archive
+   * Adds an archive to be put on the distributed cache. The path may be
+   * relative and will try to be resolved from the classpath of the test.
+   * 
+   * @param path
+   *          path to the archive
    */
   public void addCacheArchive(String path) {
     addCacheArchive(DistCacheUtils.findResource(path));
@@ -433,7 +439,9 @@
 
   /**
    * Adds an archive to be put on the distributed cache.
-   * @param uri uri of the archive
+   * 
+   * @param uri
+   *          uri of the archive
    */
   public void addCacheArchive(URI uri) {
     DistributedCache.addCacheArchive(uri, getConfiguration());
@@ -441,18 +449,20 @@
 
   /**
    * Set the list of archives to put on the distributed cache
-   * @param archives list of URIs
+   * 
+   * @param archives
+   *          list of URIs
    */
   public void setCacheArchives(URI[] archives) {
     DistributedCache.setCacheArchives(archives, getConfiguration());
   }
 
   /**
-   * Adds a file to be put on the distributed cache.
-   * The path may be relative and will try to be resolved from
-   * the classpath of the test.
-   *
-   * @param file path to the file
+   * Adds a file to be put on the distributed cache. The path may be relative
+   * and will try to be resolved from the classpath of the test.
+   * 
+   * @param file
+   *          path to the file
    * @return the driver
    */
   public T withCacheFile(String file) {
@@ -462,7 +472,9 @@
 
   /**
    * Adds a file to be put on the distributed cache.
-   * @param file uri of the file
+   * 
+   * @param file
+   *          uri of the file
    * @return the driver
    */
   public T withCacheFile(URI file) {
@@ -471,11 +483,11 @@
   }
 
   /**
-   * Adds an archive to be put on the distributed cache.
-   * The path may be relative and will try to be resolved from
-   * the classpath of the test.
-   *
-   * @param archive path to the archive
+   * Adds an archive to be put on the distributed cache. The path may be
+   * relative and will try to be resolved from the classpath of the test.
+   * 
+   * @param archive
+   *          path to the archive
    * @return the driver
    */
   public T withCacheArchive(String archive) {
@@ -485,7 +497,9 @@
 
   /**
    * Adds an archive to be put on the distributed cache.
-   * @param archive uri of the archive
+   * 
+   * @param archive
+   *          uri of the archive
    * @return the driver
    */
   public T withCacheArchive(URI archive) {
@@ -496,14 +510,15 @@
   /**
    * Runs the test but returns the result set instead of validating it (ignores
    * any addOutput(), etc calls made before this).
-   *
+   * 
    * Also optionally performs counter validation.
-   *
-   * @param validateCounters whether to run automatic counter validation
+   * 
+   * @param validateCounters
+   *          whether to run automatic counter validation
    * @return the list of (k, v) pairs returned as output from the test
    */
-  public List<Pair<K2, V2>> run(boolean validateCounters) throws IOException {
-    final List<Pair<K2, V2>> outputs = run();
+  public List<Pair<K, V>> run(boolean validateCounters) throws IOException {
+    final List<Pair<K, V>> outputs = run();
     if (validateCounters) {
       validate(counterWrapper);
     }
@@ -511,20 +526,20 @@
   }
 
   private Serialization getSerialization() {
-    if(serialization == null) {
+    if (serialization == null) {
       serialization = new Serialization(getConfiguration());
     }
     return serialization;
   }
 
   /**
-   * Initialises the test distributed cache if required. This
-   * process is referred to as "localizing" by Hadoop, but since
-   * this is a unit test all files/archives are already local.
-   *
-   * Cached files are not moved but cached archives are extracted
-   * into a temporary directory.
-   *
+   * Initialises the test distributed cache if required. This process is
+   * referred to as "localizing" by Hadoop, but since this is a unit test all
+   * files/archives are already local.
+   * 
+   * Cached files are not moved but cached archives are extracted into a
+   * temporary directory.
+   * 
    * @throws IOException
    */
   protected void initDistributedCache() throws IOException {
@@ -539,7 +554,7 @@
     List<Path> localFiles = new ArrayList<Path>();
 
     if (DistributedCache.getCacheFiles(conf) != null) {
-      for (URI uri: DistributedCache.getCacheFiles(conf)) {
+      for (URI uri : DistributedCache.getCacheFiles(conf)) {
         Path filePath = new Path(uri.getPath());
         localFiles.add(filePath);
       }
@@ -549,13 +564,13 @@
       }
     }
     if (DistributedCache.getCacheArchives(conf) != null) {
-      for (URI uri: DistributedCache.getCacheArchives(conf)) {
+      for (URI uri : DistributedCache.getCacheArchives(conf)) {
         Path archivePath = new Path(uri.getPath());
         if (tmpDistCacheDir == null) {
           tmpDistCacheDir = DistCacheUtils.createTempDirectory();
         }
-        localArchives.add(DistCacheUtils.extractArchiveToTemp(
-            archivePath, tmpDistCacheDir));
+        localArchives.add(DistCacheUtils.extractArchiveToTemp(archivePath,
+            tmpDistCacheDir));
       }
       if (!localArchives.isEmpty()) {
         DistCacheUtils.addLocalArchives(conf,
@@ -565,28 +580,28 @@
   }
 
   /**
-   * Checks whether the distributed cache has been "localized", i.e.
-   * archives extracted and paths moved so that they can be accessed
-   * through {@link DistributedCache#getLocalCacheArchives()} and
+   * Checks whether the distributed cache has been "localized", i.e. archives
+   * extracted and paths moved so that they can be accessed through
+   * {@link DistributedCache#getLocalCacheArchives()} and
    * {@link DistributedCache#getLocalCacheFiles()}
-   *
-   * @param conf the configuration
+   * 
+   * @param conf
+   *          the configuration
    * @return true if the cache is initialised
    * @throws IOException
    */
   private boolean isDistributedCacheInitialised(Configuration conf)
       throws IOException {
-    return DistributedCache.getLocalCacheArchives(conf) != null ||
-        DistributedCache.getLocalCacheFiles(conf) != null;
+    return DistributedCache.getLocalCacheArchives(conf) != null
+        || DistributedCache.getLocalCacheFiles(conf) != null;
   }
 
   /**
-   * Cleans up the distributed cache test by deleting the
-   * temporary directory and any extracted cache archives
-   * contained within
-   *
+   * Cleans up the distributed cache test by deleting the temporary directory
+   * and any extracted cache archives contained within
+   * 
    * @throws IOException
-   *  if the local fs handle cannot be retrieved
+   *           if the local fs handle cannot be retrieved
    */
   protected void cleanupDistributedCache() throws IOException {
     if (tmpDistCacheDir != null) {
@@ -600,10 +615,10 @@
   /**
    * Runs the test but returns the result set instead of validating it (ignores
    * any addOutput(), etc calls made before this)
-   *
+   * 
    * @return the list of (k, v) pairs returned as output from the test
    */
-  public abstract List<Pair<K2, V2>> run() throws IOException;
+  public abstract List<Pair<K, V>> run() throws IOException;
 
   /**
    * Runs the test and validates the results
@@ -614,7 +629,7 @@
 
   /**
    * Runs the test and validates the results
-   *
+   * 
    * @param orderMatters
    *          Whether or not output ordering is important
    */
@@ -622,7 +637,7 @@
     if (LOG.isDebugEnabled()) {
       printPreTestDebugLog();
     }
-    final List<Pair<K2, V2>> outputs = run();
+    final List<Pair<K, V>> outputs = run();
     validate(outputs, orderMatters);
     validate(counterWrapper);
     validate(mos);
@@ -637,7 +652,7 @@
 
   /**
    * Split "key \t val" into Pair(Text(key), Text(val))
-   *
+   * 
    * @param tabSeparatedPair
    * @return (k,v)
    */
@@ -647,7 +662,7 @@
 
   /**
    * Split "val,val,val,val..." into a List of Text(val) objects.
-   *
+   * 
    * @param commaDelimList
    *          A list of values separated by commas
    */
@@ -666,52 +681,55 @@
 
   /**
    * check the outputs against the expected inputs in record
-   *
+   * 
    * @param outputs
    *          The actual output (k, v) pairs
    * @param orderMatters
    *          Whether or not output ordering is important when validating test
    *          result
    */
-  protected void validate(final List<Pair<K2, V2>> outputs,
+  protected void validate(final List<Pair<K, V>> outputs,
       final boolean orderMatters) {
     // expected nothing and got nothing, everything is fine
     if (outputs.isEmpty() && expectedOutputs.isEmpty()) {
-        return;
+      return;
     }
 
     final Errors errors = new Errors(LOG);
     // expected nothing but got something
     if (!outputs.isEmpty() && expectedOutputs.isEmpty()) {
-        errors.record("Expected no output; got %d output(s).", outputs.size());
-        errors.assertNone();
+      errors.record("Expected no output; got %d output(s).", outputs.size());
+      errors.assertNone();
     }
     // expected something but got nothing
     if (outputs.isEmpty() && !expectedOutputs.isEmpty()) {
-        errors.record("Expected %d output(s); got no output.", expectedOutputs.size());
-        errors.assertNone();
+      errors.record("Expected %d output(s); got no output.",
+          expectedOutputs.size());
+      errors.assertNone();
     }
 
     // now, the smart test needs to be done
-    // check that user's key and value writables implement equals, hashCode, toString
+    // check that user's key and value writables implement equals, hashCode,
+    // toString
     checkOverrides(outputs, expectedOutputs);
 
-    final PairEquality<K2, V2> equality = new PairEquality<K2, V2>(
-            keyComparator, valueComparator);
+    final PairEquality<K, V> equality = new PairEquality<K, V>(keyComparator,
+        valueComparator);
     if (orderMatters) {
-        validateWithOrder(outputs, errors, equality);
+      validateWithOrder(outputs, errors, equality);
     } else {
-        validateWithoutOrder(outputs, errors, equality);
+      validateWithoutOrder(outputs, errors, equality);
     }
 
-    // if there are errors, it might be due to types and not clear from the message
-    if(!errors.isEmpty()) {
+    // if there are errors, it might be due to types and not clear from the
+    // message
+    if (!errors.isEmpty()) {
       Class<?> outputKeyClass = null;
       Class<?> outputValueClass = null;
       Class<?> expectedKeyClass = null;
       Class<?> expectedValueClass = null;
 
-      for (Pair<K2, V2> output : outputs) {
+      for (Pair<K, V> output : outputs) {
         if (output.getFirst() != null) {
           outputKeyClass = output.getFirst().getClass();
         }
@@ -723,7 +741,7 @@
         }
       }
 
-      for (Pair<K2, V2> expected : expectedOutputs) {
+      for (Pair<K, V> expected : expectedOutputs) {
         if (expected.getFirst() != null) {
           expectedKeyClass = expected.getFirst().getClass();
         }
@@ -735,13 +753,13 @@
         }
       }
 
-      if (outputKeyClass != null && expectedKeyClass !=null
+      if (outputKeyClass != null && expectedKeyClass != null
           && !outputKeyClass.equals(expectedKeyClass)) {
         errors.record("Mismatch in key class: expected: %s actual: %s",
             expectedKeyClass, outputKeyClass);
       }
 
-      if (outputValueClass != null && expectedValueClass !=null
+      if (outputValueClass != null && expectedValueClass != null
           && !outputValueClass.equals(expectedValueClass)) {
         errors.record("Mismatch in value class: expected: %s actual: %s",
             expectedValueClass, outputValueClass);
@@ -750,85 +768,86 @@
     errors.assertNone();
   }
 
-  private void validateWithoutOrder(final List<Pair<K2, V2>> outputs,
-      final Errors errors, final PairEquality<K2, V2> equality) {
+  private void validateWithoutOrder(final List<Pair<K, V>> outputs,
+      final Errors errors, final PairEquality<K, V> equality) {
     Set<Integer> verifiedExpecteds = new HashSet<Integer>();
     Set<Integer> unverifiedOutputs = new HashSet<Integer>();
     for (int i = 0; i < outputs.size(); i++) {
-        Pair<K2, V2> output = outputs.get(i);
-        boolean found = false;
-        for (int j = 0; j < expectedOutputs.size(); j++) {
-            if (verifiedExpecteds.contains(j)) {
-                continue;
-            }
-            Pair<K2, V2> expected = expectedOutputs.get(j);
-            if (equality.isTrueFor(output, expected)) {
-                found = true;
-                verifiedExpecteds.add(j);
-                LOG.debug(String.format("Matched expected output %s no %d at "
-                        + "position %d", output, j, i));
-                break;
-            }
+      Pair<K, V> output = outputs.get(i);
+      boolean found = false;
+      for (int j = 0; j < expectedOutputs.size(); j++) {
+        if (verifiedExpecteds.contains(j)) {
+          continue;
         }
-        if (!found) {
-            unverifiedOutputs.add(i);
+        Pair<K, V> expected = expectedOutputs.get(j);
+        if (equality.isTrueFor(output, expected)) {
+          found = true;
+          verifiedExpecteds.add(j);
+          LOG.debug(String.format("Matched expected output %s no %d at "
+              + "position %d", output, j, i));
+          break;
         }
+      }
+      if (!found) {
+        unverifiedOutputs.add(i);
+      }
     }
     for (int j = 0; j < expectedOutputs.size(); j++) {
-        if (!verifiedExpecteds.contains(j)) {
-            errors.record("Missing expected output %s", expectedOutputs.get(j));
-        }
+      if (!verifiedExpecteds.contains(j)) {
+        errors.record("Missing expected output %s", expectedOutputs.get(j));
+      }
     }
     for (int i = 0; i < outputs.size(); i++) {
-        if (unverifiedOutputs.contains(i)) {
-            errors.record("Received unexpected output %s", outputs.get(i));
-        }
+      if (unverifiedOutputs.contains(i)) {
+        errors.record("Received unexpected output %s", outputs.get(i));
+      }
     }
   }
 
-  private void validateWithOrder(final List<Pair<K2, V2>> outputs,
-      final Errors errors, final PairEquality<K2, V2> equality) {
+  private void validateWithOrder(final List<Pair<K, V>> outputs,
+      final Errors errors, final PairEquality<K, V> equality) {
     int i = 0;
-    for (i = 0; i < Math.min(outputs.size(),expectedOutputs.size()); i++) {
-        Pair<K2, V2> output = outputs.get(i);
-        Pair<K2, V2> expected = expectedOutputs.get(i);
-        if (equality.isTrueFor(output, expected)) {
-            LOG.debug(String.format("Matched expected output %s at "
-                    + "position %d", expected, i));
-        } else {
-            errors.record("Missing expected output %s at position %d, got %s.",
-                    expected, i, output);
-        }
+    for (i = 0; i < Math.min(outputs.size(), expectedOutputs.size()); i++) {
+      Pair<K, V> output = outputs.get(i);
+      Pair<K, V> expected = expectedOutputs.get(i);
+      if (equality.isTrueFor(output, expected)) {
+        LOG.debug(String.format("Matched expected output %s at "
+            + "position %d", expected, i));
+      } else {
+        errors.record("Missing expected output %s at position %d, got %s.",
+            expected, i, output);
+      }
     }
-    for(int j=i; j < outputs.size(); j++) {
-        errors.record("Received unexpected output %s at position %d.",
-                outputs.get(j), j);
+    for (int j = i; j < outputs.size(); j++) {
+      errors.record("Received unexpected output %s at position %d.",
+          outputs.get(j), j);
     }
-    for(int j=i; j < expectedOutputs.size(); j++) {
-        errors.record("Missing expected output %s at position %d.",
-                expectedOutputs.get(j), j);
+    for (int j = i; j < expectedOutputs.size(); j++) {
+      errors.record("Missing expected output %s at position %d.",
+          expectedOutputs.get(j), j);
     }
   }
 
-  private void checkOverrides(final List<Pair<K2,V2>> outputPairs, final List<Pair<K2,V2>> expectedOutputPairs) {
+  private void checkOverrides(final List<Pair<K, V>> outputPairs,
+      final List<Pair<K, V>> expectedOutputPairs) {
     Class<?> keyClass = null;
     Class<?> valueClass = null;
     // key or value could be null, try to find a class
-    for (Pair<K2,V2> pair : outputPairs) {
-        if (keyClass == null && pair.getFirst() != null) {
-            keyClass = pair.getFirst().getClass();
-        }
-        if (valueClass == null && pair.getSecond() != null) {
-        	valueClass = pair.getSecond().getClass();
-        }
+    for (Pair<K, V> pair : outputPairs) {
+      if (keyClass == null && pair.getFirst() != null) {
+        keyClass = pair.getFirst().getClass();
+      }
+      if (valueClass == null && pair.getSecond() != null) {
+        valueClass = pair.getSecond().getClass();
+      }
     }
-    for (Pair<K2,V2> pair : expectedOutputPairs) {
-        if (keyClass == null && pair.getFirst() != null) {
-            keyClass = pair.getFirst().getClass();
-        }
-        if (valueClass == null && pair.getSecond() != null) {
-        	valueClass = pair.getSecond().getClass();
-        }
+    for (Pair<K, V> pair : expectedOutputPairs) {
+      if (keyClass == null && pair.getFirst() != null) {
+        keyClass = pair.getFirst().getClass();
+      }
+      if (valueClass == null && pair.getSecond() != null) {
+        valueClass = pair.getSecond().getClass();
+      }
     }
     checkOverride(keyClass);
     checkOverride(valueClass);
@@ -836,20 +855,21 @@
 
   private void checkOverride(final Class<?> clazz) {
     if (clazz == null) {
-        return;
+      return;
     }
     try {
       if (clazz.getMethod("equals", Object.class).getDeclaringClass() != clazz) {
-        LOG.warn(clazz.getCanonicalName() + ".equals(Object) " +
-            "is not being overridden - tests may fail!");
+        LOG.warn(clazz.getCanonicalName() + ".equals(Object) "
+            + "is not being overridden - tests may fail!");
       }
       if (clazz.getMethod("hashCode").getDeclaringClass() != clazz) {
-        LOG.warn(clazz.getCanonicalName() + ".hashCode() " +
-            "is not being overridden - tests may fail!");
+        LOG.warn(clazz.getCanonicalName() + ".hashCode() "
+            + "is not being overridden - tests may fail!");
       }
       if (clazz.getMethod("toString").getDeclaringClass() != clazz) {
-        LOG.warn(clazz.getCanonicalName() + ".toString() " +
-            "is not being overridden - test failures may be difficult to diagnose.");
+        LOG.warn(clazz.getCanonicalName()
+            + ".toString() "
+            + "is not being overridden - test failures may be difficult to diagnose.");
         LOG.warn("Consider executing test using run() to access outputs");
       }
     } catch (SecurityException e) {
@@ -859,12 +879,12 @@
     }
   }
 
-  private Map<Pair<K2, V2>, List<Integer>> buildPositionMap(
-      final List<Pair<K2, V2>> values, Comparator<Pair<K2, V2>> comparator) {
-    final Map<Pair<K2, V2>, List<Integer>> valuePositions =
-        new TreeMap<Pair<K2, V2>, List<Integer>>(comparator);
+  private Map<Pair<K, V>, List<Integer>> buildPositionMap(
+      final List<Pair<K, V>> values, Comparator<Pair<K, V>> comparator) {
+    final Map<Pair<K, V>, List<Integer>> valuePositions = new TreeMap<Pair<K, V>, List<Integer>>(
+        comparator);
     for (int i = 0; i < values.size(); i++) {
-      final Pair<K2, V2> output = values.get(i);
+      final Pair<K, V> output = values.get(i);
       List<Integer> positions;
       if (valuePositions.containsKey(output)) {
         positions = valuePositions.get(output);
@@ -877,7 +897,6 @@
     return valuePositions;
   }
 
-
   /**
    * Check counters.
    */
@@ -885,6 +904,7 @@
     validateExpectedAgainstActual(counterWrapper);
     validateActualAgainstExpected(counterWrapper);
   }
+
   /**
    * Check Multiple Outputs.
    */
@@ -1025,10 +1045,10 @@
    */
   private Collection<Pair<String, String>> findExpectedCounterValues() {
     Collection<Pair<String, String>> results = new ArrayList<Pair<String, String>>();
-    for (Pair<Pair<String, String>,Long> counterAndCount : expectedStringCounters) {
+    for (Pair<Pair<String, String>, Long> counterAndCount : expectedStringCounters) {
       results.add(counterAndCount.getFirst());
     }
-    for (Pair<Enum<?>,Long> counterAndCount : expectedEnumCounters) {
+    for (Pair<Enum<?>, Long> counterAndCount : expectedEnumCounters) {
       Enum<?> first = counterAndCount.getFirst();
       String groupName = first.getDeclaringClass().getName();
       String counterName = first.name();
@@ -1038,13 +1058,12 @@
   }
 
   /**
-   * Check that provided actual counters contain all expected counters with proper
-   * values.
-   *
+   * Check that provided actual counters contain all expected counters with
+   * proper values.
+   * 
    * @param counterWrapper
    */
-  private void validateExpectedAgainstActual(
-      final CounterWrapper counterWrapper) {
+  private void validateExpectedAgainstActual(final CounterWrapper counterWrapper) {
     final Errors errors = new Errors(LOG);
 
     // Firstly check enumeration based counters
@@ -1080,16 +1099,17 @@
 
   /**
    * Check that provided actual counters are all expected.
-   *
+   * 
    * @param counterWrapper
    */
   private void validateActualAgainstExpected(final CounterWrapper counterWrapper) {
     if (strictCountersChecking) {
       final Errors errors = new Errors(LOG);
-      Collection<Pair<String, String>> unmatchedCounters = counterWrapper.findCounterValues();
+      Collection<Pair<String, String>> unmatchedCounters = counterWrapper
+          .findCounterValues();
       Collection<Pair<String, String>> findExpectedCounterValues = findExpectedCounterValues();
       unmatchedCounters.removeAll(findExpectedCounterValues);
-      if(!unmatchedCounters.isEmpty()) {
+      if (!unmatchedCounters.isEmpty()) {
         for (Pair<String, String> unmatcherCounter : unmatchedCounters) {
           errors
               .record(
@@ -1101,34 +1121,37 @@
     }
   }
 
-  protected static void formatValueList(final List<?> values,
+  public static void formatValueList(final List<?> values,
       final StringBuilder sb) {
     StringUtils.formatValueList(values, sb);
   }
 
-  protected static <KEYIN, VALUEIN> void formatPairList(final List<Pair<KEYIN,VALUEIN>> pairs,
-      final StringBuilder sb) {
+  protected static <KEYIN, VALUEIN> void formatPairList(
+      final List<Pair<KEYIN, VALUEIN>> pairs, final StringBuilder sb) {
     StringUtils.formatPairList(pairs, sb);
   }
 
   /**
    * Adds an output (k, v) pair we expect as Multiple output
-   *
+   * 
    * @param namedOutput
    * @param outputRecord
    */
-  public <K, V> void addMultiOutput(String namedOutput, final Pair<K, V> outputRecord) {
-    addMultiOutput(namedOutput, outputRecord.getFirst(), outputRecord.getSecond());
+  public <K, V> void addMultiOutput(String namedOutput,
+      final Pair<K, V> outputRecord) {
+    addMultiOutput(namedOutput, outputRecord.getFirst(),
+        outputRecord.getSecond());
   }
 
   /**
    * add a (k, v) pair we expect as Multiple output
-   *
+   * 
    * @param namedOutput
    * @param key
    * @param val
    */
-  public <K, V> void addMultiOutput(final String namedOutput, final K key, final V val) {
+  public <K, V> void addMultiOutput(final String namedOutput, final K key,
+      final V val) {
     List<Pair<?, ?>> outputs = expectedMultipleOutputs.get(namedOutput);
     if (outputs == null) {
       outputs = new ArrayList<Pair<?, ?>>();
@@ -1139,20 +1162,21 @@
 
   /**
    * works like addMultiOutput() but returns self for fluent programming style
-   *
+   * 
    * @param namedOutput
    * @param key
    * @param value
    * @return this
    */
-  public <K extends Comparable, V extends Comparable> T withMultiOutput(final String namedOutput, final K key, final V value) {
+  public <K extends Comparable, V extends Comparable> T withMultiOutput(
+      final String namedOutput, final K key, final V value) {
     addMultiOutput(namedOutput, key, value);
     return thisAsTestDriver();
   }
 
   /**
    * Works like addMultiOutput(), but returns self for fluent programming style
-   *
+   * 
    * @param namedOutput
    * @param outputRecord
    * @return this
diff --git a/src/main/java/org/apache/hadoop/mrunit/internal/driver/MultipleInputsMapReduceDriverBase.java b/src/main/java/org/apache/hadoop/mrunit/internal/driver/MultipleInputsMapReduceDriverBase.java
new file mode 100644
index 0000000..726aaeb
--- /dev/null
+++ b/src/main/java/org/apache/hadoop/mrunit/internal/driver/MultipleInputsMapReduceDriverBase.java
@@ -0,0 +1,307 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mrunit.internal.driver;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.RawComparator;
+import org.apache.hadoop.mrunit.TestDriver;
+import org.apache.hadoop.mrunit.types.Pair;
+import org.apache.hadoop.util.ReflectionUtils;
+
+import java.util.*;
+
+/**
+ * Harness that allows you to test multiple Mappers and a Reducer instance
+ * together You provide the input keys and values that should be sent to each
+ * Mapper, and outputs you expect to be sent by the Reducer to the collector for
+ * those inputs. By calling runTest(), the harness will deliver the inputs to
+ * the respective Mappers, feed the intermediate results to the Reducer (without
+ * checking them), and will check the Reducer's outputs against the expected
+ * results.
+ * 
+ * @param <M>
+ *          The type of the Mapper (to support mapred and mapreduce API)
+ * @param <K1>
+ *          The common map output key type
+ * @param <V1>
+ *          The common map output value type
+ * @param <K2>
+ *          The reduce output key type
+ * @param <V2>
+ *          The reduce output value type
+ * @param <T>
+ *          The type of the MultipleInputMapReduceDriver implementation
+ */
+public abstract class MultipleInputsMapReduceDriverBase<M, K1, V1, K2, V2, T extends MultipleInputsMapReduceDriverBase<M, K1, V1, K2, V2, T>>
+    extends TestDriver<K2, V2, T> {
+  public static final Log LOG = LogFactory
+      .getLog(MultipleInputsMapReduceDriverBase.class);
+
+  protected Map<M, Path> mapInputPaths = new HashMap<M, Path>();
+
+  /**
+   * The path passed to the specifed mapper InputSplit
+   * 
+   * @param mapper
+   *          The mapper to get the input path for
+   * @return The path
+   */
+  public Path getMapInputPath(final M mapper) {
+    return mapInputPaths.get(mapper);
+  }
+
+  /**
+   * Path that is passed to the mapper InputSplit
+   * 
+   * @param mapper
+   *          The mapper to set the input path for
+   * @param mapInputPath
+   *          The path
+   */
+  public void setMapInputPath(final M mapper, Path mapInputPath) {
+    mapInputPaths.put(mapper, mapInputPath);
+  }
+
+  /**
+   * Identical to setMapInputPath but supports a fluent programming style
+   * 
+   * @param mapper
+   *          The mapper to set the input path for
+   * @param mapInputPath
+   *          The path
+   * @return this
+   */
+  public final T withMapInputPath(final M mapper, Path mapInputPath) {
+    this.setMapInputPath(mapper, mapInputPath);
+    return thisAsMapReduceDriver();
+  }
+
+  /**
+   * Key group comparator
+   */
+  protected Comparator<K1> keyGroupComparator;
+
+  /**
+   * Set the key grouping comparator, similar to calling the following API calls
+   * but passing a real instance rather than just the class:
+   * <UL>
+   * <LI>pre 0.20.1 API:
+   * {@link org.apache.hadoop.mapred.JobConf#setOutputValueGroupingComparator(Class)}
+   * <LI>0.20.1+ API:
+   * {@link org.apache.hadoop.mapreduce.Job#setGroupingComparatorClass(Class)}
+   * </UL>
+   * 
+   * @param groupingComparator
+   */
+  @SuppressWarnings("unchecked")
+  public void setKeyGroupingComparator(
+      final RawComparator<K2> groupingComparator) {
+    keyGroupComparator = ReflectionUtils.newInstance(
+        groupingComparator.getClass(), getConfiguration());
+  }
+
+  /**
+   * Identical to {@link #setKeyGroupingComparator(RawComparator)}, but with a
+   * fluent programming style
+   * 
+   * @param groupingComparator
+   *          Comparator to use in the shuffle stage for key grouping
+   * @return this
+   */
+  public T withKeyGroupingComparator(final RawComparator<K2> groupingComparator) {
+    setKeyGroupingComparator(groupingComparator);
+    return thisAsMapReduceDriver();
+  }
+
+  /**
+   * Key value order comparator
+   */
+  protected Comparator<K1> keyValueOrderComparator;
+
+  /**
+   * Set the key value order comparator, similar to calling the following API
+   * calls but passing a real instance rather than just the class:
+   * <UL>
+   * <LI>pre 0.20.1 API:
+   * {@link org.apache.hadoop.mapred.JobConf#setOutputKeyComparatorClass(Class)}
+   * <LI>0.20.1+ API:
+   * {@link org.apache.hadoop.mapreduce.Job#setSortComparatorClass(Class)}
+   * </UL>
+   * 
+   * @param orderComparator
+   */
+  @SuppressWarnings("unchecked")
+  public void setKeyOrderComparator(final RawComparator<K2> orderComparator) {
+    keyValueOrderComparator = ReflectionUtils.newInstance(
+        orderComparator.getClass(), getConfiguration());
+  }
+
+  /**
+   * Identical to {@link #setKeyOrderComparator(RawComparator)}, but with a
+   * fluent programming style
+   * 
+   * @param orderComparator
+   *          Comparator to use in the shuffle stage for key value ordering
+   * @return this
+   */
+  public T withKeyOrderComparator(final RawComparator<K2> orderComparator) {
+    setKeyOrderComparator(orderComparator);
+    return thisAsMapReduceDriver();
+  }
+
+  @SuppressWarnings("rawtypes")
+  protected Map<M, List<Pair>> inputs = new HashMap<M, List<Pair>>();
+
+  /**
+   * Add an input to send to the specified mapper
+   * 
+   * @param mapper
+   *          The mapper to add the input to
+   * @param key
+   *          The key to add
+   * @param val
+   *          The value to add
+   * @param <K>
+   *          The key type
+   * @param <V>
+   *          The value type
+   */
+  protected <K, V> void addInput(final M mapper, final K key, final V val) {
+    if (!inputs.containsKey(mapper)) {
+      inputs.put(mapper, new ArrayList<Pair>());
+    }
+    inputs.get(mapper).add(copyPair(key, val));
+  }
+
+  /**
+   * Add an input to the specified mappper
+   * 
+   * @param mapper
+   *          The mapper to add the input to
+   * @param input
+   *          The (k, v) pair
+   * @param <K>
+   *          The key type
+   * @param <V>
+   *          The value type
+   */
+  protected <K, V> void addInput(final M mapper, final Pair<K, V> input) {
+    addInput(mapper, input.getFirst(), input.getSecond());
+  }
+
+  /**
+   * Add inputs to the specified mapper
+   * 
+   * @param mapper
+   *          The mapper to add the input to
+   * @param inputs
+   *          The (k, v) pairs
+   * @param <K>
+   *          The key type
+   * @param <V>
+   *          The value type
+   */
+  protected <K, V> void addAll(final M mapper, final List<Pair<K, V>> inputs) {
+    for (Pair<K, V> input : inputs) {
+      addInput(mapper, input);
+    }
+  }
+
+  /**
+   * Identical to addInput but supports a fluent programming style
+   * 
+   * @param mapper
+   *          The mapper to add the input to
+   * @param key
+   *          The key to add
+   * @param val
+   *          The value to add
+   * @param <K>
+   *          The key type
+   * @param <V>
+   *          The value type
+   * @return this
+   */
+  protected <K, V> T withInput(final M mapper, final K key, final V val) {
+    addInput(mapper, key, val);
+    return thisAsMapReduceDriver();
+  }
+
+  /**
+   * Identical to addInput but supports a fluent programming style
+   * 
+   * @param mapper
+   *          The mapper to add the input to
+   * @param input
+   *          The (k, v) pair to add
+   * @param <K>
+   *          The key type
+   * @param <V>
+   *          The value type
+   * @return this
+   */
+  protected <K, V> T withInput(final M mapper, final Pair<K, V> input) {
+    addInput(mapper, input);
+    return thisAsMapReduceDriver();
+  }
+
+  /**
+   * Identical to addAll but supports a fluent programming style
+   * 
+   * @param mapper
+   *          The mapper to add the input to
+   * @param inputs
+   *          The (k, v) pairs to add
+   * @param <K>
+   *          The key type
+   * @param <V>
+   *          The value type
+   * @return this
+   */
+  protected <K, V> T withAll(final M mapper, final List<Pair<K, V>> inputs) {
+    addAll(mapper, inputs);
+    return thisAsMapReduceDriver();
+  }
+
+  protected void preRunChecks(Set<M> mappers, Object reducer) {
+    for (M mapper : mappers) {
+      if (inputs.get(mapper) == null || inputs.get(mapper).isEmpty()) {
+        throw new IllegalStateException(String.format(
+            "No input was provided for mapper %s", mapper));
+      }
+    }
+
+    if (reducer == null) {
+      throw new IllegalStateException("No reducer class was provided");
+    }
+    if (driverReused()) {
+      throw new IllegalStateException("Driver reuse not allowed");
+    } else {
+      setUsedOnceStatus();
+    }
+  }
+
+  @SuppressWarnings("unchecked")
+  private T thisAsMapReduceDriver() {
+    return (T) this;
+  }
+
+}
diff --git a/src/main/java/org/apache/hadoop/mrunit/mapreduce/MapReduceDriver.java b/src/main/java/org/apache/hadoop/mrunit/mapreduce/MapReduceDriver.java
index 4bf05fd..6aa8b5a 100644
--- a/src/main/java/org/apache/hadoop/mrunit/mapreduce/MapReduceDriver.java
+++ b/src/main/java/org/apache/hadoop/mrunit/mapreduce/MapReduceDriver.java
@@ -17,25 +17,21 @@
  */
 package org.apache.hadoop.mrunit.mapreduce;
 
-import static org.apache.hadoop.mrunit.internal.util.ArgumentChecker.returnNonNull;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapreduce.*;
+import org.apache.hadoop.mrunit.MapReduceDriverBase;
+import org.apache.hadoop.mrunit.internal.counters.CounterWrapper;
+import org.apache.hadoop.mrunit.types.KeyValueReuseList;
+import org.apache.hadoop.mrunit.types.Pair;
 
 import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 
-import org.apache.commons.logging.Log;
-import org.apache.commons.logging.LogFactory;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapreduce.Counters;
-import org.apache.hadoop.mapreduce.InputFormat;
-import org.apache.hadoop.mapreduce.Mapper;
-import org.apache.hadoop.mapreduce.OutputFormat;
-import org.apache.hadoop.mapreduce.Reducer;
-import org.apache.hadoop.mrunit.MapReduceDriverBase;
-import org.apache.hadoop.mrunit.internal.counters.CounterWrapper;
-import org.apache.hadoop.mrunit.types.KeyValueReuseList;
-import org.apache.hadoop.mrunit.types.Pair;
+import static org.apache.hadoop.mrunit.internal.util.ArgumentChecker.returnNonNull;
 
 /**
  * Harness that allows you to test a Mapper and a Reducer instance together You
@@ -48,7 +44,8 @@
  * pair, representing a single unit test.
  */
 
-public class MapReduceDriver<K1, V1, K2, V2, K3, V3> extends
+public class MapReduceDriver<K1, V1, K2, V2, K3, V3>
+    extends
     MapReduceDriverBase<K1, V1, K2, V2, K3, V3, MapReduceDriver<K1, V1, K2, V2, K3, V3>> {
 
   public static final Log LOG = LogFactory.getLog(MapReduceDriver.class);
@@ -82,7 +79,7 @@
 
   /**
    * Set the Mapper instance to use with this test driver
-   *
+   * 
    * @param m
    *          the Mapper instance to use
    */
@@ -106,7 +103,7 @@
 
   /**
    * Sets the reducer object to use for this test
-   *
+   * 
    * @param r
    *          The reducer object to use
    */
@@ -116,7 +113,7 @@
 
   /**
    * Identical to setReducer(), but with fluent programming style
-   *
+   * 
    * @param r
    *          The Reducer to use
    * @return this
@@ -136,7 +133,7 @@
 
   /**
    * Sets the reducer object to use as a combiner for this test
-   *
+   * 
    * @param c
    *          The combiner object to use
    */
@@ -146,7 +143,7 @@
 
   /**
    * Identical to setCombiner(), but with fluent programming style
-   *
+   * 
    * @param c
    *          The Combiner to use
    * @return this
@@ -171,7 +168,7 @@
 
   /**
    * Sets the counters object to use for this test.
-   *
+   * 
    * @param ctrs
    *          The counters object to use.
    */
@@ -190,7 +187,7 @@
   /**
    * Configure {@link Reducer} to output with a real {@link OutputFormat}. Set
    * {@link InputFormat} to read output back in for use with run* methods
-   *
+   * 
    * @param outputFormatClass
    * @param inputFormatClass
    * @return this for fluent style
@@ -204,64 +201,26 @@
     return this;
   }
 
-  /**
-   * The private class to manage starting the reduce phase is used for type
-   * genericity reasons. This class is used in the run() method.
-   */
-  private class ReducePhaseRunner<OUTKEY, OUTVAL> {
-    private List<Pair<OUTKEY, OUTVAL>> runReduce(
-        final List<KeyValueReuseList<K2, V2>> inputs,
-        final Reducer<K2, V2, OUTKEY, OUTVAL> reducer) throws IOException {
-
-      final List<Pair<OUTKEY, OUTVAL>> reduceOutputs = new ArrayList<Pair<OUTKEY, OUTVAL>>();
-
-      if (!inputs.isEmpty()) {
-        if (LOG.isDebugEnabled()) {
-          final StringBuilder sb = new StringBuilder();
-          for (List<Pair<K2, V2>> input : inputs) {
-            formatPairList(input, sb);
-            LOG.debug("Reducing input " + sb);
-            sb.delete(0, sb.length());
-          }
-        }
-
-        final ReduceDriver<K2, V2, OUTKEY, OUTVAL> reduceDriver = ReduceDriver
-            .newReduceDriver(reducer).withCounters(getCounters())
-            .withConfiguration(getConfiguration()).withAllElements(inputs);
-
-        if (getOutputSerializationConfiguration() != null) {
-          reduceDriver
-              .withOutputSerializationConfiguration(getOutputSerializationConfiguration());
-        }
-
-        if (outputFormatClass != null) {
-          reduceDriver.withOutputFormat(outputFormatClass, inputFormatClass);
-        }
-
-        reduceOutputs.addAll(reduceDriver.run());
-      }
-
-      return reduceOutputs;
-    }
-  }
-
-  protected List<KeyValueReuseList<K2,V2>> sortAndGroup(final List<Pair<K2, V2>> mapOutputs){
-    if(mapOutputs.isEmpty()) {
+  protected List<KeyValueReuseList<K2, V2>> sortAndGroup(
+      final List<Pair<K2, V2>> mapOutputs) {
+    if (mapOutputs.isEmpty()) {
       return Collections.emptyList();
     }
 
-    if (keyValueOrderComparator == null || keyGroupComparator == null){
+    if (keyValueOrderComparator == null || keyGroupComparator == null) {
       JobConf conf = new JobConf(getConfiguration());
       conf.setMapOutputKeyClass(mapOutputs.get(0).getFirst().getClass());
-      if (keyGroupComparator == null){
+      if (keyGroupComparator == null) {
         keyGroupComparator = conf.getOutputValueGroupingComparator();
       }
       if (keyValueOrderComparator == null) {
         keyValueOrderComparator = conf.getOutputKeyComparator();
       }
     }
-    ReduceFeeder<K2,V2> reduceFeeder = new ReduceFeeder<K2,V2>(getConfiguration());
-    return reduceFeeder.sortAndGroup(mapOutputs, keyValueOrderComparator, keyGroupComparator);
+    ReduceFeeder<K2, V2> reduceFeeder = new ReduceFeeder<K2, V2>(
+        getConfiguration());
+    return reduceFeeder.sortAndGroup(mapOutputs, keyValueOrderComparator,
+        keyGroupComparator);
   }
 
   @Override
@@ -276,16 +235,20 @@
           .withCounters(getCounters()).withConfiguration(getConfiguration())
           .withAll(inputList).withMapInputPath(getMapInputPath()).run());
       if (myCombiner != null) {
-        // User has specified a combiner. Run this and replace the mapper outputs
+        // User has specified a combiner. Run this and replace the mapper
+        // outputs
         // with the result of the combiner.
         LOG.debug("Starting combine phase with combiner: " + myCombiner);
-        mapOutputs = new ReducePhaseRunner<K2, V2>().runReduce(
-            sortAndGroup(mapOutputs), myCombiner);
+        mapOutputs = new ReducePhaseRunner<K2, V2, K2, V2>(inputFormatClass,
+            getConfiguration(), counters,
+            getOutputSerializationConfiguration(), outputFormatClass)
+            .runReduce(sortAndGroup(mapOutputs), myCombiner);
       }
       // Run the reduce phase.
       LOG.debug("Starting reduce phase with reducer: " + myReducer);
-      return new ReducePhaseRunner<K3, V3>().runReduce(sortAndGroup(mapOutputs),
-          myReducer);
+      return new ReducePhaseRunner<K2, V2, K3, V3>(inputFormatClass,
+          getConfiguration(), counters, getOutputSerializationConfiguration(),
+          outputFormatClass).runReduce(sortAndGroup(mapOutputs), myReducer);
     } finally {
       cleanupDistributedCache();
     }
@@ -299,7 +262,7 @@
   /**
    * Returns a new MapReduceDriver without having to specify the generic types
    * on the right hand side of the object create statement.
-   *
+   * 
    * @return new MapReduceDriver
    */
   public static <K1, V1, K2, V2, K3, V3> MapReduceDriver<K1, V1, K2, V2, K3, V3> newMapReduceDriver() {
@@ -309,7 +272,7 @@
   /**
    * Returns a new MapReduceDriver without having to specify the generic types
    * on the right hand side of the object create statement.
-   *
+   * 
    * @param mapper
    *          passed to MapReduceDriver constructor
    * @param reducer
@@ -324,7 +287,7 @@
   /**
    * Returns a new MapReduceDriver without having to specify the generic types
    * on the right hand side of the object create statement.
-   *
+   * 
    * @param mapper
    *          passed to MapReduceDriver constructor
    * @param reducer
diff --git a/src/main/java/org/apache/hadoop/mrunit/mapreduce/MultipleInputsMapReduceDriver.java b/src/main/java/org/apache/hadoop/mrunit/mapreduce/MultipleInputsMapReduceDriver.java
new file mode 100644
index 0000000..a77ad08
--- /dev/null
+++ b/src/main/java/org/apache/hadoop/mrunit/mapreduce/MultipleInputsMapReduceDriver.java
@@ -0,0 +1,503 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mrunit.mapreduce;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapreduce.*;
+import org.apache.hadoop.mrunit.internal.counters.CounterWrapper;
+import org.apache.hadoop.mrunit.internal.driver.MultipleInputsMapReduceDriverBase;
+import org.apache.hadoop.mrunit.types.KeyValueReuseList;
+import org.apache.hadoop.mrunit.types.Pair;
+
+import java.io.IOException;
+import java.util.*;
+
+import static org.apache.hadoop.mrunit.internal.util.ArgumentChecker.returnNonNull;
+
+/**
+ * Harness that allows you to test multiple Mappers and a Reducer instance
+ * together (along with an optional combiner). You provide the input keys and
+ * values that should be sent to each Mapper, and outputs you expect to be sent
+ * by the Reducer to the collector for those inputs. By calling runTest(), the
+ * harness will deliver the inputs to the respective Mappers, feed the
+ * intermediate results to the Reducer (without checking them), and will check
+ * the Reducer's outputs against the expected results.
+ * 
+ * If a combiner is specified, it will run exactly once after all the Mappers
+ * and before the Reducer
+ * 
+ * @param <K1>
+ *          The common map output key type
+ * @param <V1>
+ *          The common map output value type
+ * @param <K2>
+ *          The reduce output key type
+ * @param <V2>
+ *          The reduce output value type
+ */
+public class MultipleInputsMapReduceDriver<K1, V1, K2, V2>
+    extends
+        MultipleInputsMapReduceDriverBase<Mapper, K1, V1, K2, V2, MultipleInputsMapReduceDriver<K1, V1, K2, V2>> {
+  public static final Log LOG = LogFactory
+      .getLog(MultipleInputsMapReduceDriver.class);
+
+  private Set<Mapper> mappers = new HashSet<Mapper>();
+
+  /**
+   * Add a mapper to use with this test driver
+   * 
+   * @param mapper
+   *          The mapper instance to add
+   * @param <K>
+   *          The input key type to the mapper
+   * @param <V>
+   *          The input value type to the mapper
+   */
+  public <K, V> void addMapper(final Mapper<K, V, K1, V1> mapper) {
+    this.mappers.add(returnNonNull(mapper));
+  }
+
+  /**
+   * Identical to addMapper but supports a fluent programming style
+   * 
+   * @param mapper
+   *          The mapper instance to add
+   * @param <K>
+   *          The input key type to the mapper
+   * @param <V>
+   *          The input value type to the mapper
+   * @return this
+   */
+  public <K, V> MultipleInputsMapReduceDriver<K1, V1, K2, V2> withMapper(
+      final Mapper<K, V, K1, V1> mapper) {
+    addMapper(mapper);
+    return this;
+  }
+
+  /**
+   * @return The Mapper instances being used by this test
+   */
+  public Collection<Mapper> getMappers() {
+    return mappers;
+  }
+
+  private Reducer<K1, V1, K1, V1> combiner;
+
+  /**
+   * Set the combiner to use with this test driver
+   * 
+   * @param combiner
+   *          The combiner instance to use
+   */
+  public void setCombiner(final Reducer<K1, V1, K1, V1> combiner) {
+    this.combiner = returnNonNull(combiner);
+  }
+
+  /**
+   * Identical to setCombiner but supports a fluent programming style
+   * 
+   * @param combiner
+   *          The combiner instance to use
+   * @return this
+   */
+  public MultipleInputsMapReduceDriver<K1, V1, K2, V2> withCombiner(
+      final Reducer<K1, V1, K1, V1> combiner) {
+    setCombiner(combiner);
+    return this;
+  }
+
+  /**
+   * @return The combiner instance being used by this test
+   */
+  public Reducer<K1, V1, K1, V1> getCombiner() {
+    return combiner;
+  }
+
+  private Reducer<K1, V1, K2, V2> reducer;
+
+  /**
+   * Set the reducer to use with this test driver
+   * 
+   * @param reducer
+   *          The reducer instance to use
+   */
+  public void setReducer(final Reducer<K1, V1, K2, V2> reducer) {
+    this.reducer = returnNonNull(reducer);
+  }
+
+  /**
+   * Identical to setReducer but supports a fluent programming style
+   * 
+   * @param reducer
+   *          The reducer instance to use
+   * @return this
+   */
+  public MultipleInputsMapReduceDriver<K1, V1, K2, V2> withReducer(
+      final Reducer<K1, V1, K2, V2> reducer) {
+    setReducer(reducer);
+    return this;
+  }
+
+  /**
+   * @return Get the reducer instance being used by this test
+   */
+  public Reducer<K1, V1, K2, V2> getReducer() {
+    return reducer;
+  }
+
+  private Counters counters;
+
+  /**
+   * @return The counters used in this test
+   */
+  public Counters getCounters() {
+    return counters;
+  }
+
+  /**
+   * Sets the counters object to use for this test
+   * 
+   * @param counters
+   *          The counters object to use
+   */
+  public void setCounters(Counters counters) {
+    this.counters = counters;
+    counterWrapper = new CounterWrapper(counters);
+  }
+
+  /**
+   * Identical to setCounters but supports a fluent programming style
+   * 
+   * @param counters
+   *          The counters object to use
+   * @return this
+   */
+  public MultipleInputsMapReduceDriver<K1, V1, K2, V2> withCounter(
+      Counters counters) {
+    setCounters(counters);
+    return this;
+  }
+
+  private Class<? extends OutputFormat> outputFormatClass;
+
+  /**
+   * Configure {@link Reducer} to output with a real {@link OutputFormat}.
+   * 
+   * @param outputFormatClass
+   *          The OutputFormat class
+   * @return this
+   */
+  public MultipleInputsMapReduceDriver<K1, V1, K2, V2> withOutputFormat(
+      final Class<? extends OutputFormat> outputFormatClass) {
+    this.outputFormatClass = returnNonNull(outputFormatClass);
+    return this;
+  }
+
+  private Class<? extends InputFormat> inputFormatClass;
+
+  /**
+   * Set the InputFormat
+   * 
+   * @param inputFormatClass
+   *          The InputFormat class
+   * @return this
+   */
+  public MultipleInputsMapReduceDriver<K1, V1, K2, V2> withInputFormat(
+      final Class<? extends InputFormat> inputFormatClass) {
+    this.inputFormatClass = returnNonNull(inputFormatClass);
+    return this;
+  }
+
+  /**
+   * Construct a driver with the specified Reducer. Note that a Combiner can be
+   * set separately.
+   * 
+   * @param reducer
+   *          The reducer to use
+   */
+  public MultipleInputsMapReduceDriver(Reducer<K1, V1, K2, V2> reducer) {
+    this();
+    this.reducer = reducer;
+  }
+
+  /**
+   * Construct a driver with the specified Combiner and Reducers
+   * 
+   * @param combiner
+   *          The combiner to use
+   * @param reducer
+   *          The reducer to use
+   */
+  public MultipleInputsMapReduceDriver(Reducer<K1, V1, K1, V1> combiner,
+                                       Reducer<K1, V1, K2, V2> reducer) {
+    this(reducer);
+    this.combiner = combiner;
+  }
+
+  /**
+   * Construct a driver without specifying a Combiner nor a Reducer. Note that
+   * these can be set with the appropriate set methods and that at least the
+   * Reducer must be set.
+   */
+  public MultipleInputsMapReduceDriver() {
+    setCounters(new Counters());
+  }
+
+  /**
+   * Static factory-style method to construct a driver instance with the
+   * specified Combiner and Reducer
+   * 
+   * @param combiner
+   *          The combiner to use
+   * @param reducer
+   *          The reducer to use
+   * @param <K1>
+   *          The common output key type of the mappers
+   * @param <V1>
+   *          The common output value type of the mappers
+   * @param <K2>
+   *          The output key type of the reducer
+   * @param <V2>
+   *          The output value type of the reducer
+   * @return this to support fluent programming style
+   */
+  public static <K1, V1, K2, V2> MultipleInputsMapReduceDriver<K1, V1, K2, V2> newMultipleInputMapReduceDriver(
+      final Reducer<K1, V1, K1, V1> combiner,
+      final Reducer<K1, V1, K2, V2> reducer) {
+    return new MultipleInputsMapReduceDriver<K1, V1, K2, V2>(combiner, reducer);
+  }
+
+  /**
+   * Static factory-style method to construct a driver instance with the
+   * specified Reducer
+   * 
+   * @param reducer
+   *          The reducer to use
+   * @param <K1>
+   *          The common output key type of the mappers
+   * @param <V1>
+   *          The common output value type of the mappers
+   * @param <K2>
+   *          The output key type of the reducer
+   * @param <V2>
+   *          The output value type of the reducer
+   * @return this to support fluent programming style
+   */
+  public static <K1, V1, K2, V2> MultipleInputsMapReduceDriver<K1, V1, K2, V2> newMultipleInputMapReduceDriver(
+      final Reducer<K1, V1, K2, V2> reducer) {
+    return new MultipleInputsMapReduceDriver<K1, V1, K2, V2>(reducer);
+  }
+
+  /**
+   * Static factory-style method to construct a driver instance without
+   * specifying a Combiner nor a Reducer. Note that these can be set separately
+   * by using the appropriate set (or with) methods and that at least a Reducer
+   * must be set
+   * 
+   * @param <K1>
+   *          The common output key type of the mappers
+   * @param <V1>
+   *          The common output value type of the mappers
+   * @param <K2>
+   *          The output key type of the reducer
+   * @param <V2>
+   *          The output value type of the reducer
+   * @return this to support fluent programming style
+   */
+  public static <K1, V1, K2, V2> MultipleInputsMapReduceDriver<K1, V1, K2, V2> newMultipleInputMapReduceDriver() {
+    return new MultipleInputsMapReduceDriver<K1, V1, K2, V2>();
+  }
+
+  /**
+   * Add the specified (key, val) pair to the specified mapper
+   * 
+   * @param mapper
+   *          The mapper to add the input pair to
+   * @param key
+   *          The key
+   * @param val
+   *          The value
+   * @param <K>
+   *          The type of the key
+   * @param <V>
+   *          The type of the value
+   */
+  public <K, V> void addInput(final Mapper<K, V, K1, V1> mapper, final K key,
+      final V val) {
+    super.addInput(mapper, key, val);
+  }
+
+  /**
+   * Add the specified input pair to the specified mapper
+   * 
+   * @param mapper
+   *          The mapper to add the input pair to
+   * @param input
+   *          The (k,v) pair to add
+   * @param <K>
+   *          The type of the key
+   * @param <V>
+   *          The type of the value
+   */
+  public <K, V> void addInput(final Mapper<K, V, K1, V1> mapper,
+      final Pair<K, V> input) {
+    super.addInput(mapper, input);
+  }
+
+  /**
+   * Add the specified input pairs to the specified mapper
+   * 
+   * @param mapper
+   *          The mapper to add the input pairs to
+   * @param inputs
+   *          The (k, v) pairs to add
+   * @param <K>
+   *          The type of the key
+   * @param <V>
+   *          The type of the value
+   */
+  public <K, V> void addAll(final Mapper<K, V, K1, V1> mapper,
+      final List<Pair<K, V>> inputs) {
+    super.addAll(mapper, inputs);
+  }
+
+  /**
+   * Identical to addInput but supports fluent programming style
+   * 
+   * @param mapper
+   *          The mapper to add the input pair to
+   * @param key
+   *          The key
+   * @param val
+   *          The value
+   * @param <K>
+   *          The type of the key
+   * @param <V>
+   *          The type of the value
+   * @return this
+   */
+  public <K, V> MultipleInputsMapReduceDriver<K1, V1, K2, V2> withInput(
+      final Mapper<K, V, K1, V1> mapper, final K key, final V val) {
+    return super.withInput(mapper, key, val);
+  }
+
+  /**
+   * Identical to addInput but supports fluent programming style
+   * 
+   * @param mapper
+   *          The mapper to add the input pairs to
+   * @param inputs
+   *          The (k, v) pairs to add
+   * @param <K>
+   *          The type of the key
+   * @param <V>
+   *          The type of the value
+   * @return this
+   */
+  public <K, V> MultipleInputsMapReduceDriver<K1, V1, K2, V2> withInput(
+      final Mapper<K, V, K1, V1> mapper, final Pair<K, V> input) {
+    return super.withInput(mapper, input);
+  }
+
+  /**
+   * Identical to addInput but supports fluent programming style
+   * 
+   * @param mapper
+   *          The mapper to add the input pairs to
+   * @param inputs
+   *          The (k, v) pairs to add
+   * @param <K>
+   *          The type of the key
+   * @param <V>
+   *          The type of the value
+   * @return this
+   */
+  public <K, V> MultipleInputsMapReduceDriver<K1, V1, K2, V2> withAll(
+      final Mapper<K, V, K1, V1> mapper, final List<Pair<K, V>> inputs) {
+    return super.withAll(mapper, inputs);
+  }
+
+  @Override
+  protected void preRunChecks(Set<Mapper> mappers, Object reducer) {
+    if (mappers.isEmpty()) {
+      throw new IllegalStateException("No mappers were provided");
+    }
+    super.preRunChecks(mappers, reducer);
+  }
+
+  protected List<KeyValueReuseList<K1, V1>> sortAndGroup(
+      final List<Pair<K1, V1>> mapOutputs) {
+    if (mapOutputs.isEmpty()) {
+      return Collections.emptyList();
+    }
+
+    if (keyValueOrderComparator == null || keyGroupComparator == null) {
+      JobConf conf = new JobConf(getConfiguration());
+      conf.setMapOutputKeyClass(mapOutputs.get(0).getFirst().getClass());
+      if (keyGroupComparator == null) {
+        keyGroupComparator = conf.getOutputValueGroupingComparator();
+      }
+      if (keyValueOrderComparator == null) {
+        keyValueOrderComparator = conf.getOutputKeyComparator();
+      }
+    }
+    ReduceFeeder<K1, V1> reduceFeeder = new ReduceFeeder<K1, V1>(
+        getConfiguration());
+    return reduceFeeder.sortAndGroup(mapOutputs, keyValueOrderComparator,
+        keyGroupComparator);
+  }
+
+  @SuppressWarnings("unchecked")
+  @Override
+  public List<Pair<K2, V2>> run() throws IOException {
+    try {
+      preRunChecks(mappers, reducer);
+      initDistributedCache();
+
+      List<Pair<K1, V1>> outputs = new ArrayList<Pair<K1, V1>>();
+
+      for (Mapper mapper : mappers) {
+        MapDriver mapDriver = MapDriver.newMapDriver(mapper);
+        mapDriver.setCounters(counters);
+        mapDriver.setConfiguration(getConfiguration());
+        mapDriver.addAll(inputs.get(mapper));
+        mapDriver.withMapInputPath(getMapInputPath(mapper));
+        outputs.addAll(mapDriver.run());
+      }
+
+      if (combiner != null) {
+        LOG.debug("Starting combine phase with combiner: " + combiner);
+        outputs = new ReducePhaseRunner<K1, V1, K1, V1>(inputFormatClass,
+            getConfiguration(), counters,
+            getOutputSerializationConfiguration(), outputFormatClass)
+            .runReduce(sortAndGroup(outputs), combiner);
+      }
+
+      LOG.debug("Starting reduce phase with reducer: " + reducer);
+
+      return new ReducePhaseRunner<K1, V1, K2, V2>(inputFormatClass,
+          getConfiguration(), counters, getOutputSerializationConfiguration(),
+          outputFormatClass).runReduce(sortAndGroup(outputs), reducer);
+    } finally {
+      cleanupDistributedCache();
+    }
+  }
+}
diff --git a/src/main/java/org/apache/hadoop/mrunit/mapreduce/ReducePhaseRunner.java b/src/main/java/org/apache/hadoop/mrunit/mapreduce/ReducePhaseRunner.java
new file mode 100644
index 0000000..73e7152
--- /dev/null
+++ b/src/main/java/org/apache/hadoop/mrunit/mapreduce/ReducePhaseRunner.java
@@ -0,0 +1,90 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mrunit.mapreduce;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.Counters;
+import org.apache.hadoop.mapreduce.InputFormat;
+import org.apache.hadoop.mapreduce.OutputFormat;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.hadoop.mrunit.TestDriver;
+import org.apache.hadoop.mrunit.types.KeyValueReuseList;
+import org.apache.hadoop.mrunit.types.Pair;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+class ReducePhaseRunner<INKEY, INVAL, OUTKEY, OUTVAL> {
+  public static final Log LOG = LogFactory.getLog(ReducePhaseRunner.class);
+
+  private final Configuration configuration;
+  private final Counters counters;
+  private Configuration outputSerializationConfiguration;
+  private Class<? extends OutputFormat> outputFormatClass;
+  private Class<? extends InputFormat> inputFormatClass;
+
+  ReducePhaseRunner(Class<? extends InputFormat> inputFormatClass,
+      Configuration configuration, Counters counters,
+      Configuration outputSerializationConfiguration,
+      Class<? extends OutputFormat> outputFormatClass) {
+    this.inputFormatClass = inputFormatClass;
+    this.configuration = configuration;
+    this.counters = counters;
+    this.outputSerializationConfiguration = outputSerializationConfiguration;
+    this.outputFormatClass = outputFormatClass;
+  }
+
+  public List<Pair<OUTKEY, OUTVAL>> runReduce(
+      final List<KeyValueReuseList<INKEY, INVAL>> inputs,
+      final Reducer<INKEY, INVAL, OUTKEY, OUTVAL> reducer) throws IOException {
+
+    final List<Pair<OUTKEY, OUTVAL>> reduceOutputs = new ArrayList<Pair<OUTKEY, OUTVAL>>();
+
+    if (!inputs.isEmpty()) {
+      if (LOG.isDebugEnabled()) {
+        final StringBuilder sb = new StringBuilder();
+        for (List<Pair<INKEY, INVAL>> input : inputs) {
+          TestDriver.formatValueList(input, sb);
+          LOG.debug("Reducing input " + sb);
+          sb.delete(0, sb.length());
+        }
+      }
+
+      final ReduceDriver<INKEY, INVAL, OUTKEY, OUTVAL> reduceDriver = ReduceDriver
+          .newReduceDriver(reducer).withCounters(counters)
+          .withConfiguration(configuration).withAllElements(inputs);
+
+      if (outputSerializationConfiguration != null) {
+        reduceDriver
+            .withOutputSerializationConfiguration(outputSerializationConfiguration);
+      }
+
+      if (outputFormatClass != null) {
+        reduceDriver.withOutputFormat(outputFormatClass, inputFormatClass);
+      }
+
+      reduceOutputs.addAll(reduceDriver.run());
+    }
+
+    return reduceOutputs;
+  }
+}
\ No newline at end of file
diff --git a/src/test/java/org/apache/hadoop/mrunit/TestMapOutputShuffler.java b/src/test/java/org/apache/hadoop/mrunit/TestMapOutputShuffler.java
new file mode 100644
index 0000000..1157c74
--- /dev/null
+++ b/src/test/java/org/apache/hadoop/mrunit/TestMapOutputShuffler.java
@@ -0,0 +1,130 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mrunit;
+
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mrunit.types.Pair;
+import org.junit.Before;
+import org.junit.Test;
+
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.hadoop.mrunit.ExtendedAssert.assertListEquals;
+import static org.junit.Assert.assertEquals;
+
+public class TestMapOutputShuffler {
+  private MapOutputShuffler<Text, Text> shuffler;
+
+  @Before
+  public void setUp() {
+    shuffler = new MapOutputShuffler<Text, Text>(null, null, null);
+  }
+
+  @Test
+  public void testEmptyShuffle() {
+    final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
+    final List<Pair<Text, List<Text>>> outputs = shuffler.shuffle(inputs);
+    assertEquals(0, outputs.size());
+  }
+
+  // just shuffle a single (k, v) pair
+  @Test
+  public void testSingleShuffle() {
+    final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
+    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("b")));
+
+    final List<Pair<Text, List<Text>>> outputs = shuffler.shuffle(inputs);
+
+    final List<Pair<Text, List<Text>>> expected = new ArrayList<Pair<Text, List<Text>>>();
+    final List<Text> sublist = new ArrayList<Text>();
+    sublist.add(new Text("b"));
+    expected.add(new Pair<Text, List<Text>>(new Text("a"), sublist));
+
+    assertListEquals(expected, outputs);
+  }
+
+  // shuffle multiple values from the same key.
+  @Test
+  public void testShuffleOneKey() {
+    final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
+    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("b")));
+    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("c")));
+
+    final List<Pair<Text, List<Text>>> outputs = shuffler.shuffle(inputs);
+
+    final List<Pair<Text, List<Text>>> expected = new ArrayList<Pair<Text, List<Text>>>();
+    final List<Text> sublist = new ArrayList<Text>();
+    sublist.add(new Text("b"));
+    sublist.add(new Text("c"));
+    expected.add(new Pair<Text, List<Text>>(new Text("a"), sublist));
+
+    assertListEquals(expected, outputs);
+  }
+
+  // shuffle multiple keys
+  @Test
+  public void testMultiShuffle1() {
+    final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
+    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("x")));
+    inputs.add(new Pair<Text, Text>(new Text("b"), new Text("z")));
+    inputs.add(new Pair<Text, Text>(new Text("b"), new Text("w")));
+    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("y")));
+
+    final List<Pair<Text, List<Text>>> outputs = shuffler.shuffle(inputs);
+
+    final List<Pair<Text, List<Text>>> expected = new ArrayList<Pair<Text, List<Text>>>();
+    final List<Text> sublist1 = new ArrayList<Text>();
+    sublist1.add(new Text("x"));
+    sublist1.add(new Text("y"));
+    expected.add(new Pair<Text, List<Text>>(new Text("a"), sublist1));
+
+    final List<Text> sublist2 = new ArrayList<Text>();
+    sublist2.add(new Text("z"));
+    sublist2.add(new Text("w"));
+    expected.add(new Pair<Text, List<Text>>(new Text("b"), sublist2));
+
+    assertListEquals(expected, outputs);
+  }
+
+  // shuffle multiple keys that are out-of-order to start.
+  @Test
+  public void testMultiShuffle2() {
+    final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
+    inputs.add(new Pair<Text, Text>(new Text("b"), new Text("z")));
+    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("x")));
+    inputs.add(new Pair<Text, Text>(new Text("b"), new Text("w")));
+    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("y")));
+
+    final List<Pair<Text, List<Text>>> outputs = shuffler.shuffle(inputs);
+
+    final List<Pair<Text, List<Text>>> expected = new ArrayList<Pair<Text, List<Text>>>();
+    final List<Text> sublist1 = new ArrayList<Text>();
+    sublist1.add(new Text("x"));
+    sublist1.add(new Text("y"));
+    expected.add(new Pair<Text, List<Text>>(new Text("a"), sublist1));
+
+    final List<Text> sublist2 = new ArrayList<Text>();
+    sublist2.add(new Text("z"));
+    sublist2.add(new Text("w"));
+    expected.add(new Pair<Text, List<Text>>(new Text("b"), sublist2));
+
+    assertListEquals(expected, outputs);
+  }
+}
diff --git a/src/test/java/org/apache/hadoop/mrunit/TestMapReduceDriver.java b/src/test/java/org/apache/hadoop/mrunit/TestMapReduceDriver.java
index 50da602..3a60f73 100644
--- a/src/test/java/org/apache/hadoop/mrunit/TestMapReduceDriver.java
+++ b/src/test/java/org/apache/hadoop/mrunit/TestMapReduceDriver.java
@@ -17,14 +17,6 @@
  */
 package org.apache.hadoop.mrunit;
 
-import static org.apache.hadoop.mrunit.ExtendedAssert.*;
-import static org.junit.Assert.*;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Iterator;
-import java.util.List;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.IntWritable;
@@ -32,17 +24,7 @@
 import org.apache.hadoop.io.RawComparator;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.serializer.JavaSerializationComparator;
-import org.apache.hadoop.mapred.FileSplit;
-import org.apache.hadoop.mapred.JobConf;
-import org.apache.hadoop.mapred.MapReduceBase;
-import org.apache.hadoop.mapred.Mapper;
-import org.apache.hadoop.mapred.OutputCollector;
-import org.apache.hadoop.mapred.Reducer;
-import org.apache.hadoop.mapred.Reporter;
-import org.apache.hadoop.mapred.SequenceFileInputFormat;
-import org.apache.hadoop.mapred.SequenceFileOutputFormat;
-import org.apache.hadoop.mapred.TextInputFormat;
-import org.apache.hadoop.mapred.TextOutputFormat;
+import org.apache.hadoop.mapred.*;
 import org.apache.hadoop.mapred.lib.IdentityMapper;
 import org.apache.hadoop.mapred.lib.IdentityReducer;
 import org.apache.hadoop.mapred.lib.LongSumReducer;
@@ -53,7 +35,14 @@
 import org.junit.Rule;
 import org.junit.Test;
 
-import com.google.common.collect.Lists;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import static org.apache.hadoop.mrunit.ExtendedAssert.assertListEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
 
 public class TestMapReduceDriver {
 
@@ -100,10 +89,10 @@
   public void testUncomparable() throws IOException {
     Text k = new Text("test");
     Object v = new UncomparableWritable(2);
-    MapReduceDriver.newMapReduceDriver(
-        new IdentityMapper<Text, Object>(),
-        new IdentityReducer<Text, Object>())
-        .withInput(k, v).withOutput(k, v).runTest();
+    MapReduceDriver
+        .newMapReduceDriver(new IdentityMapper<Text, Object>(),
+            new IdentityReducer<Text, Object>()).withInput(k, v)
+        .withOutput(k, v).runTest();
   }
 
   @Test
@@ -127,8 +116,10 @@
   @Test
   public void testTestRun3() throws IOException {
     thrown.expectAssertionErrorMessage("2 Error(s)");
-    thrown.expectAssertionErrorMessage("Missing expected output (foo, 52) at position 0, got (bar, 12).");
-    thrown.expectAssertionErrorMessage("Missing expected output (bar, 12) at position 1, got (foo, 52).");
+    thrown
+        .expectAssertionErrorMessage("Missing expected output (foo, 52) at position 0, got (bar, 12).");
+    thrown
+        .expectAssertionErrorMessage("Missing expected output (bar, 12) at position 1, got (foo, 52).");
     driver.withInput(new Text("foo"), new LongWritable(FOO_IN_A))
         .withInput(new Text("bar"), new LongWritable(BAR_IN))
         .withInput(new Text("foo"), new LongWritable(FOO_IN_B))
@@ -139,13 +130,18 @@
   @Test
   public void testAddAll() throws IOException {
     final List<Pair<Text, LongWritable>> inputs = new ArrayList<Pair<Text, LongWritable>>();
-    inputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(FOO_IN_A)));
-    inputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(FOO_IN_B)));
-    inputs.add(new Pair<Text, LongWritable>(new Text("bar"), new LongWritable(BAR_IN)));
+    inputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(
+        FOO_IN_A)));
+    inputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(
+        FOO_IN_B)));
+    inputs.add(new Pair<Text, LongWritable>(new Text("bar"), new LongWritable(
+        BAR_IN)));
 
     final List<Pair<Text, LongWritable>> outputs = new ArrayList<Pair<Text, LongWritable>>();
-    outputs.add(new Pair<Text, LongWritable>(new Text("bar"), new LongWritable(BAR_IN)));
-    outputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(FOO_OUT)));
+    outputs.add(new Pair<Text, LongWritable>(new Text("bar"), new LongWritable(
+        BAR_IN)));
+    outputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(
+        FOO_OUT)));
 
     driver.withAll(inputs).withAllOutput(outputs).runTest();
   }
@@ -177,97 +173,6 @@
     driver.runTest();
   }
 
-  @Test
-  public void testEmptyShuffle() {
-    final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
-    final List<Pair<Text, List<Text>>> outputs = driver2.shuffle(inputs);
-    assertEquals(0, outputs.size());
-  }
-
-  // just shuffle a single (k, v) pair
-  @Test
-  public void testSingleShuffle() {
-    final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
-    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("b")));
-
-    final List<Pair<Text, List<Text>>> outputs = driver2.shuffle(inputs);
-
-    final List<Pair<Text, List<Text>>> expected = new ArrayList<Pair<Text, List<Text>>>();
-    final List<Text> sublist = new ArrayList<Text>();
-    sublist.add(new Text("b"));
-    expected.add(new Pair<Text, List<Text>>(new Text("a"), sublist));
-
-    assertListEquals(expected, outputs);
-  }
-
-  // shuffle multiple values from the same key.
-  @Test
-  public void testShuffleOneKey() {
-    final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
-    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("b")));
-    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("c")));
-
-    final List<Pair<Text, List<Text>>> outputs = driver2.shuffle(inputs);
-
-    final List<Pair<Text, List<Text>>> expected = new ArrayList<Pair<Text, List<Text>>>();
-    final List<Text> sublist = new ArrayList<Text>();
-    sublist.add(new Text("b"));
-    sublist.add(new Text("c"));
-    expected.add(new Pair<Text, List<Text>>(new Text("a"), sublist));
-
-    assertListEquals(expected, outputs);
-  }
-
-  // shuffle multiple keys
-  @Test
-  public void testMultiShuffle1() {
-    final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
-    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("x")));
-    inputs.add(new Pair<Text, Text>(new Text("b"), new Text("z")));
-    inputs.add(new Pair<Text, Text>(new Text("b"), new Text("w")));
-    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("y")));
-
-    final List<Pair<Text, List<Text>>> outputs = driver2.shuffle(inputs);
-
-    final List<Pair<Text, List<Text>>> expected = new ArrayList<Pair<Text, List<Text>>>();
-    final List<Text> sublist1 = new ArrayList<Text>();
-    sublist1.add(new Text("x"));
-    sublist1.add(new Text("y"));
-    expected.add(new Pair<Text, List<Text>>(new Text("a"), sublist1));
-
-    final List<Text> sublist2 = new ArrayList<Text>();
-    sublist2.add(new Text("z"));
-    sublist2.add(new Text("w"));
-    expected.add(new Pair<Text, List<Text>>(new Text("b"), sublist2));
-
-    assertListEquals(expected, outputs);
-  }
-
-  // shuffle multiple keys that are out-of-order to start.
-  @Test
-  public void testMultiShuffle2() {
-    final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
-    inputs.add(new Pair<Text, Text>(new Text("b"), new Text("z")));
-    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("x")));
-    inputs.add(new Pair<Text, Text>(new Text("b"), new Text("w")));
-    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("y")));
-
-    final List<Pair<Text, List<Text>>> outputs = driver2.shuffle(inputs);
-
-    final List<Pair<Text, List<Text>>> expected = new ArrayList<Pair<Text, List<Text>>>();
-    final List<Text> sublist1 = new ArrayList<Text>();
-    sublist1.add(new Text("x"));
-    sublist1.add(new Text("y"));
-    expected.add(new Pair<Text, List<Text>>(new Text("a"), sublist1));
-
-    final List<Text> sublist2 = new ArrayList<Text>();
-    sublist2.add(new Text("z"));
-    sublist2.add(new Text("w"));
-    expected.add(new Pair<Text, List<Text>>(new Text("b"), sublist2));
-
-    assertListEquals(expected, outputs);
-  }
-
   // Test "combining" with an IdentityReducer. Result should be the same.
   @Test
   public void testIdentityCombiner() throws IOException {
@@ -306,12 +211,13 @@
   @Test
   public void testRepeatRun() throws IOException {
     driver.withCombiner(new IdentityReducer<Text, LongWritable>())
-            .withInput(new Text("foo"), new LongWritable(FOO_IN_A))
-            .withInput(new Text("foo"), new LongWritable(FOO_IN_B))
-            .withInput(new Text("bar"), new LongWritable(BAR_IN))
-            .withOutput(new Text("bar"), new LongWritable(BAR_IN))
-            .withOutput(new Text("foo"), new LongWritable(FOO_OUT)).runTest();
-    thrown.expectMessage(IllegalStateException.class, "Driver reuse not allowed");
+        .withInput(new Text("foo"), new LongWritable(FOO_IN_A))
+        .withInput(new Text("foo"), new LongWritable(FOO_IN_B))
+        .withInput(new Text("bar"), new LongWritable(BAR_IN))
+        .withOutput(new Text("bar"), new LongWritable(BAR_IN))
+        .withOutput(new Text("foo"), new LongWritable(FOO_OUT)).runTest();
+    thrown.expectMessage(IllegalStateException.class,
+        "Driver reuse not allowed");
     driver.runTest();
   }
 
@@ -592,8 +498,7 @@
 
   @Test
   public void testMapInputFile() throws IOException {
-    InputPathStoringMapper<LongWritable,LongWritable> mapper =
-        new InputPathStoringMapper<LongWritable,LongWritable>();
+    InputPathStoringMapper<LongWritable, LongWritable> mapper = new InputPathStoringMapper<LongWritable, LongWritable>();
     Path mapInputPath = new Path("myfile");
     driver = MapReduceDriver.newMapReduceDriver(mapper, reducer);
     driver.setMapInputPath(mapInputPath);
@@ -606,16 +511,15 @@
 
   @Test
   public void testGroupingComparatorBehaviour1() throws IOException {
-    driver.withInput(new Text("A1"),new LongWritable(1L))
-      .withInput(new Text("A2"),new LongWritable(1L))
-      .withInput(new Text("B1"),new LongWritable(1L))
-      .withInput(new Text("B2"),new LongWritable(1L))
-      .withInput(new Text("C1"),new LongWritable(1L))
-      .withOutput(new Text("A1"),new LongWritable(2L))
-      .withOutput(new Text("B1"),new LongWritable(2L))
-      .withOutput(new Text("C1"),new LongWritable(1L))
-      .withKeyGroupingComparator(new FirstCharComparator())
-      .runTest(false);
+    driver.withInput(new Text("A1"), new LongWritable(1L))
+        .withInput(new Text("A2"), new LongWritable(1L))
+        .withInput(new Text("B1"), new LongWritable(1L))
+        .withInput(new Text("B2"), new LongWritable(1L))
+        .withInput(new Text("C1"), new LongWritable(1L))
+        .withOutput(new Text("A1"), new LongWritable(2L))
+        .withOutput(new Text("B1"), new LongWritable(2L))
+        .withOutput(new Text("C1"), new LongWritable(1L))
+        .withKeyGroupingComparator(new FirstCharComparator()).runTest(false);
   }
 
   @Test
@@ -624,34 +528,32 @@
     // grouping of reduce keys in "shuffle".
     // MapReduce doesn't group keys which aren't in a contiguous
     // range when sorted by their sorting comparator.
-    driver.withInput(new Text("1A"),new LongWritable(1L))
-      .withInput(new Text("2A"),new LongWritable(1L))
-      .withInput(new Text("1B"),new LongWritable(1L))
-      .withInput(new Text("2B"),new LongWritable(1L))
-      .withInput(new Text("1C"),new LongWritable(1L))
-      .withOutput(new Text("1A"),new LongWritable(1L))
-      .withOutput(new Text("2A"),new LongWritable(1L))
-      .withOutput(new Text("1B"),new LongWritable(1L))
-      .withOutput(new Text("2B"),new LongWritable(1L))
-      .withOutput(new Text("1C"),new LongWritable(1L))
-      .withKeyGroupingComparator(new SecondCharComparator())
-      .runTest(false);
+    driver.withInput(new Text("1A"), new LongWritable(1L))
+        .withInput(new Text("2A"), new LongWritable(1L))
+        .withInput(new Text("1B"), new LongWritable(1L))
+        .withInput(new Text("2B"), new LongWritable(1L))
+        .withInput(new Text("1C"), new LongWritable(1L))
+        .withOutput(new Text("1A"), new LongWritable(1L))
+        .withOutput(new Text("2A"), new LongWritable(1L))
+        .withOutput(new Text("1B"), new LongWritable(1L))
+        .withOutput(new Text("2B"), new LongWritable(1L))
+        .withOutput(new Text("1C"), new LongWritable(1L))
+        .withKeyGroupingComparator(new SecondCharComparator()).runTest(false);
   }
 
   @Test
   public void testGroupingComparatorSpecifiedByConf() throws IOException {
     JobConf conf = new JobConf(new Configuration());
     conf.setOutputValueGroupingComparator(FirstCharComparator.class);
-    driver.withInput(new Text("A1"),new LongWritable(1L))
-      .withInput(new Text("A2"),new LongWritable(1L))
-      .withInput(new Text("B1"),new LongWritable(1L))
-      .withInput(new Text("B2"),new LongWritable(1L))
-      .withInput(new Text("C1"),new LongWritable(1L))
-      .withOutput(new Text("A1"),new LongWritable(2L))
-      .withOutput(new Text("B1"),new LongWritable(2L))
-      .withOutput(new Text("C1"),new LongWritable(1L))
-      .withConfiguration(conf)
-      .runTest(false);
+    driver.withInput(new Text("A1"), new LongWritable(1L))
+        .withInput(new Text("A2"), new LongWritable(1L))
+        .withInput(new Text("B1"), new LongWritable(1L))
+        .withInput(new Text("B2"), new LongWritable(1L))
+        .withInput(new Text("C1"), new LongWritable(1L))
+        .withOutput(new Text("A1"), new LongWritable(2L))
+        .withOutput(new Text("B1"), new LongWritable(2L))
+        .withOutput(new Text("C1"), new LongWritable(1L))
+        .withConfiguration(conf).runTest(false);
   }
 
   @Test
@@ -660,13 +562,14 @@
         .newMapReduceDriver(new IdentityMapper<TestWritable, Text>(),
             new IdentityReducer<TestWritable, Text>());
     driver.withInput(new TestWritable("A1"), new Text("A1"))
-      .withInput(new TestWritable("A2"), new Text("A2"))
-      .withInput(new TestWritable("A3"), new Text("A3"))
-      .withKeyGroupingComparator(new TestWritable.SingleGroupComparator())
-      .withOutput(new TestWritable("A3"), new Text("A3"))
-      .withOutput(new TestWritable("A3"), new Text("A2"))
-      .withOutput(new TestWritable("A3"), new Text("A1"))
-      .runTest(true); //ordering is important
+        .withInput(new TestWritable("A2"), new Text("A2"))
+        .withInput(new TestWritable("A3"), new Text("A3"))
+        .withKeyGroupingComparator(new TestWritable.SingleGroupComparator())
+        .withOutput(new TestWritable("A3"), new Text("A3"))
+        .withOutput(new TestWritable("A3"), new Text("A2"))
+        .withOutput(new TestWritable("A3"), new Text("A1")).runTest(true); // ordering
+                                                                           // is
+                                                                           // important
   }
 
 }
diff --git a/src/test/java/org/apache/hadoop/mrunit/TestMultipleInputsMapReduceDriver.java b/src/test/java/org/apache/hadoop/mrunit/TestMultipleInputsMapReduceDriver.java
new file mode 100644
index 0000000..0a36ce8
--- /dev/null
+++ b/src/test/java/org/apache/hadoop/mrunit/TestMultipleInputsMapReduceDriver.java
@@ -0,0 +1,675 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mrunit;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.serializer.JavaSerializationComparator;
+import org.apache.hadoop.mapred.*;
+import org.apache.hadoop.mapred.lib.IdentityMapper;
+import org.apache.hadoop.mapred.lib.IdentityReducer;
+import org.apache.hadoop.mapred.lib.LongSumReducer;
+import org.apache.hadoop.mrunit.types.Pair;
+import org.apache.hadoop.mrunit.types.TestWritable;
+import org.apache.hadoop.mrunit.types.UncomparableWritable;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+
+import static org.apache.hadoop.mrunit.ExtendedAssert.assertListEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+public class TestMultipleInputsMapReduceDriver {
+  @Rule
+  public final ExpectedSuppliedException thrown = ExpectedSuppliedException
+      .none();
+
+  private static final int FOO_IN_A = 42;
+  private static final int FOO_IN_B = 10;
+  private static final int TOKEN_IN_A = 1;
+  private static final int TOKEN_IN_B = 2;
+  private static final int BAR_IN = 12;
+  private static final int BAR_OUT = BAR_IN + TOKEN_IN_A + TOKEN_IN_B;
+  private static final int FOO_OUT = FOO_IN_A + FOO_IN_B + TOKEN_IN_A + 2
+      * TOKEN_IN_B;
+  private static final String TOKEN_A = "foo bar";
+  private static final String TOKEN_B = "foo foo bar";
+
+  private Mapper<Text, LongWritable, Text, LongWritable> mapper;
+  private Reducer<Text, LongWritable, Text, LongWritable> reducer;
+  private TokenMapper tokenMapper;
+  private MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable> driver;
+
+  @Before
+  public void setUp() {
+    mapper = new IdentityMapper<Text, LongWritable>();
+    reducer = new LongSumReducer<Text>();
+    tokenMapper = new TokenMapper();
+    driver = new MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable>(
+        reducer);
+    driver.addMapper(mapper);
+    driver.addMapper(tokenMapper);
+  }
+
+  @Test
+  public void testRun() throws IOException {
+    final List<Pair<Text, LongWritable>> out = driver
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A))
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_B))
+        .withInput(mapper, new Text("bar"), new LongWritable(BAR_IN))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_A), new Text(TOKEN_A))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_B), new Text(TOKEN_B))
+        .run();
+
+    final List<Pair<Text, LongWritable>> expected = new ArrayList<Pair<Text, LongWritable>>();
+    expected.add(new Pair<Text, LongWritable>(new Text("bar"),
+        new LongWritable(BAR_OUT)));
+    expected.add(new Pair<Text, LongWritable>(new Text("foo"),
+        new LongWritable(FOO_OUT)));
+
+    assertListEquals(expected, out);
+  }
+
+  @Test
+  public void testUncomparable() throws IOException {
+    MultipleInputsMapReduceDriver<Text, Object, Text, Object> testDriver = MultipleInputsMapReduceDriver
+        .newMultipleInputMapReduceDriver(new IdentityReducer<Text, Object>());
+
+    Mapper<Text, Object, Text, Object> identity = new IdentityMapper<Text, Object>();
+    testDriver.addMapper(identity);
+    Text k1 = new Text("foo");
+    Object v1 = new UncomparableWritable(1);
+    testDriver.withInput(identity, k1, v1);
+
+    ReverseIdentityMapper<Object, Text> reverse = new ReverseIdentityMapper<Object, Text>();
+    testDriver.addMapper(reverse);
+    Text k2 = new Text("bar");
+    Object v2 = new UncomparableWritable(2);
+    testDriver.withInput(reverse, v2, k2);
+
+    testDriver.withOutput(k1, v1).withOutput(k2, v2);
+
+    testDriver.runTest(false);
+  }
+
+  @Test
+  public void testTestRun() throws IOException {
+    driver
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A))
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_B))
+        .withInput(mapper, new Text("bar"), new LongWritable(BAR_IN))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_A), new Text(TOKEN_A))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_B), new Text(TOKEN_B))
+        .withOutput(new Text("bar"), new LongWritable(BAR_OUT))
+        .withOutput(new Text("foo"), new LongWritable(FOO_OUT)).runTest(false);
+  }
+
+  @Test
+  public void testAddAll() throws IOException {
+    final List<Pair<Text, LongWritable>> mapperInputs = new ArrayList<Pair<Text, LongWritable>>();
+    mapperInputs.add(new Pair<Text, LongWritable>(new Text("foo"),
+        new LongWritable(FOO_IN_A)));
+    mapperInputs.add(new Pair<Text, LongWritable>(new Text("foo"),
+        new LongWritable(FOO_IN_B)));
+    mapperInputs.add(new Pair<Text, LongWritable>(new Text("bar"),
+        new LongWritable(BAR_IN)));
+
+    final List<Pair<LongWritable, Text>> tokenMapperInputs = new ArrayList<Pair<LongWritable, Text>>();
+    tokenMapperInputs.add(new Pair<LongWritable, Text>(new LongWritable(
+        TOKEN_IN_A), new Text(TOKEN_A)));
+    tokenMapperInputs.add(new Pair<LongWritable, Text>(new LongWritable(
+        TOKEN_IN_B), new Text(TOKEN_B)));
+
+    final List<Pair<Text, LongWritable>> outputs = new ArrayList<Pair<Text, LongWritable>>();
+    outputs.add(new Pair<Text, LongWritable>(new Text("bar"), new LongWritable(
+        BAR_OUT)));
+    outputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(
+        FOO_OUT)));
+
+    driver.withAll(mapper, mapperInputs)
+        .withAll(tokenMapper, tokenMapperInputs).withAllOutput(outputs)
+        .runTest(false);
+  }
+
+  @Test
+  public void testNoInput() throws IOException {
+    thrown.expectMessage(IllegalStateException.class,
+        "No input was provided for mapper");
+    driver.runTest(false);
+  }
+
+  @Test
+  public void testNoInputForMapper() throws IOException {
+    MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable> testDriver = new MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable>();
+    testDriver.addMapper(mapper);
+    testDriver.addMapper(tokenMapper);
+    testDriver.withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A));
+    thrown.expectMessage(IllegalStateException.class,
+        String.format("No input was provided for mapper %s", tokenMapper));
+    testDriver.runTest(false);
+  }
+
+  @Test
+  public void testNoReducer() throws IOException {
+    MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable> testDriver = new MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable>();
+    testDriver.addMapper(mapper);
+    testDriver.withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A));
+    thrown.expectMessage(IllegalStateException.class,
+        "No reducer class was provided");
+    testDriver.runTest(false);
+  }
+
+  @Test
+  public void testIdentityCombiner() throws IOException {
+    driver
+        .withCombiner(new IdentityReducer<Text, LongWritable>())
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A))
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_B))
+        .withInput(mapper, new Text("bar"), new LongWritable(BAR_IN))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_A), new Text(TOKEN_A))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_B), new Text(TOKEN_B))
+        .withOutput(new Text("foo"), new LongWritable(FOO_OUT))
+        .withOutput(new Text("bar"), new LongWritable(BAR_OUT)).runTest(false);
+  }
+
+  @Test
+  public void testLongSumCombiner() throws IOException {
+    driver
+        .withCombiner(new LongSumReducer<Text>())
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A))
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_B))
+        .withInput(mapper, new Text("bar"), new LongWritable(BAR_IN))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_A), new Text(TOKEN_A))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_B), new Text(TOKEN_B))
+        .withOutput(new Text("foo"), new LongWritable(FOO_OUT))
+        .withOutput(new Text("bar"), new LongWritable(BAR_OUT)).runTest(false);
+  }
+
+  @Test
+  public void testLongSumCombinerAndIdentityReducer() throws IOException {
+    driver
+        .withCombiner(new LongSumReducer<Text>())
+        .withReducer(new IdentityReducer<Text, LongWritable>())
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A))
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_B))
+        .withInput(mapper, new Text("bar"), new LongWritable(BAR_IN))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_A), new Text(TOKEN_A))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_B), new Text(TOKEN_B))
+        .withOutput(new Text("foo"), new LongWritable(FOO_OUT))
+        .withOutput(new Text("bar"), new LongWritable(BAR_OUT)).runTest(false);
+  }
+
+  @Test
+  public void testRepeatRun() throws IOException {
+    driver
+        .withCombiner(new IdentityReducer<Text, LongWritable>())
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A))
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_B))
+        .withInput(mapper, new Text("bar"), new LongWritable(BAR_IN))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_A), new Text(TOKEN_A))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_B), new Text(TOKEN_B))
+        .withOutput(new Text("foo"), new LongWritable(FOO_OUT))
+        .withOutput(new Text("bar"), new LongWritable(BAR_OUT)).runTest(false);
+    thrown.expectMessage(IllegalStateException.class,
+        "Driver reuse not allowed");
+    driver.runTest(false);
+  }
+
+  // Test the key grouping and value ordering comparators
+  @Test
+  public void testComparators() throws IOException {
+    // reducer to track the order of the input values using bit shifting
+    driver.withReducer(new Reducer<Text, LongWritable, Text, LongWritable>() {
+      @Override
+      public void reduce(final Text key, final Iterator<LongWritable> values,
+          final OutputCollector<Text, LongWritable> output,
+          final Reporter reporter) throws IOException {
+        long outputValue = 0;
+        int count = 0;
+        while (values.hasNext()) {
+          outputValue |= (values.next().get() << (count++ * 8));
+        }
+
+        output.collect(key, new LongWritable(outputValue));
+      }
+
+      @Override
+      public void configure(final JobConf job) {
+      }
+
+      @Override
+      public void close() throws IOException {
+      }
+    });
+
+    driver
+        .withKeyGroupingComparator(new TestMapReduceDriver.FirstCharComparator());
+    driver
+        .withKeyOrderComparator(new TestMapReduceDriver.SecondCharComparator());
+
+    driver.addInput(mapper, new Text("a1"), new LongWritable(1));
+    driver.addInput(mapper, new Text("b1"), new LongWritable(1));
+    driver.addInput(mapper, new Text("a3"), new LongWritable(3));
+    driver.addInput(mapper, new Text("a2"), new LongWritable(2));
+
+    driver.addInput(tokenMapper, new LongWritable(1), new Text("c1 d1"));
+
+    driver.addOutput(new Text("a1"), new LongWritable(0x1));
+    driver.addOutput(new Text("b1"), new LongWritable(0x1));
+    driver.addOutput(new Text("a2"), new LongWritable(0x2 | (0x3 << 8)));
+    driver.addOutput(new Text("c1"), new LongWritable(0x1));
+    driver.addOutput(new Text("d1"), new LongWritable(0x1));
+
+    driver.runTest(false);
+  }
+
+  @Test
+  public void testNoMapper() throws IOException {
+    MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable> testDriver = new MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable>();
+    testDriver.withReducer(reducer);
+    thrown.expectMessage(IllegalStateException.class,
+        "No mappers were provided");
+    testDriver.runTest(false);
+  }
+
+  @Test
+  public void testWithCounter() throws IOException {
+    MultipleInputsMapReduceDriver<Text, Text, Text, Text> testDriver = new MultipleInputsMapReduceDriver<Text, Text, Text, Text>();
+    Mapper<Text, Text, Text, Text> mapperWithCounters = new TestMapDriver.MapperWithCounters<Text, Text, Text, Text>();
+    Mapper<Text, Text, Text, Text> tokenMapperWithCounters = new TokenMapperWithCounters();
+    testDriver
+        .withMapper(mapperWithCounters)
+        .withInput(mapperWithCounters, new Text("hie"), new Text("Hi"))
+        .withMapper(tokenMapperWithCounters)
+        .withInput(tokenMapperWithCounters, new Text("bie"),
+            new Text("Goodbye Bye"))
+        .withOutput(new Text("hie"), new Text("Hi"))
+        .withOutput(new Text("bie"), new Text("Goodbye"))
+        .withOutput(new Text("bie"), new Text("Bye"))
+        .withCounter(TestMapDriver.MapperWithCounters.Counters.X, 1)
+        .withCounter(TokenMapperWithCounters.Counters.Y, 2)
+        .withCounter("category", "name", 3)
+        .withReducer(
+            new TestReduceDriver.ReducerWithCounters<Text, Text, Text, Text>())
+        .withCounter(TestReduceDriver.ReducerWithCounters.Counters.COUNT, 2)
+        .withCounter(TestReduceDriver.ReducerWithCounters.Counters.SUM, 3)
+        .withCounter("category", "count", 2).withCounter("category", "sum", 3)
+        .runTest(false);
+  }
+
+  @Test
+  public void testWithCounterAndEnumCounterMissing() throws IOException {
+    MultipleInputsMapReduceDriver<Text, Text, Text, Text> testDriver = new MultipleInputsMapReduceDriver<Text, Text, Text, Text>();
+
+    thrown
+        .expectAssertionErrorMessage("2 Error(s): (Actual counter ("
+            + "\"org.apache.hadoop.mrunit.TestMapDriver$MapperWithCounters$Counters\",\"X\")"
+            + " was not found in expected counters, Actual counter ("
+            + "\"org.apache.hadoop.mrunit.TestMultipleInputsMapReduceDriver$TokenMapperWithCounters$Counters\",\"Y\")"
+            + " was not found in expected counters");
+
+    Mapper<Text, Text, Text, Text> mapperWithCounters = new TestMapDriver.MapperWithCounters<Text, Text, Text, Text>();
+    Mapper<Text, Text, Text, Text> tokenMapperWithCounters = new TokenMapperWithCounters();
+
+    testDriver
+        .withMapper(mapperWithCounters)
+        .withInput(mapperWithCounters, new Text("hie"), new Text("Hi"))
+        .withMapper(tokenMapperWithCounters)
+        .withInput(tokenMapperWithCounters, new Text("bie"),
+            new Text("Goodbye Bye"))
+        .withOutput(new Text("hie"), new Text("Hi"))
+        .withOutput(new Text("bie"), new Text("Goodbye"))
+        .withOutput(new Text("bie"), new Text("Bye"))
+        .withStrictCounterChecking()
+        .withCounter("category", "name", 3)
+        .withReducer(
+            new TestReduceDriver.ReducerWithCounters<Text, Text, Text, Text>())
+        .withCounter(TestReduceDriver.ReducerWithCounters.Counters.COUNT, 2)
+        .withCounter(TestReduceDriver.ReducerWithCounters.Counters.SUM, 3)
+        .withCounter("category", "count", 2).withCounter("category", "sum", 3)
+        .runTest(false);
+  }
+
+  @Test
+  public void testWithCounterAndStringCounterMissing() throws IOException {
+    MultipleInputsMapReduceDriver<Text, Text, Text, Text> testDriver = new MultipleInputsMapReduceDriver<Text, Text, Text, Text>();
+
+    thrown.expectAssertionErrorMessage("1 Error(s): (Actual counter ("
+        + "\"category\",\"name\")" + " was not found in expected counters");
+
+    Mapper<Text, Text, Text, Text> mapperWithCounters = new TestMapDriver.MapperWithCounters<Text, Text, Text, Text>();
+    Mapper<Text, Text, Text, Text> tokenMapperWithCounters = new TokenMapperWithCounters();
+
+    testDriver
+        .withMapper(mapperWithCounters)
+        .withInput(mapperWithCounters, new Text("hie"), new Text("Hi"))
+        .withMapper(tokenMapperWithCounters)
+        .withInput(tokenMapperWithCounters, new Text("bie"),
+            new Text("Goodbye Bye"))
+        .withOutput(new Text("hie"), new Text("Hi"))
+        .withOutput(new Text("bie"), new Text("Goodbye"))
+        .withOutput(new Text("bie"), new Text("Bye"))
+        .withStrictCounterChecking()
+        .withCounter(TestMapDriver.MapperWithCounters.Counters.X, 1)
+        .withCounter(TokenMapperWithCounters.Counters.Y, 2)
+        .withReducer(
+            new TestReduceDriver.ReducerWithCounters<Text, Text, Text, Text>())
+        .withCounter(TestReduceDriver.ReducerWithCounters.Counters.COUNT, 2)
+        .withCounter(TestReduceDriver.ReducerWithCounters.Counters.SUM, 3)
+        .withCounter("category", "count", 2).withCounter("category", "sum", 3)
+        .runTest(false);
+  }
+
+  @Test
+  public void testWithFailedCounter() throws IOException {
+    MultipleInputsMapReduceDriver<Text, Text, Text, Text> testDriver = new MultipleInputsMapReduceDriver<Text, Text, Text, Text>();
+
+    thrown
+        .expectAssertionErrorMessage("3 Error(s): ("
+            + "Counter org.apache.hadoop.mrunit.TestMapDriver.MapperWithCounters.Counters.X has value 1 instead of expected 20, "
+            + "Counter org.apache.hadoop.mrunit.TestMultipleInputsMapReduceDriver.TokenMapperWithCounters.Counters.Y has value 2 instead of expected 30, "
+            + "Counter with category category and name name has value 3 instead of expected 20)");
+
+    Mapper<Text, Text, Text, Text> mapperWithCounters = new TestMapDriver.MapperWithCounters<Text, Text, Text, Text>();
+    Mapper<Text, Text, Text, Text> tokenMapperWithCounters = new TokenMapperWithCounters();
+
+    testDriver
+        .withMapper(mapperWithCounters)
+        .withInput(mapperWithCounters, new Text("hie"), new Text("Hi"))
+        .withMapper(tokenMapperWithCounters)
+        .withInput(tokenMapperWithCounters, new Text("bie"),
+            new Text("Goodbye Bye"))
+        .withOutput(new Text("hie"), new Text("Hi"))
+        .withOutput(new Text("bie"), new Text("Goodbye"))
+        .withOutput(new Text("bie"), new Text("Bye"))
+        .withCounter(TestMapDriver.MapperWithCounters.Counters.X, 20)
+        .withCounter(TokenMapperWithCounters.Counters.Y, 30)
+        .withReducer(
+            new TestReduceDriver.ReducerWithCounters<Text, Text, Text, Text>())
+        .withCounter("category", "name", 20).runTest(false);
+  }
+
+  @Test
+  public void testJavaSerialization() throws IOException {
+    final Configuration conf = new Configuration();
+    conf.setStrings("io.serializations", conf.get("io.serializations"),
+        "org.apache.hadoop.io.serializer.JavaSerialization");
+    final MultipleInputsMapReduceDriver<Integer, IntWritable, Integer, IntWritable> testDriver = MultipleInputsMapReduceDriver
+        .newMultipleInputMapReduceDriver(
+                new IdentityReducer<Integer, IntWritable>())
+        .withConfiguration(conf);
+    Mapper<Integer, IntWritable, Integer, IntWritable> identityMapper = new IdentityMapper<Integer, IntWritable>();
+    Mapper<Integer, IntWritable, Integer, IntWritable> anotherIdentityMapper = new IdentityMapper<Integer, IntWritable>();
+    testDriver.addMapper(identityMapper);
+    testDriver.withInput(identityMapper, 1, new IntWritable(2)).withInput(
+        identityMapper, 2, new IntWritable(3));
+    testDriver.addMapper(anotherIdentityMapper);
+    testDriver.withInput(anotherIdentityMapper, 3, new IntWritable(4))
+        .withInput(anotherIdentityMapper, 4, new IntWritable(5));
+    testDriver
+        .withKeyOrderComparator(new JavaSerializationComparator<Integer>());
+    testDriver
+        .withKeyGroupingComparator(TestMapReduceDriver.INTEGER_COMPARATOR);
+
+    testDriver.withOutput(1, new IntWritable(2))
+        .withOutput(2, new IntWritable(3)).withOutput(3, new IntWritable(4))
+        .withOutput(4, new IntWritable(5));
+    testDriver.runTest(false);
+  }
+
+  @Test
+  public void testCopy() throws IOException {
+    final Text key = new Text("a");
+    final LongWritable value = new LongWritable(1);
+    driver.addInput(mapper, key, value);
+    key.set("b");
+    value.set(2);
+    driver.addInput(mapper, key, value);
+
+    key.set("a");
+    value.set(1);
+    driver.addOutput(key, value);
+    key.set("b");
+    value.set(2);
+    driver.addOutput(key, value);
+
+    final LongWritable longKey = new LongWritable(3);
+    final Text textValue = new Text("c d");
+    driver.addInput(tokenMapper, longKey, textValue);
+    longKey.set(4);
+    textValue.set("e f g");
+    driver.addInput(tokenMapper, longKey, textValue);
+
+    key.set("c");
+    value.set(3);
+    driver.addOutput(key, value);
+    key.set("d");
+    value.set(3);
+    driver.addOutput(key, value);
+    key.set("e");
+    value.set(4);
+    driver.addOutput(key, value);
+    key.set("f");
+    value.set(4);
+    driver.addOutput(key, value);
+    key.set("g");
+    value.set(4);
+    driver.addOutput(key, value);
+
+    driver.runTest(false);
+  }
+
+  @Test
+  public void testOutputFormat() throws IOException {
+    driver.withInputFormat(SequenceFileInputFormat.class);
+    driver.withOutputFormat(SequenceFileOutputFormat.class);
+    driver.withInput(mapper, new Text("a"), new LongWritable(1));
+    driver.withInput(mapper, new Text("a"), new LongWritable(2));
+    driver.withInput(tokenMapper, new LongWritable(3), new Text("a b"));
+    driver.withOutput(new Text("a"), new LongWritable(6));
+    driver.withOutput(new Text("b"), new LongWritable(3));
+    driver.runTest(false);
+  }
+
+  @SuppressWarnings({ "unchecked", "rawtypes" })
+  @Test
+  public void testOutputFormatWithMismatchInOutputClasses() throws IOException {
+    final MultipleInputsMapReduceDriver testDriver = this.driver;
+    testDriver.withInputFormat(TextInputFormat.class);
+    testDriver.withOutputFormat(TextOutputFormat.class);
+    testDriver.withInput(mapper, new Text("a"), new LongWritable(1));
+    testDriver.withInput(mapper, new Text("a"), new LongWritable(2));
+    testDriver.withInput(tokenMapper, new LongWritable(3), new Text("a b"));
+    testDriver.withOutput(new LongWritable(0), new Text("a\t6"));
+    testDriver.withOutput(new LongWritable(4), new Text("b\t3"));
+    testDriver.runTest(false);
+  }
+
+  @Test
+  public void testMapInputFile() throws IOException {
+    MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable> testDriver = new MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable>(
+        reducer);
+
+    InputPathStoringMapper<LongWritable, LongWritable> inputPathStoringMapper = new InputPathStoringMapper<LongWritable, LongWritable>();
+    Path mapInputPath = new Path("myfile");
+    testDriver.addMapper(inputPathStoringMapper);
+    testDriver.setMapInputPath(inputPathStoringMapper, mapInputPath);
+    assertEquals(mapInputPath.getName(),
+        testDriver.getMapInputPath(inputPathStoringMapper).getName());
+    testDriver.withInput(inputPathStoringMapper, new Text("a"),
+        new LongWritable(1));
+
+    InputPathStoringMapper<LongWritable, LongWritable> anotherInputPathStoringMapper = new InputPathStoringMapper<LongWritable, LongWritable>();
+    Path anotherMapInputPath = new Path("myotherfile");
+    testDriver.addMapper(anotherInputPathStoringMapper);
+    testDriver.setMapInputPath(anotherInputPathStoringMapper,
+        anotherMapInputPath);
+    assertEquals(anotherMapInputPath.getName(),
+        testDriver.getMapInputPath(anotherInputPathStoringMapper).getName());
+    testDriver.withInput(anotherInputPathStoringMapper, new Text("b"),
+        new LongWritable(2));
+
+    testDriver.runTest(false);
+    assertNotNull(inputPathStoringMapper.getMapInputPath());
+    assertEquals(mapInputPath.getName(), inputPathStoringMapper
+        .getMapInputPath().getName());
+  }
+
+  @Test
+  public void testGroupComparatorBehaviorFirst() throws IOException {
+    driver
+        .withInput(mapper, new Text("A1"), new LongWritable(1L))
+        .withInput(mapper, new Text("A2"), new LongWritable(1L))
+        .withInput(mapper, new Text("B1"), new LongWritable(1L))
+        .withInput(mapper, new Text("B2"), new LongWritable(1L))
+        .withInput(mapper, new Text("C1"), new LongWritable(1L))
+        .withInput(tokenMapper, new LongWritable(3L), new Text("D1 D2 E1"))
+        .withOutput(new Text("A1"), new LongWritable(2L))
+        .withOutput(new Text("B1"), new LongWritable(2L))
+        .withOutput(new Text("C1"), new LongWritable(1L))
+        .withOutput(new Text("D1"), new LongWritable(6L))
+        .withOutput(new Text("E1"), new LongWritable(3L))
+        .withKeyGroupingComparator(
+            new TestMapReduceDriver.FirstCharComparator()).runTest(false);
+  }
+
+  @Test
+  public void testGroupComparatorBehaviorSecond() throws IOException {
+    driver
+        .withInput(mapper, new Text("1A"), new LongWritable(1L))
+        .withInput(mapper, new Text("2A"), new LongWritable(1L))
+        .withInput(mapper, new Text("1B"), new LongWritable(1L))
+        .withInput(mapper, new Text("2B"), new LongWritable(1L))
+        .withInput(mapper, new Text("1C"), new LongWritable(1L))
+        .withInput(tokenMapper, new LongWritable(2L), new Text("1D 2D 1E"))
+        .withOutput(new Text("1A"), new LongWritable(1L))
+        .withOutput(new Text("2A"), new LongWritable(1L))
+        .withOutput(new Text("1B"), new LongWritable(1L))
+        .withOutput(new Text("2B"), new LongWritable(1L))
+        .withOutput(new Text("1C"), new LongWritable(1L))
+        .withOutput(new Text("1D"), new LongWritable(2L))
+        .withOutput(new Text("2D"), new LongWritable(2L))
+        .withOutput(new Text("1E"), new LongWritable(2L))
+        .withKeyGroupingComparator(
+            new TestMapReduceDriver.SecondCharComparator()).runTest(false);
+  }
+
+  @Test
+  public void testGroupingComparatorSpecifiedByConf() throws IOException {
+    JobConf conf = new JobConf(new Configuration());
+    conf.setOutputValueGroupingComparator(TestMapReduceDriver.FirstCharComparator.class);
+    driver.withInput(mapper, new Text("A1"), new LongWritable(1L))
+        .withInput(mapper, new Text("A2"), new LongWritable(1L))
+        .withInput(mapper, new Text("B1"), new LongWritable(1L))
+        .withInput(mapper, new Text("B2"), new LongWritable(1L))
+        .withInput(mapper, new Text("C1"), new LongWritable(1L))
+        .withInput(tokenMapper, new LongWritable(3L), new Text("D1 D2 E1"))
+        .withOutput(new Text("A1"), new LongWritable(2L))
+        .withOutput(new Text("B1"), new LongWritable(2L))
+        .withOutput(new Text("C1"), new LongWritable(1L))
+        .withOutput(new Text("D1"), new LongWritable(6L))
+        .withOutput(new Text("E1"), new LongWritable(3L))
+        .withConfiguration(conf).runTest(false);
+  }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  public void testUseOfWritableRegisteredComparator() throws IOException {
+    MultipleInputsMapReduceDriver<TestWritable, Text, TestWritable, Text> testDriver = new MultipleInputsMapReduceDriver<TestWritable, Text, TestWritable, Text>(
+        new IdentityReducer<TestWritable, Text>());
+
+    IdentityMapper<TestWritable, Text> identityMapper = new IdentityMapper<TestWritable, Text>();
+    IdentityMapper<TestWritable, Text> anotherIdentityMapper = new IdentityMapper<TestWritable, Text>();
+    testDriver.addMapper(identityMapper);
+    testDriver.addMapper(anotherIdentityMapper);
+    testDriver
+        .withInput(identityMapper, new TestWritable("A1"), new Text("A1"))
+        .withInput(identityMapper, new TestWritable("A2"), new Text("A2"))
+        .withInput(identityMapper, new TestWritable("A3"), new Text("A3"))
+        .withInput(anotherIdentityMapper, new TestWritable("B1"),
+            new Text("B1"))
+        .withInput(anotherIdentityMapper, new TestWritable("B2"),
+            new Text("B2"))
+        .withKeyGroupingComparator(new TestWritable.SingleGroupComparator())
+        .withOutput(new TestWritable("B2"), new Text("B2"))
+        .withOutput(new TestWritable("B2"), new Text("B1"))
+        .withOutput(new TestWritable("B2"), new Text("A3"))
+        .withOutput(new TestWritable("B2"), new Text("A2"))
+        .withOutput(new TestWritable("B2"), new Text("A1")).runTest(true); // ordering
+                                                                           // is
+                                                                           // important
+  }
+
+  static class TokenMapperWithCounters extends MapReduceBase implements
+      Mapper<Text, Text, Text, Text> {
+    private final Text output = new Text();
+
+    @Override
+    public void map(Text key, Text value,
+        OutputCollector<Text, Text> collector, Reporter reporter)
+        throws IOException {
+      String[] tokens = value.toString().split("\\s");
+      for (String token : tokens) {
+        output.set(token);
+        collector.collect(key, output);
+        reporter.getCounter(Counters.Y).increment(1);
+        reporter.getCounter("category", "name").increment(1);
+      }
+    }
+
+    public static enum Counters {
+      Y
+    }
+  }
+
+  static class TokenMapper extends MapReduceBase implements
+      Mapper<LongWritable, Text, Text, LongWritable> {
+    private final Text output = new Text();
+
+    @Override
+    public void map(LongWritable longWritable, Text text,
+        OutputCollector<Text, LongWritable> textLongWritableOutputCollector,
+        Reporter reporter) throws IOException {
+      String[] tokens = text.toString().split("\\s");
+      for (String token : tokens) {
+        output.set(token);
+        textLongWritableOutputCollector.collect(output, longWritable);
+      }
+    }
+  }
+
+  static class ReverseIdentityMapper<KEYIN, VALUEIN> extends MapReduceBase
+      implements Mapper<KEYIN, VALUEIN, VALUEIN, KEYIN> {
+    @Override
+    public void map(KEYIN key, VALUEIN value,
+        OutputCollector<VALUEIN, KEYIN> vkOutputCollector, Reporter reporter)
+        throws IOException {
+      vkOutputCollector.collect(value, key);
+    }
+  }
+}
diff --git a/src/test/java/org/apache/hadoop/mrunit/mapreduce/TestMapReduceDriver.java b/src/test/java/org/apache/hadoop/mrunit/mapreduce/TestMapReduceDriver.java
index 0f6a521..bbd0c82 100644
--- a/src/test/java/org/apache/hadoop/mrunit/mapreduce/TestMapReduceDriver.java
+++ b/src/test/java/org/apache/hadoop/mrunit/mapreduce/TestMapReduceDriver.java
@@ -17,13 +17,6 @@
  */
 package org.apache.hadoop.mrunit.mapreduce;
 
-import static org.apache.hadoop.mrunit.ExtendedAssert.assertListEquals;
-import static org.junit.Assert.*;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.List;
-
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.io.IntWritable;
@@ -31,8 +24,6 @@
 import org.apache.hadoop.io.NullWritable;
 import org.apache.hadoop.io.Text;
 import org.apache.hadoop.io.serializer.JavaSerializationComparator;
-import org.apache.hadoop.mapred.lib.IdentityMapper;
-import org.apache.hadoop.mapred.lib.IdentityReducer;
 import org.apache.hadoop.mapreduce.Mapper;
 import org.apache.hadoop.mapreduce.Reducer;
 import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
@@ -43,7 +34,6 @@
 import org.apache.hadoop.mapreduce.lib.reduce.IntSumReducer;
 import org.apache.hadoop.mapreduce.lib.reduce.LongSumReducer;
 import org.apache.hadoop.mrunit.ExpectedSuppliedException;
-import org.apache.hadoop.mrunit.mapreduce.MapReduceDriver;
 import org.apache.hadoop.mrunit.TestMapReduceDriver.FirstCharComparator;
 import org.apache.hadoop.mrunit.TestMapReduceDriver.SecondCharComparator;
 import org.apache.hadoop.mrunit.mapreduce.TestMapDriver.ConfigurationMapper;
@@ -56,6 +46,14 @@
 import org.junit.Rule;
 import org.junit.Test;
 
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.hadoop.mrunit.ExtendedAssert.assertListEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
 public class TestMapReduceDriver {
 
   private static final int FOO_IN_A = 42;
@@ -103,10 +101,10 @@
   public void testUncomparable() throws IOException {
     Text k = new Text("test");
     Object v = new UncomparableWritable(2);
-    MapReduceDriver.newMapReduceDriver(
-        new Mapper<Text, Object,Text, Object>(),
-        new Reducer<Text, Object,Text, Object>())
-        .withInput(k, v).withOutput(k, v).runTest();
+    MapReduceDriver
+        .newMapReduceDriver(new Mapper<Text, Object, Text, Object>(),
+            new Reducer<Text, Object, Text, Object>()).withInput(k, v)
+        .withOutput(k, v).runTest();
   }
 
   @Test
@@ -130,8 +128,10 @@
   @Test
   public void testTestRun3() throws IOException {
     thrown.expectAssertionErrorMessage("2 Error(s)");
-    thrown.expectAssertionErrorMessage("Missing expected output (foo, 52) at position 0, got (bar, 12).");
-    thrown.expectAssertionErrorMessage("Missing expected output (bar, 12) at position 1, got (foo, 52).");
+    thrown
+        .expectAssertionErrorMessage("Missing expected output (foo, 52) at position 0, got (bar, 12).");
+    thrown
+        .expectAssertionErrorMessage("Missing expected output (bar, 12) at position 1, got (foo, 52).");
     driver.withInput(new Text("foo"), new LongWritable(FOO_IN_A))
         .withInput(new Text("bar"), new LongWritable(BAR_IN))
         .withInput(new Text("foo"), new LongWritable(FOO_IN_B))
@@ -142,13 +142,18 @@
   @Test
   public void testAddAll() throws IOException {
     final List<Pair<Text, LongWritable>> inputs = new ArrayList<Pair<Text, LongWritable>>();
-    inputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(FOO_IN_A)));
-    inputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(FOO_IN_B)));
-    inputs.add(new Pair<Text, LongWritable>(new Text("bar"), new LongWritable(BAR_IN)));
+    inputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(
+        FOO_IN_A)));
+    inputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(
+        FOO_IN_B)));
+    inputs.add(new Pair<Text, LongWritable>(new Text("bar"), new LongWritable(
+        BAR_IN)));
 
     final List<Pair<Text, LongWritable>> outputs = new ArrayList<Pair<Text, LongWritable>>();
-    outputs.add(new Pair<Text, LongWritable>(new Text("bar"), new LongWritable(BAR_IN)));
-    outputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(FOO_OUT)));
+    outputs.add(new Pair<Text, LongWritable>(new Text("bar"), new LongWritable(
+        BAR_IN)));
+    outputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(
+        FOO_OUT)));
 
     driver.withAll(inputs).withAllOutput(outputs).runTest();
   }
@@ -170,100 +175,10 @@
   }
 
   @Test
-  public void testEmptyShuffle() {
-    final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
-    final List<Pair<Text, List<Text>>> outputs = driver2.shuffle(inputs);
-    assertEquals(0, outputs.size());
-  }
-
-  // just shuffle a single (k, v) pair
-  @Test
-  public void testSingleShuffle() {
-    final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
-    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("b")));
-
-    final List<Pair<Text, List<Text>>> outputs = driver2.shuffle(inputs);
-
-    final List<Pair<Text, List<Text>>> expected = new ArrayList<Pair<Text, List<Text>>>();
-    final List<Text> sublist = new ArrayList<Text>();
-    sublist.add(new Text("b"));
-    expected.add(new Pair<Text, List<Text>>(new Text("a"), sublist));
-
-    assertListEquals(expected, outputs);
-  }
-
-  // shuffle multiple values from the same key.
-  @Test
-  public void testShuffleOneKey() {
-    final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
-    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("b")));
-    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("c")));
-
-    final List<Pair<Text, List<Text>>> outputs = driver2.shuffle(inputs);
-
-    final List<Pair<Text, List<Text>>> expected = new ArrayList<Pair<Text, List<Text>>>();
-    final List<Text> sublist = new ArrayList<Text>();
-    sublist.add(new Text("b"));
-    sublist.add(new Text("c"));
-    expected.add(new Pair<Text, List<Text>>(new Text("a"), sublist));
-
-    assertListEquals(expected, outputs);
-  }
-
-  // shuffle multiple keys
-  @Test
-  public void testMultiShuffle1() {
-    final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
-    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("x")));
-    inputs.add(new Pair<Text, Text>(new Text("b"), new Text("z")));
-    inputs.add(new Pair<Text, Text>(new Text("b"), new Text("w")));
-    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("y")));
-
-    final List<Pair<Text, List<Text>>> outputs = driver2.shuffle(inputs);
-
-    final List<Pair<Text, List<Text>>> expected = new ArrayList<Pair<Text, List<Text>>>();
-    final List<Text> sublist1 = new ArrayList<Text>();
-    sublist1.add(new Text("x"));
-    sublist1.add(new Text("y"));
-    expected.add(new Pair<Text, List<Text>>(new Text("a"), sublist1));
-
-    final List<Text> sublist2 = new ArrayList<Text>();
-    sublist2.add(new Text("z"));
-    sublist2.add(new Text("w"));
-    expected.add(new Pair<Text, List<Text>>(new Text("b"), sublist2));
-
-    assertListEquals(expected, outputs);
-  }
-
-  // shuffle multiple keys that are out-of-order to start.
-  @Test
-  public void testMultiShuffle2() {
-    final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
-    inputs.add(new Pair<Text, Text>(new Text("b"), new Text("z")));
-    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("x")));
-    inputs.add(new Pair<Text, Text>(new Text("b"), new Text("w")));
-    inputs.add(new Pair<Text, Text>(new Text("a"), new Text("y")));
-
-    final List<Pair<Text, List<Text>>> outputs = driver2.shuffle(inputs);
-
-    final List<Pair<Text, List<Text>>> expected = new ArrayList<Pair<Text, List<Text>>>();
-    final List<Text> sublist1 = new ArrayList<Text>();
-    sublist1.add(new Text("x"));
-    sublist1.add(new Text("y"));
-    expected.add(new Pair<Text, List<Text>>(new Text("a"), sublist1));
-
-    final List<Text> sublist2 = new ArrayList<Text>();
-    sublist2.add(new Text("z"));
-    sublist2.add(new Text("w"));
-    expected.add(new Pair<Text, List<Text>>(new Text("b"), sublist2));
-
-    assertListEquals(expected, outputs);
-  }
-
-  @Test
   public void testEmptySortAndGroup() {
     final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
-    final List<KeyValueReuseList<Text, Text>> outputs = driver2.sortAndGroup(inputs);
+    final List<KeyValueReuseList<Text, Text>> outputs = driver2
+        .sortAndGroup(inputs);
     assertEquals(0, outputs.size());
   }
 
@@ -273,10 +188,12 @@
     final List<Pair<Text, Text>> inputs = new ArrayList<Pair<Text, Text>>();
     inputs.add(new Pair<Text, Text>(new Text("a"), new Text("b")));
 
-    final List<KeyValueReuseList<Text, Text>> outputs = driver2.sortAndGroup(inputs);
+    final List<KeyValueReuseList<Text, Text>> outputs = driver2
+        .sortAndGroup(inputs);
 
     final List<KeyValueReuseList<Text, Text>> expected = new ArrayList<KeyValueReuseList<Text, Text>>();
-    final KeyValueReuseList<Text, Text> sublist = new KeyValueReuseList<Text, Text>(new Text(), new Text(), driver2.getConfiguration());
+    final KeyValueReuseList<Text, Text> sublist = new KeyValueReuseList<Text, Text>(
+        new Text(), new Text(), driver2.getConfiguration());
     sublist.add(new Pair<Text, Text>(new Text("a"), new Text("b")));
     expected.add(sublist);
 
@@ -290,10 +207,12 @@
     inputs.add(new Pair<Text, Text>(new Text("a"), new Text("b")));
     inputs.add(new Pair<Text, Text>(new Text("a"), new Text("c")));
 
-    final List<KeyValueReuseList<Text, Text>> outputs = driver2.sortAndGroup(inputs);
+    final List<KeyValueReuseList<Text, Text>> outputs = driver2
+        .sortAndGroup(inputs);
 
     final List<KeyValueReuseList<Text, Text>> expected = new ArrayList<KeyValueReuseList<Text, Text>>();
-    final KeyValueReuseList<Text, Text> sublist = new KeyValueReuseList<Text, Text>(new Text(), new Text(), driver2.getConfiguration());
+    final KeyValueReuseList<Text, Text> sublist = new KeyValueReuseList<Text, Text>(
+        new Text(), new Text(), driver2.getConfiguration());
     sublist.add(new Pair<Text, Text>(new Text("a"), new Text("b")));
     sublist.add(new Pair<Text, Text>(new Text("a"), new Text("c")));
     expected.add(sublist);
@@ -310,15 +229,18 @@
     inputs.add(new Pair<Text, Text>(new Text("b"), new Text("w")));
     inputs.add(new Pair<Text, Text>(new Text("a"), new Text("y")));
 
-    final List<KeyValueReuseList<Text, Text>> outputs = driver2.sortAndGroup(inputs);
+    final List<KeyValueReuseList<Text, Text>> outputs = driver2
+        .sortAndGroup(inputs);
 
     final List<KeyValueReuseList<Text, Text>> expected = new ArrayList<KeyValueReuseList<Text, Text>>();
-    final KeyValueReuseList<Text, Text> sublist1 = new KeyValueReuseList<Text, Text>(new Text(), new Text(), driver2.getConfiguration());
+    final KeyValueReuseList<Text, Text> sublist1 = new KeyValueReuseList<Text, Text>(
+        new Text(), new Text(), driver2.getConfiguration());
     sublist1.add(new Pair<Text, Text>(new Text("a"), new Text("x")));
     sublist1.add(new Pair<Text, Text>(new Text("a"), new Text("y")));
     expected.add(sublist1);
 
-    final KeyValueReuseList<Text, Text> sublist2 = new KeyValueReuseList<Text, Text>(new Text(), new Text(), driver2.getConfiguration());
+    final KeyValueReuseList<Text, Text> sublist2 = new KeyValueReuseList<Text, Text>(
+        new Text(), new Text(), driver2.getConfiguration());
     sublist2.add(new Pair<Text, Text>(new Text("b"), new Text("z")));
     sublist2.add(new Pair<Text, Text>(new Text("b"), new Text("w")));
     expected.add(sublist2);
@@ -335,15 +257,18 @@
     inputs.add(new Pair<Text, Text>(new Text("b"), new Text("w")));
     inputs.add(new Pair<Text, Text>(new Text("a"), new Text("y")));
 
-    final List<KeyValueReuseList<Text, Text>> outputs = driver2.sortAndGroup(inputs);
+    final List<KeyValueReuseList<Text, Text>> outputs = driver2
+        .sortAndGroup(inputs);
 
     final List<KeyValueReuseList<Text, Text>> expected = new ArrayList<KeyValueReuseList<Text, Text>>();
-    final KeyValueReuseList<Text, Text> sublist1 = new KeyValueReuseList<Text, Text>(new Text(), new Text(), driver2.getConfiguration());
+    final KeyValueReuseList<Text, Text> sublist1 = new KeyValueReuseList<Text, Text>(
+        new Text(), new Text(), driver2.getConfiguration());
     sublist1.add(new Pair<Text, Text>(new Text("a"), new Text("x")));
     sublist1.add(new Pair<Text, Text>(new Text("a"), new Text("y")));
     expected.add(sublist1);
 
-    final KeyValueReuseList<Text, Text> sublist2 = new KeyValueReuseList<Text, Text>(new Text(), new Text(), driver2.getConfiguration());
+    final KeyValueReuseList<Text, Text> sublist2 = new KeyValueReuseList<Text, Text>(
+        new Text(), new Text(), driver2.getConfiguration());
     sublist2.add(new Pair<Text, Text>(new Text("b"), new Text("z")));
     sublist2.add(new Pair<Text, Text>(new Text("b"), new Text("w")));
     expected.add(sublist2);
@@ -614,16 +539,15 @@
 
   @Test
   public void testGroupingComparatorBehaviour1() throws IOException {
-    driver.withInput(new Text("A1"),new LongWritable(1L))
-      .withInput(new Text("A2"),new LongWritable(1L))
-      .withInput(new Text("B1"),new LongWritable(1L))
-      .withInput(new Text("B2"),new LongWritable(1L))
-      .withInput(new Text("C1"),new LongWritable(1L))
-      .withOutput(new Text("A2"),new LongWritable(2L))
-      .withOutput(new Text("B2"),new LongWritable(2L))
-      .withOutput(new Text("C1"),new LongWritable(1L))
-      .withKeyGroupingComparator(new FirstCharComparator())
-      .runTest(false);
+    driver.withInput(new Text("A1"), new LongWritable(1L))
+        .withInput(new Text("A2"), new LongWritable(1L))
+        .withInput(new Text("B1"), new LongWritable(1L))
+        .withInput(new Text("B2"), new LongWritable(1L))
+        .withInput(new Text("C1"), new LongWritable(1L))
+        .withOutput(new Text("A2"), new LongWritable(2L))
+        .withOutput(new Text("B2"), new LongWritable(2L))
+        .withOutput(new Text("C1"), new LongWritable(1L))
+        .withKeyGroupingComparator(new FirstCharComparator()).runTest(false);
   }
 
   @Test
@@ -632,18 +556,17 @@
     // grouping of reduce keys in "shuffle".
     // MapReduce doesn't group keys which aren't in a contiguous
     // range when sorted by their sorting comparator.
-    driver.withInput(new Text("1A"),new LongWritable(1L))
-      .withInput(new Text("2A"),new LongWritable(1L))
-      .withInput(new Text("1B"),new LongWritable(1L))
-      .withInput(new Text("2B"),new LongWritable(1L))
-      .withInput(new Text("1C"),new LongWritable(1L))
-      .withOutput(new Text("1A"),new LongWritable(1L))
-      .withOutput(new Text("2A"),new LongWritable(1L))
-      .withOutput(new Text("1B"),new LongWritable(1L))
-      .withOutput(new Text("2B"),new LongWritable(1L))
-      .withOutput(new Text("1C"),new LongWritable(1L))
-      .withKeyGroupingComparator(new SecondCharComparator())
-      .runTest(false);
+    driver.withInput(new Text("1A"), new LongWritable(1L))
+        .withInput(new Text("2A"), new LongWritable(1L))
+        .withInput(new Text("1B"), new LongWritable(1L))
+        .withInput(new Text("2B"), new LongWritable(1L))
+        .withInput(new Text("1C"), new LongWritable(1L))
+        .withOutput(new Text("1A"), new LongWritable(1L))
+        .withOutput(new Text("2A"), new LongWritable(1L))
+        .withOutput(new Text("1B"), new LongWritable(1L))
+        .withOutput(new Text("2B"), new LongWritable(1L))
+        .withOutput(new Text("1C"), new LongWritable(1L))
+        .withKeyGroupingComparator(new SecondCharComparator()).runTest(false);
   }
 
   @Test
@@ -651,33 +574,36 @@
 
     // this test should use the comparator registered inside TestWritable
     // to output the keys in reverse order
-    MapReduceDriver<TestWritable,Text,TestWritable,Text,TestWritable,Text> driver
-      = MapReduceDriver.newMapReduceDriver(new Mapper(), new Reducer());
+    MapReduceDriver<TestWritable, Text, TestWritable, Text, TestWritable, Text> driver = MapReduceDriver
+        .newMapReduceDriver(new Mapper(), new Reducer());
 
-    driver.withInput(new TestWritable("A1"), new Text("A1"))
-      .withInput(new TestWritable("A2"), new Text("A2"))
-      .withInput(new TestWritable("A3"), new Text("A3"))
-      .withKeyGroupingComparator(new TestWritable.SingleGroupComparator())
-      // these output keys are incorrect because of MRUNIT-129
-      //.withOutput(new TestWritable("A3"), new Text("A3"))
-      //.withOutput(new TestWritable("A3"), new Text("A2"))
-      //.withOutput(new TestWritable("A3"), new Text("A1"))
-      //the following are the actual correct outputs
-      .withOutput(new TestWritable("A3"), new Text("A3"))
-      .withOutput(new TestWritable("A2"), new Text("A2"))
-      .withOutput(new TestWritable("A1"), new Text("A1"))
-      .runTest(true); //ordering is important
+    driver
+        .withInput(new TestWritable("A1"), new Text("A1"))
+        .withInput(new TestWritable("A2"), new Text("A2"))
+        .withInput(new TestWritable("A3"), new Text("A3"))
+        .withKeyGroupingComparator(new TestWritable.SingleGroupComparator())
+        // these output keys are incorrect because of MRUNIT-129
+        // .withOutput(new TestWritable("A3"), new Text("A3"))
+        // .withOutput(new TestWritable("A3"), new Text("A2"))
+        // .withOutput(new TestWritable("A3"), new Text("A1"))
+        // the following are the actual correct outputs
+        .withOutput(new TestWritable("A3"), new Text("A3"))
+        .withOutput(new TestWritable("A2"), new Text("A2"))
+        .withOutput(new TestWritable("A1"), new Text("A1")).runTest(true); // ordering
+                                                                           // is
+                                                                           // important
   }
 
   @Test
   public void testRepeatRun() throws IOException {
     driver.withCombiner(new Reducer<Text, LongWritable, Text, LongWritable>())
-            .withInput(new Text("foo"), new LongWritable(FOO_IN_A))
-            .withInput(new Text("foo"), new LongWritable(FOO_IN_B))
-            .withInput(new Text("bar"), new LongWritable(BAR_IN))
-            .withOutput(new Text("bar"), new LongWritable(BAR_IN))
-            .withOutput(new Text("foo"), new LongWritable(FOO_OUT)).runTest();
-    thrown.expectMessage(IllegalStateException.class, "Driver reuse not allowed");
+        .withInput(new Text("foo"), new LongWritable(FOO_IN_A))
+        .withInput(new Text("foo"), new LongWritable(FOO_IN_B))
+        .withInput(new Text("bar"), new LongWritable(BAR_IN))
+        .withOutput(new Text("bar"), new LongWritable(BAR_IN))
+        .withOutput(new Text("foo"), new LongWritable(FOO_OUT)).runTest();
+    thrown.expectMessage(IllegalStateException.class,
+        "Driver reuse not allowed");
     driver.runTest();
   }
 
diff --git a/src/test/java/org/apache/hadoop/mrunit/mapreduce/TestMultipleInputsMapReduceDriver.java b/src/test/java/org/apache/hadoop/mrunit/mapreduce/TestMultipleInputsMapReduceDriver.java
new file mode 100644
index 0000000..b6a7b86
--- /dev/null
+++ b/src/test/java/org/apache/hadoop/mrunit/mapreduce/TestMultipleInputsMapReduceDriver.java
@@ -0,0 +1,658 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.mrunit.mapreduce;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.LongWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.io.serializer.JavaSerializationComparator;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.hadoop.mapreduce.lib.input.SequenceFileInputFormat;
+import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
+import org.apache.hadoop.mapreduce.lib.output.SequenceFileOutputFormat;
+import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
+import org.apache.hadoop.mapreduce.lib.reduce.LongSumReducer;
+import org.apache.hadoop.mrunit.ExpectedSuppliedException;
+import org.apache.hadoop.mrunit.TestMapReduceDriver;
+import org.apache.hadoop.mrunit.types.Pair;
+import org.apache.hadoop.mrunit.types.TestWritable;
+import org.apache.hadoop.mrunit.types.UncomparableWritable;
+import org.junit.Before;
+import org.junit.Rule;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+
+import static org.apache.hadoop.mrunit.ExtendedAssert.assertListEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNotNull;
+
+public class TestMultipleInputsMapReduceDriver {
+  @Rule
+  public final ExpectedSuppliedException thrown = ExpectedSuppliedException
+      .none();
+
+  private static final int FOO_IN_A = 42;
+  private static final int FOO_IN_B = 10;
+  private static final int TOKEN_IN_A = 1;
+  private static final int TOKEN_IN_B = 2;
+  private static final int BAR_IN = 12;
+  private static final int BAR_OUT = BAR_IN + TOKEN_IN_A + TOKEN_IN_B;
+  private static final int FOO_OUT = FOO_IN_A + FOO_IN_B + TOKEN_IN_A + 2
+      * TOKEN_IN_B;
+  private static final String TOKEN_A = "foo bar";
+  private static final String TOKEN_B = "foo foo bar";
+
+  private Mapper<Text, LongWritable, Text, LongWritable> mapper;
+  private Reducer<Text, LongWritable, Text, LongWritable> reducer;
+  private TokenMapper tokenMapper;
+  private MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable> driver;
+
+  @Before
+  public void setUp() {
+    mapper = new Mapper<Text, LongWritable, Text, LongWritable>();
+    reducer = new LongSumReducer<Text>();
+    tokenMapper = new TokenMapper();
+    driver = new MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable>(
+        reducer);
+    driver.addMapper(mapper);
+    driver.addMapper(tokenMapper);
+  }
+
+  @Test
+  public void testRun() throws IOException {
+    final List<Pair<Text, LongWritable>> out = driver
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A))
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_B))
+        .withInput(mapper, new Text("bar"), new LongWritable(BAR_IN))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_A), new Text(TOKEN_A))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_B), new Text(TOKEN_B))
+        .run();
+
+    final List<Pair<Text, LongWritable>> expected = new ArrayList<Pair<Text, LongWritable>>();
+    expected.add(new Pair<Text, LongWritable>(new Text("bar"),
+        new LongWritable(BAR_OUT)));
+    expected.add(new Pair<Text, LongWritable>(new Text("foo"),
+        new LongWritable(FOO_OUT)));
+
+    assertListEquals(expected, out);
+  }
+
+  @Test
+  public void testUncomparable() throws IOException {
+    MultipleInputsMapReduceDriver<Text, Object, Text, Object> testDriver = MultipleInputsMapReduceDriver
+        .newMultipleInputMapReduceDriver(new Reducer<Text, Object, Text, Object>());
+
+    Mapper<Text, Object, Text, Object> identity = new Mapper<Text, Object, Text, Object>();
+    testDriver.addMapper(identity);
+    Text k1 = new Text("foo");
+    Object v1 = new UncomparableWritable(1);
+    testDriver.withInput(identity, k1, v1);
+
+    ReverseMapper<Object, Text> reverse = new ReverseMapper<Object, Text>();
+    testDriver.addMapper(reverse);
+    Text k2 = new Text("bar");
+    Object v2 = new UncomparableWritable(2);
+    testDriver.withInput(reverse, v2, k2);
+
+    testDriver.withOutput(k1, v1).withOutput(k2, v2);
+
+    testDriver.runTest(false);
+  }
+
+  @Test
+  public void testTestRun() throws IOException {
+    driver
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A))
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_B))
+        .withInput(mapper, new Text("bar"), new LongWritable(BAR_IN))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_A), new Text(TOKEN_A))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_B), new Text(TOKEN_B))
+        .withOutput(new Text("bar"), new LongWritable(BAR_OUT))
+        .withOutput(new Text("foo"), new LongWritable(FOO_OUT)).runTest(false);
+  }
+
+  @Test
+  public void testAddAll() throws IOException {
+    final List<Pair<Text, LongWritable>> mapperInputs = new ArrayList<Pair<Text, LongWritable>>();
+    mapperInputs.add(new Pair<Text, LongWritable>(new Text("foo"),
+        new LongWritable(FOO_IN_A)));
+    mapperInputs.add(new Pair<Text, LongWritable>(new Text("foo"),
+        new LongWritable(FOO_IN_B)));
+    mapperInputs.add(new Pair<Text, LongWritable>(new Text("bar"),
+        new LongWritable(BAR_IN)));
+
+    final List<Pair<LongWritable, Text>> tokenMapperInputs = new ArrayList<Pair<LongWritable, Text>>();
+    tokenMapperInputs.add(new Pair<LongWritable, Text>(new LongWritable(
+        TOKEN_IN_A), new Text(TOKEN_A)));
+    tokenMapperInputs.add(new Pair<LongWritable, Text>(new LongWritable(
+        TOKEN_IN_B), new Text(TOKEN_B)));
+
+    final List<Pair<Text, LongWritable>> outputs = new ArrayList<Pair<Text, LongWritable>>();
+    outputs.add(new Pair<Text, LongWritable>(new Text("bar"), new LongWritable(
+        BAR_OUT)));
+    outputs.add(new Pair<Text, LongWritable>(new Text("foo"), new LongWritable(
+        FOO_OUT)));
+
+    driver.withAll(mapper, mapperInputs)
+        .withAll(tokenMapper, tokenMapperInputs).withAllOutput(outputs)
+        .runTest(false);
+  }
+
+  @Test
+  public void testNoInput() throws IOException {
+    thrown.expectMessage(IllegalStateException.class,
+        "No input was provided for mapper");
+    driver.runTest(false);
+  }
+
+  @Test
+  public void testNoInputForMapper() throws IOException {
+    MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable> testDriver = new MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable>();
+    testDriver.addMapper(mapper);
+    testDriver.addMapper(tokenMapper);
+    testDriver.withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A));
+    thrown.expectMessage(IllegalStateException.class,
+        String.format("No input was provided for mapper %s", tokenMapper));
+    testDriver.runTest(false);
+  }
+
+  @Test
+  public void testNoReducer() throws IOException {
+    MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable> testDriver = new MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable>();
+    testDriver.addMapper(mapper);
+    testDriver.withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A));
+    thrown.expectMessage(IllegalStateException.class,
+        "No reducer class was provided");
+    testDriver.runTest(false);
+  }
+
+  @Test
+  public void testIdentityCombiner() throws IOException {
+    driver
+        .withCombiner(new Reducer<Text, LongWritable, Text, LongWritable>())
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A))
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_B))
+        .withInput(mapper, new Text("bar"), new LongWritable(BAR_IN))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_A), new Text(TOKEN_A))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_B), new Text(TOKEN_B))
+        .withOutput(new Text("foo"), new LongWritable(FOO_OUT))
+        .withOutput(new Text("bar"), new LongWritable(BAR_OUT)).runTest(false);
+  }
+
+  @Test
+  public void testLongSumCombiner() throws IOException {
+    driver
+        .withCombiner(new LongSumReducer<Text>())
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A))
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_B))
+        .withInput(mapper, new Text("bar"), new LongWritable(BAR_IN))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_A), new Text(TOKEN_A))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_B), new Text(TOKEN_B))
+        .withOutput(new Text("foo"), new LongWritable(FOO_OUT))
+        .withOutput(new Text("bar"), new LongWritable(BAR_OUT)).runTest(false);
+  }
+
+  @Test
+  public void testLongSumCombinerAndIdentityReducer() throws IOException {
+    driver
+        .withCombiner(new LongSumReducer<Text>())
+        .withReducer(new Reducer<Text, LongWritable, Text, LongWritable>())
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A))
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_B))
+        .withInput(mapper, new Text("bar"), new LongWritable(BAR_IN))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_A), new Text(TOKEN_A))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_B), new Text(TOKEN_B))
+        .withOutput(new Text("foo"), new LongWritable(FOO_OUT))
+        .withOutput(new Text("bar"), new LongWritable(BAR_OUT)).runTest(false);
+  }
+
+  @Test
+  public void testRepeatRun() throws IOException {
+    driver
+        .withCombiner(new Reducer<Text, LongWritable, Text, LongWritable>())
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_A))
+        .withInput(mapper, new Text("foo"), new LongWritable(FOO_IN_B))
+        .withInput(mapper, new Text("bar"), new LongWritable(BAR_IN))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_A), new Text(TOKEN_A))
+        .withInput(tokenMapper, new LongWritable(TOKEN_IN_B), new Text(TOKEN_B))
+        .withOutput(new Text("foo"), new LongWritable(FOO_OUT))
+        .withOutput(new Text("bar"), new LongWritable(BAR_OUT)).runTest(false);
+    thrown.expectMessage(IllegalStateException.class,
+        "Driver reuse not allowed");
+    driver.runTest(false);
+  }
+
+  // Test the key grouping and value ordering comparators
+  @Test
+  public void testComparators() throws IOException {
+    // reducer to track the order of the input values using bit shifting
+    driver.withReducer(new Reducer<Text, LongWritable, Text, LongWritable>() {
+      @Override
+      protected void reduce(Text key, Iterable<LongWritable> values,
+          Context context) throws IOException, InterruptedException {
+        Text outKey = new Text(key);
+        long outputValue = 0;
+        int count = 0;
+        for (LongWritable value : values) {
+          outputValue |= (value.get() << (count++ * 8));
+        }
+
+        context.write(outKey, new LongWritable(outputValue));
+      }
+    });
+
+    driver
+        .withKeyGroupingComparator(new org.apache.hadoop.mrunit.TestMapReduceDriver.FirstCharComparator());
+    driver
+        .withKeyOrderComparator(new org.apache.hadoop.mrunit.TestMapReduceDriver.SecondCharComparator());
+
+    driver.addInput(mapper, new Text("a1"), new LongWritable(1));
+    driver.addInput(mapper, new Text("b1"), new LongWritable(1));
+    driver.addInput(mapper, new Text("a3"), new LongWritable(3));
+    driver.addInput(mapper, new Text("a2"), new LongWritable(2));
+
+    driver.addInput(tokenMapper, new LongWritable(1), new Text("c1 d1"));
+
+    driver.addOutput(new Text("a1"), new LongWritable(0x1));
+    driver.addOutput(new Text("b1"), new LongWritable(0x1));
+    driver.addOutput(new Text("a2"), new LongWritable(0x2 | (0x3 << 8)));
+    driver.addOutput(new Text("c1"), new LongWritable(0x1));
+    driver.addOutput(new Text("d1"), new LongWritable(0x1));
+
+    driver.runTest(false);
+  }
+
+  @Test
+  public void testNoMapper() throws IOException {
+    MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable> testDriver = new MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable>();
+    testDriver.withReducer(reducer);
+    thrown.expectMessage(IllegalStateException.class,
+        "No mappers were provided");
+    testDriver.runTest(false);
+  }
+
+  @Test
+  public void testWithCounter() throws IOException {
+    MultipleInputsMapReduceDriver<Text, Text, Text, Text> testDriver = new MultipleInputsMapReduceDriver<Text, Text, Text, Text>();
+    Mapper<Text, Text, Text, Text> mapperWithCounters = new TestMapDriver.MapperWithCounters<Text, Text, Text, Text>();
+    Mapper<Text, Text, Text, Text> tokenMapperWithCounters = new TokenMapperWithCounters();
+    testDriver
+        .withMapper(mapperWithCounters)
+        .withInput(mapperWithCounters, new Text("hie"), new Text("Hi"))
+        .withMapper(tokenMapperWithCounters)
+        .withInput(tokenMapperWithCounters, new Text("bie"),
+            new Text("Goodbye Bye"))
+        .withCounter(TestMapDriver.MapperWithCounters.Counters.X, 1)
+        .withCounter(TokenMapperWithCounters.Counters.Y, 2)
+        .withCounter("category", "name", 3)
+        .withReducer(
+            new TestReduceDriver.ReducerWithCounters<Text, Text, Text, Text>())
+        .withCounter(TestReduceDriver.ReducerWithCounters.Counters.COUNT, 2)
+        .withCounter(TestReduceDriver.ReducerWithCounters.Counters.SUM, 3)
+        .withCounter("category", "count", 2).withCounter("category", "sum", 3)
+        .runTest(false);
+  }
+
+  @Test
+  public void testWithCounterAndEnumCounterMissing() throws IOException {
+    MultipleInputsMapReduceDriver<Text, Text, Text, Text> testDriver = new MultipleInputsMapReduceDriver<Text, Text, Text, Text>();
+
+    thrown
+        .expectAssertionErrorMessage("2 Error(s): (Actual counter ("
+            + "\"org.apache.hadoop.mrunit.mapreduce.TestMapDriver$MapperWithCounters$Counters\",\"X\")"
+            + " was not found in expected counters, Actual counter ("
+            + "\"org.apache.hadoop.mrunit.mapreduce.TestMultipleInputsMapReduceDriver$TokenMapperWithCounters$Counters\",\"Y\")"
+            + " was not found in expected counters");
+
+    Mapper<Text, Text, Text, Text> mapperWithCounters = new TestMapDriver.MapperWithCounters<Text, Text, Text, Text>();
+    Mapper<Text, Text, Text, Text> tokenMapperWithCounters = new TokenMapperWithCounters();
+
+    testDriver
+        .withMapper(mapperWithCounters)
+        .withInput(mapperWithCounters, new Text("hie"), new Text("Hi"))
+        .withMapper(tokenMapperWithCounters)
+        .withInput(tokenMapperWithCounters, new Text("bie"),
+            new Text("Goodbye Bye"))
+        .withStrictCounterChecking()
+        .withCounter("category", "name", 3)
+        .withReducer(
+            new TestReduceDriver.ReducerWithCounters<Text, Text, Text, Text>())
+        .withCounter(TestReduceDriver.ReducerWithCounters.Counters.COUNT, 2)
+        .withCounter(TestReduceDriver.ReducerWithCounters.Counters.SUM, 3)
+        .withCounter("category", "count", 2).withCounter("category", "sum", 3)
+        .runTest(false);
+  }
+
+  @Test
+  public void testWithCounterAndStringCounterMissing() throws IOException {
+    MultipleInputsMapReduceDriver<Text, Text, Text, Text> testDriver = new MultipleInputsMapReduceDriver<Text, Text, Text, Text>();
+
+    thrown.expectAssertionErrorMessage("1 Error(s): (Actual counter ("
+        + "\"category\",\"name\")" + " was not found in expected counters");
+
+    Mapper<Text, Text, Text, Text> mapperWithCounters = new TestMapDriver.MapperWithCounters<Text, Text, Text, Text>();
+    Mapper<Text, Text, Text, Text> tokenMapperWithCounters = new TokenMapperWithCounters();
+
+    testDriver
+        .withMapper(mapperWithCounters)
+        .withInput(mapperWithCounters, new Text("hie"), new Text("Hi"))
+        .withMapper(tokenMapperWithCounters)
+        .withInput(tokenMapperWithCounters, new Text("bie"),
+            new Text("Goodbye Bye"))
+        .withStrictCounterChecking()
+        .withCounter(TestMapDriver.MapperWithCounters.Counters.X, 1)
+        .withCounter(TokenMapperWithCounters.Counters.Y, 2)
+        .withReducer(
+            new TestReduceDriver.ReducerWithCounters<Text, Text, Text, Text>())
+        .withCounter(TestReduceDriver.ReducerWithCounters.Counters.COUNT, 2)
+        .withCounter(TestReduceDriver.ReducerWithCounters.Counters.SUM, 3)
+        .withCounter("category", "count", 2).withCounter("category", "sum", 3)
+        .runTest(false);
+  }
+
+  @Test
+  public void testWithFailedCounter() throws IOException {
+    MultipleInputsMapReduceDriver<Text, Text, Text, Text> testDriver = new MultipleInputsMapReduceDriver<Text, Text, Text, Text>();
+
+    thrown
+        .expectAssertionErrorMessage("3 Error(s): ("
+            + "Counter org.apache.hadoop.mrunit.mapreduce.TestMapDriver.MapperWithCounters.Counters.X has value 1 instead of expected 20, "
+            + "Counter org.apache.hadoop.mrunit.mapreduce.TestMultipleInputsMapReduceDriver.TokenMapperWithCounters.Counters.Y has value 2 instead of expected 30, "
+            + "Counter with category category and name name has value 3 instead of expected 20)");
+
+    Mapper<Text, Text, Text, Text> mapperWithCounters = new TestMapDriver.MapperWithCounters<Text, Text, Text, Text>();
+    Mapper<Text, Text, Text, Text> tokenMapperWithCounters = new TokenMapperWithCounters();
+
+    testDriver
+        .withMapper(mapperWithCounters)
+        .withInput(mapperWithCounters, new Text("hie"), new Text("Hi"))
+        .withMapper(tokenMapperWithCounters)
+        .withInput(tokenMapperWithCounters, new Text("bie"),
+            new Text("Goodbye Bye"))
+        .withCounter(TestMapDriver.MapperWithCounters.Counters.X, 20)
+        .withCounter(TokenMapperWithCounters.Counters.Y, 30)
+        .withReducer(
+            new TestReduceDriver.ReducerWithCounters<Text, Text, Text, Text>())
+        .withCounter("category", "name", 20).runTest(false);
+  }
+
+  @Test
+  public void testJavaSerialization() throws IOException {
+    final Configuration conf = new Configuration();
+    conf.setStrings("io.serializations", conf.get("io.serializations"),
+        "org.apache.hadoop.io.serializer.JavaSerialization");
+    final MultipleInputsMapReduceDriver<Integer, IntWritable, Integer, IntWritable> testDriver = MultipleInputsMapReduceDriver
+        .newMultipleInputMapReduceDriver(
+                new Reducer<Integer, IntWritable, Integer, IntWritable>())
+        .withConfiguration(conf);
+    Mapper<Integer, IntWritable, Integer, IntWritable> identityMapper = new Mapper<Integer, IntWritable, Integer, IntWritable>();
+    Mapper<Integer, IntWritable, Integer, IntWritable> anotherIdentityMapper = new Mapper<Integer, IntWritable, Integer, IntWritable>();
+    testDriver.addMapper(identityMapper);
+    testDriver.withInput(identityMapper, 1, new IntWritable(2)).withInput(
+        identityMapper, 2, new IntWritable(3));
+    testDriver.addMapper(anotherIdentityMapper);
+    testDriver.withInput(anotherIdentityMapper, 3, new IntWritable(4))
+        .withInput(anotherIdentityMapper, 4, new IntWritable(5));
+    testDriver
+        .withKeyOrderComparator(new JavaSerializationComparator<Integer>());
+    testDriver
+        .withKeyGroupingComparator(org.apache.hadoop.mrunit.TestMapReduceDriver.INTEGER_COMPARATOR);
+
+    testDriver.withOutput(1, new IntWritable(2))
+        .withOutput(2, new IntWritable(3)).withOutput(3, new IntWritable(4))
+        .withOutput(4, new IntWritable(5));
+    testDriver.runTest(false);
+  }
+
+  @Test
+  public void testCopy() throws IOException {
+    final Text key = new Text("a");
+    final LongWritable value = new LongWritable(1);
+    driver.addInput(mapper, key, value);
+    key.set("b");
+    value.set(2);
+    driver.addInput(mapper, key, value);
+
+    key.set("a");
+    value.set(1);
+    driver.addOutput(key, value);
+    key.set("b");
+    value.set(2);
+    driver.addOutput(key, value);
+
+    final LongWritable longKey = new LongWritable(3);
+    final Text textValue = new Text("c d");
+    driver.addInput(tokenMapper, longKey, textValue);
+    longKey.set(4);
+    textValue.set("e f g");
+    driver.addInput(tokenMapper, longKey, textValue);
+
+    key.set("c");
+    value.set(3);
+    driver.addOutput(key, value);
+    key.set("d");
+    value.set(3);
+    driver.addOutput(key, value);
+    key.set("e");
+    value.set(4);
+    driver.addOutput(key, value);
+    key.set("f");
+    value.set(4);
+    driver.addOutput(key, value);
+    key.set("g");
+    value.set(4);
+    driver.addOutput(key, value);
+
+    driver.runTest(false);
+  }
+
+  @Test
+  public void testOutputFormat() throws IOException {
+    driver.withInputFormat(SequenceFileInputFormat.class);
+    driver.withOutputFormat(SequenceFileOutputFormat.class);
+    driver.withInput(mapper, new Text("a"), new LongWritable(1));
+    driver.withInput(mapper, new Text("a"), new LongWritable(2));
+    driver.withInput(tokenMapper, new LongWritable(3), new Text("a b"));
+    driver.withOutput(new Text("a"), new LongWritable(6));
+    driver.withOutput(new Text("b"), new LongWritable(3));
+    driver.runTest(false);
+  }
+
+  @SuppressWarnings({ "unchecked", "rawtypes" })
+  @Test
+  public void testOutputFormatWithMismatchInOutputClasses() throws IOException {
+    final MultipleInputsMapReduceDriver testDriver = this.driver;
+    testDriver.withInputFormat(TextInputFormat.class);
+    testDriver.withOutputFormat(TextOutputFormat.class);
+    testDriver.withInput(mapper, new Text("a"), new LongWritable(1));
+    testDriver.withInput(mapper, new Text("a"), new LongWritable(2));
+    testDriver.withInput(tokenMapper, new LongWritable(3), new Text("a b"));
+    testDriver.withOutput(new LongWritable(0), new Text("a\t6"));
+    testDriver.withOutput(new LongWritable(4), new Text("b\t3"));
+    testDriver.runTest(false);
+  }
+
+  @Test
+  public void testMapInputFile() throws IOException {
+    MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable> testDriver = new MultipleInputsMapReduceDriver<Text, LongWritable, Text, LongWritable>(
+        reducer);
+
+    InputPathStoringMapper<LongWritable, LongWritable> inputPathStoringMapper = new InputPathStoringMapper<LongWritable, LongWritable>();
+    Path mapInputPath = new Path("myfile");
+    testDriver.addMapper(inputPathStoringMapper);
+    testDriver.setMapInputPath(inputPathStoringMapper, mapInputPath);
+    assertEquals(mapInputPath.getName(),
+        testDriver.getMapInputPath(inputPathStoringMapper).getName());
+    testDriver.withInput(inputPathStoringMapper, new Text("a"),
+        new LongWritable(1));
+
+    InputPathStoringMapper<LongWritable, LongWritable> anotherInputPathStoringMapper = new InputPathStoringMapper<LongWritable, LongWritable>();
+    Path anotherMapInputPath = new Path("myotherfile");
+    testDriver.addMapper(anotherInputPathStoringMapper);
+    testDriver.setMapInputPath(anotherInputPathStoringMapper,
+        anotherMapInputPath);
+    assertEquals(anotherMapInputPath.getName(),
+        testDriver.getMapInputPath(anotherInputPathStoringMapper).getName());
+    testDriver.withInput(anotherInputPathStoringMapper, new Text("b"),
+        new LongWritable(2));
+
+    testDriver.runTest(false);
+    assertNotNull(inputPathStoringMapper.getMapInputPath());
+    assertEquals(mapInputPath.getName(), inputPathStoringMapper
+        .getMapInputPath().getName());
+  }
+
+  @Test
+  public void testGroupComparatorBehaviorFirst() throws IOException {
+    driver
+        .withInput(mapper, new Text("A1"), new LongWritable(1L))
+        .withInput(mapper, new Text("A2"), new LongWritable(1L))
+        .withInput(mapper, new Text("B1"), new LongWritable(1L))
+        .withInput(mapper, new Text("B2"), new LongWritable(1L))
+        .withInput(mapper, new Text("C1"), new LongWritable(1L))
+        .withInput(tokenMapper, new LongWritable(3L), new Text("D1 D2 E1"))
+        .withOutput(new Text("A2"), new LongWritable(2L))
+        .withOutput(new Text("B2"), new LongWritable(2L))
+        .withOutput(new Text("C1"), new LongWritable(1L))
+        .withOutput(new Text("D2"), new LongWritable(6L))
+        .withOutput(new Text("E1"), new LongWritable(3L))
+        .withKeyGroupingComparator(
+            new org.apache.hadoop.mrunit.TestMapReduceDriver.FirstCharComparator())
+        .runTest(false);
+  }
+
+  @Test
+  public void testGroupComparatorBehaviorSecond() throws IOException {
+    driver
+        .withInput(mapper, new Text("1A"), new LongWritable(1L))
+        .withInput(mapper, new Text("2A"), new LongWritable(1L))
+        .withInput(mapper, new Text("1B"), new LongWritable(1L))
+        .withInput(mapper, new Text("2B"), new LongWritable(1L))
+        .withInput(mapper, new Text("1C"), new LongWritable(1L))
+        .withInput(tokenMapper, new LongWritable(2L), new Text("1D 2D 1E"))
+        .withOutput(new Text("1A"), new LongWritable(1L))
+        .withOutput(new Text("2A"), new LongWritable(1L))
+        .withOutput(new Text("1B"), new LongWritable(1L))
+        .withOutput(new Text("2B"), new LongWritable(1L))
+        .withOutput(new Text("1C"), new LongWritable(1L))
+        .withOutput(new Text("1D"), new LongWritable(2L))
+        .withOutput(new Text("2D"), new LongWritable(2L))
+        .withOutput(new Text("1E"), new LongWritable(2L))
+        .withKeyGroupingComparator(
+            new org.apache.hadoop.mrunit.TestMapReduceDriver.SecondCharComparator())
+        .runTest(false);
+  }
+
+  @Test
+  public void testGroupingComparatorSpecifiedByConf() throws IOException {
+    JobConf conf = new JobConf(new Configuration());
+    conf.setOutputValueGroupingComparator(TestMapReduceDriver.FirstCharComparator.class);
+    driver.withInput(mapper, new Text("A1"), new LongWritable(1L))
+        .withInput(mapper, new Text("A2"), new LongWritable(1L))
+        .withInput(mapper, new Text("B1"), new LongWritable(1L))
+        .withInput(mapper, new Text("B2"), new LongWritable(1L))
+        .withInput(mapper, new Text("C1"), new LongWritable(1L))
+        .withInput(tokenMapper, new LongWritable(3L), new Text("D1 D2 E1"))
+        .withOutput(new Text("A2"), new LongWritable(2L))
+        .withOutput(new Text("B2"), new LongWritable(2L))
+        .withOutput(new Text("C1"), new LongWritable(1L))
+        .withOutput(new Text("D2"), new LongWritable(6L))
+        .withOutput(new Text("E1"), new LongWritable(3L))
+        .withConfiguration(conf).runTest(false);
+  }
+
+  @SuppressWarnings("unchecked")
+  @Test
+  public void testUseOfWritableRegisteredComparator() throws IOException {
+    MultipleInputsMapReduceDriver<TestWritable, Text, TestWritable, Text> testDriver = new MultipleInputsMapReduceDriver<TestWritable, Text, TestWritable, Text>(
+        new Reducer<TestWritable, Text, TestWritable, Text>());
+
+    Mapper<TestWritable, Text, TestWritable, Text> identityMapper = new Mapper<TestWritable, Text, TestWritable, Text>();
+    Mapper<TestWritable, Text, TestWritable, Text> anotherIdentityMapper = new Mapper<TestWritable, Text, TestWritable, Text>();
+    testDriver.addMapper(identityMapper);
+    testDriver.addMapper(anotherIdentityMapper);
+    testDriver
+        .withInput(identityMapper, new TestWritable("A1"), new Text("A1"))
+        .withInput(identityMapper, new TestWritable("A2"), new Text("A2"))
+        .withInput(identityMapper, new TestWritable("A3"), new Text("A3"))
+        .withInput(anotherIdentityMapper, new TestWritable("B1"),
+            new Text("B1"))
+        .withInput(anotherIdentityMapper, new TestWritable("B2"),
+            new Text("B2"))
+        .withKeyGroupingComparator(new TestWritable.SingleGroupComparator())
+        .withOutput(new TestWritable("B2"), new Text("B2"))
+        .withOutput(new TestWritable("B1"), new Text("B1"))
+        .withOutput(new TestWritable("A3"), new Text("A3"))
+        .withOutput(new TestWritable("A2"), new Text("A2"))
+        .withOutput(new TestWritable("A1"), new Text("A1")).runTest(true); // ordering
+                                                                           // is
+                                                                           // important
+  }
+
+  static class TokenMapperWithCounters extends Mapper<Text, Text, Text, Text> {
+    private final Text output = new Text();
+
+    @Override
+    protected void map(Text key, Text value, Context context)
+        throws IOException, InterruptedException {
+      String[] tokens = value.toString().split("\\s");
+      for (String token : tokens) {
+        output.set(token);
+        context.write(key, output);
+        context.getCounter(Counters.Y).increment(1);
+        context.getCounter("category", "name").increment(1);
+      }
+    }
+
+    public static enum Counters {
+      Y
+    }
+  }
+
+  static class TokenMapper extends
+      Mapper<LongWritable, Text, Text, LongWritable> {
+    private final Text output = new Text();
+
+    @Override
+    protected void map(LongWritable key, Text value, Context context)
+        throws IOException, InterruptedException {
+      String[] tokens = value.toString().split("\\s");
+      for (String token : tokens) {
+        output.set(token);
+        context.write(output, key);
+      }
+    }
+  }
+
+  static class ReverseMapper<KEYIN, VALUEIN> extends
+      Mapper<KEYIN, VALUEIN, VALUEIN, KEYIN> {
+    @Override
+    protected void map(KEYIN key, VALUEIN value, Context context)
+        throws IOException, InterruptedException {
+      context.write(value, key);
+    }
+  }
+}