tez-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From ss...@apache.org
Subject [2/6] TEZ-1417. Rename *Configurer to ConfigBuilder/Config. (sseth)
Date Tue, 19 Aug 2014 00:02:08 GMT
http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedGroupedMergedKVInputConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedGroupedMergedKVInputConfig.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedGroupedMergedKVInputConfig.java
new file mode 100644
index 0000000..1ca167f
--- /dev/null
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedGroupedMergedKVInputConfig.java
@@ -0,0 +1,160 @@
+/*
+ * *
+ *  * 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.tez.runtime.library.conf;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import com.google.common.collect.Maps;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.junit.Test;
+
+public class TestOrderedGroupedMergedKVInputConfig {
+
+  @Test
+  public void testNullParams() {
+    try {
+      OrderedGroupedKVInputConfig.newBuilder(null, "VALUE");
+      fail("Expecting a null parameter list to fail");
+    } catch (NullPointerException npe) {
+      assertTrue(npe.getMessage().contains("cannot be null"));
+    }
+
+    try {
+      OrderedGroupedKVInputConfig.newBuilder("KEY", null);
+      fail("Expecting a null parameter list to fail");
+    } catch (NullPointerException npe) {
+      assertTrue(npe.getMessage().contains("cannot be null"));
+    }
+  }
+
+  @Test
+  public void testSetters() {
+    Configuration fromConf = new Configuration(false);
+    fromConf.set("test.conf.key.1", "confkey1");
+    fromConf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 1111);
+    fromConf.set("io.shouldExist", "io");
+    Map<String, String> additionalConf = new HashMap<String, String>();
+    additionalConf.put("test.key.2", "key2");
+    additionalConf.put(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR, "3");
+    additionalConf.put("file.shouldExist", "file");
+    OrderedGroupedKVInputConfig.Builder builder =
+        OrderedGroupedKVInputConfig.newBuilder("KEY", "VALUE")
+            .setKeyComparatorClass("KEY_COMPARATOR", null)
+            .setCompression(true, "CustomCodec", null)
+            .setMaxSingleMemorySegmentFraction(0.11f)
+            .setMergeFraction(0.22f)
+            .setPostMergeBufferFraction(0.33f)
+            .setShuffleBufferFraction(0.44f)
+            .setAdditionalConfiguration("fs.shouldExist", "fs")
+            .setAdditionalConfiguration("test.key.1", "key1")
+            .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+                String.valueOf(false))
+            .setAdditionalConfiguration(additionalConf)
+            .setFromConfiguration(fromConf);
+
+    OrderedGroupedKVInputConfig configuration = builder.build();
+
+
+    OrderedGroupedKVInputConfig rebuilt = new OrderedGroupedKVInputConfig();
+    rebuilt.fromUserPayload(configuration.toUserPayload());
+
+    Configuration conf = rebuilt.conf;
+
+    // Verify programmatic API usage
+    assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
+    assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
+    assertEquals("CustomCodec",
+        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+    assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
+        false));
+    assertEquals(0.11f, conf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT, 0.0f), 0.001f);
+    assertEquals(0.22f, conf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, 0.0f), 0.001f);
+    assertEquals(0.33f, conf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_INPUT_BUFFER_PERCENT, 0.0f), 0.001f);
+    assertEquals(0.44f, conf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, 0.00f), 0.001f);
+
+    // Verify additional configs
+    assertEquals(false, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
+    assertEquals(1111, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT));
+    assertEquals(3, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR, -1));
+    assertEquals("io", conf.get("io.shouldExist"));
+    assertEquals("file", conf.get("file.shouldExist"));
+    assertEquals("fs", conf.get("fs.shouldExist"));
+    assertNull(conf.get("test.conf.key.1"));
+    assertNull(conf.get("test.key.1"));
+    assertNull(conf.get("test.key.2"));
+  }
+
+  @Test
+  public void testDefaultConfigsUsed() {
+    OrderedGroupedKVInputConfig.Builder builder =
+        OrderedGroupedKVInputConfig.newBuilder("KEY", "VALUE");
+    OrderedGroupedKVInputConfig configuration = builder.build();
+
+    OrderedGroupedKVInputConfig rebuilt = new OrderedGroupedKVInputConfig();
+    rebuilt.fromUserPayload(configuration.toUserPayload());
+
+    Configuration conf = rebuilt.conf;
+
+    assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
+
+    // Default property present.
+    assertEquals("TestCodec",
+        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+
+    // Verify whatever was configured
+    assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
+    assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
+  }
+
+  @Test
+  public void testCombinerConfigs() {
+    Map<String, String> combinerConf = Maps.newHashMap();
+    combinerConf.put("combiner.test.key", "COMBINERKEY");
+    combinerConf
+        .put(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "InvalidKeyOverride");
+    OrderedGroupedKVInputConfig.Builder builder =
+        OrderedGroupedKVInputConfig.newBuilder("KEY", "VALUE")
+            .setCombiner("COMBINER", combinerConf);
+
+    OrderedGroupedKVInputConfig configuration = builder.build();
+
+    OrderedGroupedKVInputConfig rebuilt = new OrderedGroupedKVInputConfig();
+    rebuilt.fromUserPayload(configuration.toUserPayload());
+
+    Configuration conf = rebuilt.conf;
+
+    // Default Output property should not be overridden based on partitioner config
+    assertEquals("TestCodec",
+        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+
+    assertEquals("COMBINERKEY", conf.get("combiner.test.key"));
+  }
+}

http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedPartitionedKVEdgeConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedPartitionedKVEdgeConfig.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedPartitionedKVEdgeConfig.java
new file mode 100644
index 0000000..54d4a90
--- /dev/null
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedPartitionedKVEdgeConfig.java
@@ -0,0 +1,297 @@
+/*
+ * *
+ *  * 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.tez.runtime.library.conf;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import com.google.common.collect.Maps;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.junit.Test;
+
+public class TestOrderedPartitionedKVEdgeConfig {
+
+  @Test
+  public void testNullParams() {
+    try {
+      OrderedPartitionedKVEdgeConfig.newBuilder(null, "VALUE", "PARTITIONER");
+      fail("Expecting a null parameter list to fail");
+    } catch (NullPointerException npe) {
+      assertTrue(npe.getMessage().contains("cannot be null"));
+    }
+
+    try {
+      OrderedPartitionedKVEdgeConfig.newBuilder("KEY", null, "PARTITIONER");
+      fail("Expecting a null parameter list to fail");
+    } catch (NullPointerException npe) {
+      assertTrue(npe.getMessage().contains("cannot be null"));
+    }
+
+    try {
+      OrderedPartitionedKVEdgeConfig.newBuilder("KEY", "VALUE", null);
+      fail("Expecting a null parameter list to fail");
+    } catch (NullPointerException npe) {
+      assertTrue(npe.getMessage().contains("cannot be null"));
+    }
+  }
+
+  @Test
+  public void testDefaultConfigsUsed() {
+    OrderedPartitionedKVEdgeConfig.Builder builder = OrderedPartitionedKVEdgeConfig
+        .newBuilder("KEY", "VALUE", "PARTITIONER");
+
+    OrderedPartitionedKVEdgeConfig configuration = builder.build();
+
+    OrderedPartitionedKVOutputConfig
+        rebuiltOutput = new OrderedPartitionedKVOutputConfig();
+    rebuiltOutput.fromUserPayload(configuration.getOutputPayload());
+    OrderedGroupedKVInputConfig rebuiltInput = new OrderedGroupedKVInputConfig();
+    rebuiltInput.fromUserPayload(configuration.getInputPayload());
+
+    Configuration outputConf = rebuiltOutput.conf;
+    assertEquals(true, outputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
+    assertEquals("TestCodec",
+        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+
+    Configuration inputConf = rebuiltInput.conf;
+    assertEquals(true, inputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
+    assertEquals("TestCodec",
+        inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+  }
+
+  @Test
+  public void testSpecificIOConfs() {
+    // Ensures that Output and Input confs are not mixed.
+    OrderedPartitionedKVEdgeConfig.Builder builder = OrderedPartitionedKVEdgeConfig
+        .newBuilder("KEY", "VALUE", "PARTITIONER");
+
+    OrderedPartitionedKVEdgeConfig configuration = builder.build();
+
+    OrderedPartitionedKVOutputConfig
+        rebuiltOutput = new OrderedPartitionedKVOutputConfig();
+    rebuiltOutput.fromUserPayload(configuration.getOutputPayload());
+    OrderedGroupedKVInputConfig rebuiltInput = new OrderedGroupedKVInputConfig();
+    rebuiltInput.fromUserPayload(configuration.getInputPayload());
+
+    Configuration outputConf = rebuiltOutput.conf;
+    assertEquals("TestCodec",
+        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "DEFAULT"));
+
+    Configuration inputConf = rebuiltInput.conf;
+    assertEquals("TestCodec",
+        inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "DEFAULT"));
+  }
+
+  @Test
+  public void tetCommonConf() {
+
+    Configuration fromConf = new Configuration(false);
+    fromConf.set("test.conf.key.1", "confkey1");
+    fromConf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR, 3);
+    fromConf.setFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, 0.11f);
+    fromConf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB, 123);
+    fromConf.set("io.shouldExist", "io");
+    Map<String, String> additionalConfs = new HashMap<String, String>();
+    additionalConfs.put("test.key.2", "key2");
+    additionalConfs.put(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, "1111");
+    additionalConfs.put(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT, "0.22f");
+    additionalConfs.put(TezRuntimeConfiguration.TEZ_RUNTIME_INTERNAL_SORTER_CLASS, "CustomSorter");
+    additionalConfs.put("file.shouldExist", "file");
+
+    OrderedPartitionedKVEdgeConfig.Builder builder = OrderedPartitionedKVEdgeConfig
+        .newBuilder("KEY", "VALUE", "PARTITIONER")
+        .setAdditionalConfiguration("fs.shouldExist", "fs")
+        .setAdditionalConfiguration("test.key.1", "key1")
+        .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE, "2222")
+        .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, "0.33f")
+        .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES, "3333")
+        .setAdditionalConfiguration(additionalConfs)
+        .setFromConfiguration(fromConf);
+
+    OrderedPartitionedKVEdgeConfig configuration = builder.build();
+
+    OrderedPartitionedKVOutputConfig
+        rebuiltOutput = new OrderedPartitionedKVOutputConfig();
+    rebuiltOutput.fromUserPayload(configuration.getOutputPayload());
+    OrderedGroupedKVInputConfig rebuiltInput = new OrderedGroupedKVInputConfig();
+    rebuiltInput.fromUserPayload(configuration.getInputPayload());
+
+    Configuration outputConf = rebuiltOutput.conf;
+    Configuration inputConf = rebuiltInput.conf;
+
+    assertEquals(3, outputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR, 0));
+    assertEquals(1111, outputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 0));
+    assertEquals(2222, outputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE, 0));
+    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT));
+    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT));
+    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT));
+    assertEquals(123, outputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB, 0));
+    assertEquals("CustomSorter", outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_INTERNAL_SORTER_CLASS));
+    assertEquals(3333,
+        outputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES, 0));
+    assertEquals("io", outputConf.get("io.shouldExist"));
+    assertEquals("file", outputConf.get("file.shouldExist"));
+    assertEquals("fs", outputConf.get("fs.shouldExist"));
+
+
+    assertEquals(3, inputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR, 0));
+    assertEquals(1111, inputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 0));
+    assertEquals(2222, inputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE, 0));
+    assertEquals(0.11f,
+        inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, 0.0f), 0.001f);
+    assertEquals(0.22f,
+        inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT, 0.0f), 0.001f);
+    assertEquals(0.33f, inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, 0.0f),
+        0.001f);
+    assertNull(inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB));
+    assertNull(inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_INTERNAL_SORTER_CLASS));
+    assertNull(inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES));
+    assertEquals("io", inputConf.get("io.shouldExist"));
+    assertEquals("file", inputConf.get("file.shouldExist"));
+    assertEquals("fs", inputConf.get("fs.shouldExist"));
+
+  }
+
+  @Test
+  public void testSetters() {
+    Map<String, String> comparatorConf = Maps.newHashMap();
+    comparatorConf.put("comparator.test.key", "comparatorValue");
+    OrderedPartitionedKVEdgeConfig.Builder builder = OrderedPartitionedKVEdgeConfig
+        .newBuilder("KEY", "VALUE", "PARTITIONER")
+        .setKeyComparatorClass("KEY_COMPARATOR", comparatorConf)
+        .configureOutput().setSortBufferSize(1111).setSorterNumThreads(2).done()
+        .configureInput().setMaxSingleMemorySegmentFraction(0.11f).setMergeFraction(0.22f)
+        .setPostMergeBufferFraction(0.33f).setShuffleBufferFraction(0.44f).done()
+        .setCompression(true, "CustomCodec", null);
+
+    OrderedPartitionedKVEdgeConfig configuration = builder.build();
+
+    OrderedPartitionedKVOutputConfig
+        rebuiltOutput = new OrderedPartitionedKVOutputConfig();
+    rebuiltOutput.fromUserPayload(configuration.getOutputPayload());
+    OrderedGroupedKVInputConfig rebuiltInput = new OrderedGroupedKVInputConfig();
+    rebuiltInput.fromUserPayload(configuration.getInputPayload());
+
+    Configuration outputConf = rebuiltOutput.conf;
+    Configuration inputConf = rebuiltInput.conf;
+
+    assertEquals("KEY", outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
+    assertEquals("VALUE",
+        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
+    assertEquals("PARTITIONER", outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, ""));
+    assertEquals(1111, outputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB, 0));
+    assertEquals("CustomCodec",
+        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+    assertEquals(true,
+        outputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
+            false));
+    assertEquals("KEY_COMPARATOR",
+        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS));
+    assertEquals("comparatorValue", outputConf.get("comparator.test.key"));
+    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT));
+    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT));
+    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_INPUT_BUFFER_PERCENT));
+    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT));
+
+
+    assertEquals("KEY_COMPARATOR", inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS));
+    assertEquals("comparatorValue", inputConf.get("comparator.test.key"));
+    assertEquals("KEY", inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
+    assertEquals("VALUE",
+        inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
+    assertEquals("CustomCodec",
+        inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+    assertEquals(true,
+        inputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
+            false));
+
+    assertEquals(0.11f,
+        inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT, 0.0f), 0.001f);
+    assertEquals(0.22f, inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, 0.0f),
+        0.001f);
+    assertEquals(0.33f, inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_INPUT_BUFFER_PERCENT, 0.0f),
+        0.001f);
+    assertEquals(0.44f,
+        inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, 0.00f), 0.001f);
+    assertNull(inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB));
+
+  }
+
+  @Test
+  public void testSerialization() {
+    OrderedPartitionedKVEdgeConfig.Builder builder = OrderedPartitionedKVEdgeConfig
+        .newBuilder("KEY", "VALUE", "PARTITIONER")
+        .setCompression(true, "CustomCodec", null)
+        .setKeySerializationClass("serClass1", "SomeComparator1", null)
+        .setValueSerializationClass("serClass2", null);
+
+    OrderedPartitionedKVEdgeConfig configuration = builder.build();
+
+    OrderedPartitionedKVOutputConfig
+        rebuiltOutput = new OrderedPartitionedKVOutputConfig();
+    rebuiltOutput.fromUserPayload(configuration.getOutputPayload());
+    OrderedGroupedKVInputConfig rebuiltInput = new OrderedGroupedKVInputConfig();
+    rebuiltInput.fromUserPayload(configuration.getInputPayload());
+
+    Configuration outputConf = rebuiltOutput.conf;
+    Configuration inputConf = rebuiltInput.conf;
+
+    assertEquals("KEY", outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
+    assertEquals("VALUE",
+        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
+    assertEquals("PARTITIONER", outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, ""));
+    assertEquals("CustomCodec",
+        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+    assertEquals(true,
+        outputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
+            false));
+    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT));
+    //verify comparator and serialization class
+    assertEquals("SomeComparator1",
+        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS));
+    assertTrue(outputConf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY).trim().startsWith
+        ("serClass2,serClass1"));
+
+
+    //verify comparator and serialization class
+    assertEquals("SomeComparator1", inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS));
+    assertTrue(inputConf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY).trim().startsWith
+        ("serClass2,serClass1"));
+
+    assertEquals("KEY", inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
+    assertEquals("VALUE",
+        inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
+    assertEquals("CustomCodec",
+        inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+    assertEquals(true,
+        inputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
+            false));
+  }
+}

http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedPartitionedKVEdgeConfigurer.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedPartitionedKVEdgeConfigurer.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedPartitionedKVEdgeConfigurer.java
deleted file mode 100644
index 924b988..0000000
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedPartitionedKVEdgeConfigurer.java
+++ /dev/null
@@ -1,292 +0,0 @@
-/*
- * *
- *  * 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.tez.runtime.library.conf;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import com.google.common.collect.Maps;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.fs.CommonConfigurationKeys;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
-import org.junit.Test;
-
-public class TestOrderedPartitionedKVEdgeConfigurer {
-
-  @Test
-  public void testNullParams() {
-    try {
-      OrderedPartitionedKVEdgeConfigurer.newBuilder(null, "VALUE", "PARTITIONER");
-      fail("Expecting a null parameter list to fail");
-    } catch (NullPointerException npe) {
-      assertTrue(npe.getMessage().contains("cannot be null"));
-    }
-
-    try {
-      OrderedPartitionedKVEdgeConfigurer.newBuilder("KEY", null, "PARTITIONER");
-      fail("Expecting a null parameter list to fail");
-    } catch (NullPointerException npe) {
-      assertTrue(npe.getMessage().contains("cannot be null"));
-    }
-
-    try {
-      OrderedPartitionedKVEdgeConfigurer.newBuilder("KEY", "VALUE", null);
-      fail("Expecting a null parameter list to fail");
-    } catch (NullPointerException npe) {
-      assertTrue(npe.getMessage().contains("cannot be null"));
-    }
-  }
-
-  @Test
-  public void testDefaultConfigsUsed() {
-    OrderedPartitionedKVEdgeConfigurer.Builder builder = OrderedPartitionedKVEdgeConfigurer
-        .newBuilder("KEY", "VALUE", "PARTITIONER");
-
-    OrderedPartitionedKVEdgeConfigurer configuration = builder.build();
-
-    OrderedPartitionedKVOutputConfigurer rebuiltOutput = new OrderedPartitionedKVOutputConfigurer();
-    rebuiltOutput.fromUserPayload(configuration.getOutputPayload());
-    OrderedGroupedKVInputConfigurer rebuiltInput = new OrderedGroupedKVInputConfigurer();
-    rebuiltInput.fromUserPayload(configuration.getInputPayload());
-
-    Configuration outputConf = rebuiltOutput.conf;
-    assertEquals(true, outputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
-        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
-    assertEquals("TestCodec",
-        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-
-    Configuration inputConf = rebuiltInput.conf;
-    assertEquals(true, inputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
-        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
-    assertEquals("TestCodec",
-        inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-  }
-
-  @Test
-  public void testSpecificIOConfs() {
-    // Ensures that Output and Input confs are not mixed.
-    OrderedPartitionedKVEdgeConfigurer.Builder builder = OrderedPartitionedKVEdgeConfigurer
-        .newBuilder("KEY", "VALUE", "PARTITIONER");
-
-    OrderedPartitionedKVEdgeConfigurer configuration = builder.build();
-
-    OrderedPartitionedKVOutputConfigurer rebuiltOutput = new OrderedPartitionedKVOutputConfigurer();
-    rebuiltOutput.fromUserPayload(configuration.getOutputPayload());
-    OrderedGroupedKVInputConfigurer rebuiltInput = new OrderedGroupedKVInputConfigurer();
-    rebuiltInput.fromUserPayload(configuration.getInputPayload());
-
-    Configuration outputConf = rebuiltOutput.conf;
-    assertEquals("TestCodec",
-        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "DEFAULT"));
-
-    Configuration inputConf = rebuiltInput.conf;
-    assertEquals("TestCodec",
-        inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "DEFAULT"));
-  }
-
-  @Test
-  public void tetCommonConf() {
-
-    Configuration fromConf = new Configuration(false);
-    fromConf.set("test.conf.key.1", "confkey1");
-    fromConf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR, 3);
-    fromConf.setFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, 0.11f);
-    fromConf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB, 123);
-    fromConf.set("io.shouldExist", "io");
-    Map<String, String> additionalConfs = new HashMap<String, String>();
-    additionalConfs.put("test.key.2", "key2");
-    additionalConfs.put(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, "1111");
-    additionalConfs.put(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT, "0.22f");
-    additionalConfs.put(TezRuntimeConfiguration.TEZ_RUNTIME_INTERNAL_SORTER_CLASS, "CustomSorter");
-    additionalConfs.put("file.shouldExist", "file");
-
-    OrderedPartitionedKVEdgeConfigurer.Builder builder = OrderedPartitionedKVEdgeConfigurer
-        .newBuilder("KEY", "VALUE", "PARTITIONER")
-        .setAdditionalConfiguration("fs.shouldExist", "fs")
-        .setAdditionalConfiguration("test.key.1", "key1")
-        .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE, "2222")
-        .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, "0.33f")
-        .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES, "3333")
-        .setAdditionalConfiguration(additionalConfs)
-        .setFromConfiguration(fromConf);
-
-    OrderedPartitionedKVEdgeConfigurer configuration = builder.build();
-
-    OrderedPartitionedKVOutputConfigurer rebuiltOutput = new OrderedPartitionedKVOutputConfigurer();
-    rebuiltOutput.fromUserPayload(configuration.getOutputPayload());
-    OrderedGroupedKVInputConfigurer rebuiltInput = new OrderedGroupedKVInputConfigurer();
-    rebuiltInput.fromUserPayload(configuration.getInputPayload());
-
-    Configuration outputConf = rebuiltOutput.conf;
-    Configuration inputConf = rebuiltInput.conf;
-
-    assertEquals(3, outputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR, 0));
-    assertEquals(1111, outputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 0));
-    assertEquals(2222, outputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE, 0));
-    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT));
-    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT));
-    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT));
-    assertEquals(123, outputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB, 0));
-    assertEquals("CustomSorter", outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_INTERNAL_SORTER_CLASS));
-    assertEquals(3333,
-        outputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES, 0));
-    assertEquals("io", outputConf.get("io.shouldExist"));
-    assertEquals("file", outputConf.get("file.shouldExist"));
-    assertEquals("fs", outputConf.get("fs.shouldExist"));
-
-
-    assertEquals(3, inputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR, 0));
-    assertEquals(1111, inputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 0));
-    assertEquals(2222, inputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE, 0));
-    assertEquals(0.11f,
-        inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, 0.0f), 0.001f);
-    assertEquals(0.22f,
-        inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT, 0.0f), 0.001f);
-    assertEquals(0.33f, inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, 0.0f),
-        0.001f);
-    assertNull(inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB));
-    assertNull(inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_INTERNAL_SORTER_CLASS));
-    assertNull(inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_INDEX_CACHE_MEMORY_LIMIT_BYTES));
-    assertEquals("io", inputConf.get("io.shouldExist"));
-    assertEquals("file", inputConf.get("file.shouldExist"));
-    assertEquals("fs", inputConf.get("fs.shouldExist"));
-
-  }
-
-  @Test
-  public void testSetters() {
-    Map<String, String> comparatorConf = Maps.newHashMap();
-    comparatorConf.put("comparator.test.key", "comparatorValue");
-    OrderedPartitionedKVEdgeConfigurer.Builder builder = OrderedPartitionedKVEdgeConfigurer
-        .newBuilder("KEY", "VALUE", "PARTITIONER")
-        .setKeyComparatorClass("KEY_COMPARATOR", comparatorConf)
-        .configureOutput().setSortBufferSize(1111).setSorterNumThreads(2).done()
-        .configureInput().setMaxSingleMemorySegmentFraction(0.11f).setMergeFraction(0.22f)
-        .setPostMergeBufferFraction(0.33f).setShuffleBufferFraction(0.44f).done()
-        .setCompression(true, "CustomCodec", null);
-
-    OrderedPartitionedKVEdgeConfigurer configuration = builder.build();
-
-    OrderedPartitionedKVOutputConfigurer rebuiltOutput = new OrderedPartitionedKVOutputConfigurer();
-    rebuiltOutput.fromUserPayload(configuration.getOutputPayload());
-    OrderedGroupedKVInputConfigurer rebuiltInput = new OrderedGroupedKVInputConfigurer();
-    rebuiltInput.fromUserPayload(configuration.getInputPayload());
-
-    Configuration outputConf = rebuiltOutput.conf;
-    Configuration inputConf = rebuiltInput.conf;
-
-    assertEquals("KEY", outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
-    assertEquals("VALUE",
-        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
-    assertEquals("PARTITIONER", outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, ""));
-    assertEquals(1111, outputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB, 0));
-    assertEquals("CustomCodec",
-        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-    assertEquals(true,
-        outputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
-            false));
-    assertEquals("KEY_COMPARATOR",
-        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS));
-    assertEquals("comparatorValue", outputConf.get("comparator.test.key"));
-    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT));
-    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT));
-    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_INPUT_BUFFER_PERCENT));
-    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT));
-
-
-    assertEquals("KEY_COMPARATOR", inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS));
-    assertEquals("comparatorValue", inputConf.get("comparator.test.key"));
-    assertEquals("KEY", inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
-    assertEquals("VALUE",
-        inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
-    assertEquals("CustomCodec",
-        inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-    assertEquals(true,
-        inputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
-            false));
-
-    assertEquals(0.11f,
-        inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT, 0.0f), 0.001f);
-    assertEquals(0.22f, inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, 0.0f),
-        0.001f);
-    assertEquals(0.33f, inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_INPUT_BUFFER_PERCENT, 0.0f),
-        0.001f);
-    assertEquals(0.44f,
-        inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, 0.00f), 0.001f);
-    assertNull(inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB));
-
-  }
-
-  @Test
-  public void testSerialization() {
-    OrderedPartitionedKVEdgeConfigurer.Builder builder = OrderedPartitionedKVEdgeConfigurer
-        .newBuilder("KEY", "VALUE", "PARTITIONER")
-        .setCompression(true, "CustomCodec", null)
-        .setKeySerializationClass("serClass1", "SomeComparator1", null)
-        .setValueSerializationClass("serClass2", null);
-
-    OrderedPartitionedKVEdgeConfigurer configuration = builder.build();
-
-    OrderedPartitionedKVOutputConfigurer rebuiltOutput = new OrderedPartitionedKVOutputConfigurer();
-    rebuiltOutput.fromUserPayload(configuration.getOutputPayload());
-    OrderedGroupedKVInputConfigurer rebuiltInput = new OrderedGroupedKVInputConfigurer();
-    rebuiltInput.fromUserPayload(configuration.getInputPayload());
-
-    Configuration outputConf = rebuiltOutput.conf;
-    Configuration inputConf = rebuiltInput.conf;
-
-    assertEquals("KEY", outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
-    assertEquals("VALUE",
-        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
-    assertEquals("PARTITIONER", outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, ""));
-    assertEquals("CustomCodec",
-        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-    assertEquals(true,
-        outputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
-            false));
-    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT));
-    //verify comparator and serialization class
-    assertEquals("SomeComparator1",
-        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS));
-    assertTrue(outputConf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY).trim().startsWith
-        ("serClass2,serClass1"));
-
-
-    //verify comparator and serialization class
-    assertEquals("SomeComparator1", inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS));
-    assertTrue(inputConf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY).trim().startsWith
-        ("serClass2,serClass1"));
-
-    assertEquals("KEY", inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
-    assertEquals("VALUE",
-        inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
-    assertEquals("CustomCodec",
-        inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-    assertEquals(true,
-        inputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
-            false));
-  }
-}

http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedPartitionedKVOutputConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedPartitionedKVOutputConfig.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedPartitionedKVOutputConfig.java
new file mode 100644
index 0000000..68690ea
--- /dev/null
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestOrderedPartitionedKVOutputConfig.java
@@ -0,0 +1,191 @@
+/*
+ * *
+ *  * 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.tez.runtime.library.conf;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import com.google.common.collect.Maps;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.junit.Test;
+
+public class TestOrderedPartitionedKVOutputConfig {
+
+  @Test
+  public void testNullParams() {
+    try {
+      OrderedPartitionedKVOutputConfig.newBuilder(
+          null, "VALUE", "PARTITIONER", null);
+      fail("Expecting a null parameter list to fail");
+    } catch (NullPointerException npe) {
+      assertTrue(npe.getMessage().contains("cannot be null"));
+    }
+
+    try {
+      OrderedPartitionedKVOutputConfig.newBuilder(
+          "KEY", null, "PARTITIONER", null);
+      fail("Expecting a null parameter list to fail");
+    } catch (NullPointerException npe) {
+      assertTrue(npe.getMessage().contains("cannot be null"));
+    }
+
+    try {
+      OrderedPartitionedKVOutputConfig.newBuilder(
+          "KEY", "VALUE", null, null);
+      fail("Expecting a null parameter list to fail");
+    } catch (NullPointerException npe) {
+      assertTrue(npe.getMessage().contains("cannot be null"));
+    }
+  }
+
+  @Test
+  public void testSetters() {
+    Configuration fromConf = new Configuration(false);
+    fromConf.set("test.conf.key.1", "confkey1");
+    fromConf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 1111);
+    fromConf.set("fs.shouldExist", "fs");
+    Map<String, String> additionalConf = new HashMap<String, String>();
+    additionalConf.put("test.key.2", "key2");
+    additionalConf.put("io.shouldExist", "io");
+    additionalConf.put(TezRuntimeConfiguration.TEZ_RUNTIME_INTERNAL_SORTER_CLASS, "TestInternalSorter");
+    OrderedPartitionedKVOutputConfig.Builder builder =
+        OrderedPartitionedKVOutputConfig.newBuilder("KEY", "VALUE", "PARTITIONER", null)
+            .setKeyComparatorClass("KEY_COMPARATOR", null)
+            .setCompression(true, "CustomCodec", null)
+            .setSortBufferSize(2048)
+            .setAdditionalConfiguration("test.key.1", "key1")
+            .setAdditionalConfiguration("file.shouldExist", "file")
+            .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+                String.valueOf(false))
+            .setAdditionalConfiguration(additionalConf)
+            .setFromConfiguration(fromConf);
+
+    OrderedPartitionedKVOutputConfig configuration = builder.build();
+
+    OrderedPartitionedKVOutputConfig rebuilt = new OrderedPartitionedKVOutputConfig();
+    rebuilt.fromUserPayload(configuration.toUserPayload());
+
+    Configuration conf = rebuilt.conf;
+
+    // Verify programmatic API usage
+    assertEquals(2048, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_MB, 0));
+    assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
+    assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
+    assertEquals("PARTITIONER", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, ""));
+    assertEquals("CustomCodec",
+        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+    assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
+        false));
+    assertEquals("KEY_COMPARATOR", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_COMPARATOR_CLASS));
+
+    // Verify additional configs
+    assertEquals(false, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
+    assertEquals(1111, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT));
+    assertEquals("TestInternalSorter",
+        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_INTERNAL_SORTER_CLASS, ""));
+    assertEquals("io", conf.get("io.shouldExist"));
+    assertEquals("file", conf.get("file.shouldExist"));
+    assertEquals("fs", conf.get("fs.shouldExist"));
+    assertNull(conf.get("test.conf.key.1"));
+    assertNull(conf.get("test.key.1"));
+    assertNull(conf.get("test.key.2"));
+  }
+
+  @Test
+  public void testDefaultConfigsUsed() {
+    OrderedPartitionedKVOutputConfig.Builder builder =
+        OrderedPartitionedKVOutputConfig.newBuilder("KEY", "VALUE", "PARTITIONER", null);
+    OrderedPartitionedKVOutputConfig configuration = builder.build();
+
+    OrderedPartitionedKVOutputConfig rebuilt = new OrderedPartitionedKVOutputConfig();
+    rebuilt.fromUserPayload(configuration.toUserPayload());
+
+    Configuration conf = rebuilt.conf;
+
+    assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
+
+    // Property present
+    assertEquals("TestCodec",
+        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+
+    // Verify whatever was configured
+    assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
+    assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
+    assertEquals("PARTITIONER", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_PARTITIONER_CLASS, ""));
+  }
+
+  @Test
+  public void testPartitionerConfigs() {
+    Map<String, String> partitionerConf = Maps.newHashMap();
+    partitionerConf.put("partitioner.test.key", "PARTITIONERKEY");
+    partitionerConf
+        .put(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "InvalidKeyOverride");
+    OrderedPartitionedKVOutputConfig.Builder builder =
+        OrderedPartitionedKVOutputConfig
+            .newBuilder("KEY", "VALUE", "PARTITIONER", partitionerConf);
+
+    OrderedPartitionedKVOutputConfig configuration = builder.build();
+
+    OrderedPartitionedKVOutputConfig rebuilt = new OrderedPartitionedKVOutputConfig();
+    rebuilt.fromUserPayload(configuration.toUserPayload());
+
+    Configuration conf = rebuilt.conf;
+
+    // Default Output property should not be overridden based on partitioner config
+    assertEquals("TestCodec",
+        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+
+    assertEquals("PARTITIONERKEY", conf.get("partitioner.test.key"));
+  }
+
+  @Test
+  public void testCombinerConfigs() {
+    Map<String, String> combinerConf = Maps.newHashMap();
+    combinerConf.put("combiner.test.key", "COMBINERKEY");
+    combinerConf
+        .put(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "InvalidKeyOverride");
+    OrderedPartitionedKVOutputConfig.Builder builder =
+        OrderedPartitionedKVOutputConfig.newBuilder("KEY", "VALUE", "PARTITIONER", null)
+            .setCombiner("COMBINER", combinerConf);
+
+    OrderedPartitionedKVOutputConfig configuration = builder.build();
+
+    OrderedPartitionedKVOutputConfig rebuilt = new OrderedPartitionedKVOutputConfig();
+    rebuilt.fromUserPayload(configuration.toUserPayload());
+
+    Configuration conf = rebuilt.conf;
+
+    // Default Output property should not be overridden based on partitioner config
+    assertEquals("TestCodec",
+        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+
+    assertEquals("COMBINERKEY", conf.get("combiner.test.key"));
+  }
+}

http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestShuffledMergedInputConfigurer.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestShuffledMergedInputConfigurer.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestShuffledMergedInputConfigurer.java
deleted file mode 100644
index a805601..0000000
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestShuffledMergedInputConfigurer.java
+++ /dev/null
@@ -1,160 +0,0 @@
-/*
- * *
- *  * 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.tez.runtime.library.conf;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import com.google.common.collect.Maps;
-import org.apache.hadoop.conf.Configuration;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
-import org.junit.Test;
-
-public class TestShuffledMergedInputConfigurer {
-
-  @Test
-  public void testNullParams() {
-    try {
-      OrderedGroupedKVInputConfigurer.newBuilder(null, "VALUE");
-      fail("Expecting a null parameter list to fail");
-    } catch (NullPointerException npe) {
-      assertTrue(npe.getMessage().contains("cannot be null"));
-    }
-
-    try {
-      OrderedGroupedKVInputConfigurer.newBuilder("KEY", null);
-      fail("Expecting a null parameter list to fail");
-    } catch (NullPointerException npe) {
-      assertTrue(npe.getMessage().contains("cannot be null"));
-    }
-  }
-
-  @Test
-  public void testSetters() {
-    Configuration fromConf = new Configuration(false);
-    fromConf.set("test.conf.key.1", "confkey1");
-    fromConf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 1111);
-    fromConf.set("io.shouldExist", "io");
-    Map<String, String> additionalConf = new HashMap<String, String>();
-    additionalConf.put("test.key.2", "key2");
-    additionalConf.put(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR, "3");
-    additionalConf.put("file.shouldExist", "file");
-    OrderedGroupedKVInputConfigurer.Builder builder =
-        OrderedGroupedKVInputConfigurer.newBuilder("KEY", "VALUE")
-            .setKeyComparatorClass("KEY_COMPARATOR", null)
-            .setCompression(true, "CustomCodec", null)
-            .setMaxSingleMemorySegmentFraction(0.11f)
-            .setMergeFraction(0.22f)
-            .setPostMergeBufferFraction(0.33f)
-            .setShuffleBufferFraction(0.44f)
-            .setAdditionalConfiguration("fs.shouldExist", "fs")
-            .setAdditionalConfiguration("test.key.1", "key1")
-            .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
-                String.valueOf(false))
-            .setAdditionalConfiguration(additionalConf)
-            .setFromConfiguration(fromConf);
-
-    OrderedGroupedKVInputConfigurer configuration = builder.build();
-
-
-    OrderedGroupedKVInputConfigurer rebuilt = new OrderedGroupedKVInputConfigurer();
-    rebuilt.fromUserPayload(configuration.toUserPayload());
-
-    Configuration conf = rebuilt.conf;
-
-    // Verify programmatic API usage
-    assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
-    assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
-    assertEquals("CustomCodec",
-        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-    assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
-        false));
-    assertEquals(0.11f, conf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT, 0.0f), 0.001f);
-    assertEquals(0.22f, conf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, 0.0f), 0.001f);
-    assertEquals(0.33f, conf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_INPUT_BUFFER_PERCENT, 0.0f), 0.001f);
-    assertEquals(0.44f, conf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, 0.00f), 0.001f);
-
-    // Verify additional configs
-    assertEquals(false, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
-        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
-    assertEquals(1111, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES,
-        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT));
-    assertEquals(3, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR, -1));
-    assertEquals("io", conf.get("io.shouldExist"));
-    assertEquals("file", conf.get("file.shouldExist"));
-    assertEquals("fs", conf.get("fs.shouldExist"));
-    assertNull(conf.get("test.conf.key.1"));
-    assertNull(conf.get("test.key.1"));
-    assertNull(conf.get("test.key.2"));
-  }
-
-  @Test
-  public void testDefaultConfigsUsed() {
-    OrderedGroupedKVInputConfigurer.Builder builder =
-        OrderedGroupedKVInputConfigurer.newBuilder("KEY", "VALUE");
-    OrderedGroupedKVInputConfigurer configuration = builder.build();
-
-    OrderedGroupedKVInputConfigurer rebuilt = new OrderedGroupedKVInputConfigurer();
-    rebuilt.fromUserPayload(configuration.toUserPayload());
-
-    Configuration conf = rebuilt.conf;
-
-    assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
-        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
-
-    // Default property present.
-    assertEquals("TestCodec",
-        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-
-    // Verify whatever was configured
-    assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
-    assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
-  }
-
-  @Test
-  public void testCombinerConfigs() {
-    Map<String, String> combinerConf = Maps.newHashMap();
-    combinerConf.put("combiner.test.key", "COMBINERKEY");
-    combinerConf
-        .put(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "InvalidKeyOverride");
-    OrderedGroupedKVInputConfigurer.Builder builder =
-        OrderedGroupedKVInputConfigurer.newBuilder("KEY", "VALUE")
-            .setCombiner("COMBINER", combinerConf);
-
-    OrderedGroupedKVInputConfigurer configuration = builder.build();
-
-    OrderedGroupedKVInputConfigurer rebuilt = new OrderedGroupedKVInputConfigurer();
-    rebuilt.fromUserPayload(configuration.toUserPayload());
-
-    Configuration conf = rebuilt.conf;
-
-    // Default Output property should not be overridden based on partitioner config
-    assertEquals("TestCodec",
-        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-
-    assertEquals("COMBINERKEY", conf.get("combiner.test.key"));
-  }
-}

http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestShuffledUnorderedKVInputConfigurer.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestShuffledUnorderedKVInputConfigurer.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestShuffledUnorderedKVInputConfigurer.java
deleted file mode 100644
index d4324d3..0000000
--- a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestShuffledUnorderedKVInputConfigurer.java
+++ /dev/null
@@ -1,131 +0,0 @@
-/*
- * *
- *  * 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.tez.runtime.library.conf;
-
-import static org.junit.Assert.assertEquals;
-import static org.junit.Assert.assertNull;
-import static org.junit.Assert.assertTrue;
-import static org.junit.Assert.fail;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.hadoop.conf.Configuration;
-import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
-import org.junit.Test;
-
-public class TestShuffledUnorderedKVInputConfigurer {
-
-  @Test
-  public void testNullParams() {
-    try {
-      UnorderedKVInputConfigurer.newBuilder(null, "VALUE");
-      fail("Expecting a null parameter list to fail");
-    } catch (NullPointerException npe) {
-      assertTrue(npe.getMessage().contains("cannot be null"));
-    }
-
-    try {
-      UnorderedKVInputConfigurer.newBuilder("KEY", null);
-      fail("Expecting a null parameter list to fail");
-    } catch (NullPointerException npe) {
-      assertTrue(npe.getMessage().contains("cannot be null"));
-    }
-  }
-
-  @Test
-  public void testSetters() {
-    Configuration fromConf = new Configuration(false);
-    fromConf.set("test.conf.key.1", "confkey1");
-    fromConf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 1111);
-    fromConf.set("io.shouldExist", "io");
-    Map<String, String> additionalConf = new HashMap<String, String>();
-    additionalConf.put("test.key.2", "key2");
-    additionalConf.put(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR, "3");
-    additionalConf.put("file.shouldExist", "file");
-    UnorderedKVInputConfigurer.Builder builder =
-        UnorderedKVInputConfigurer.newBuilder("KEY", "VALUE")
-            .setCompression(true, "CustomCodec", null)
-            .setMaxSingleMemorySegmentFraction(0.11f)
-            .setMergeFraction(0.22f)
-            .setShuffleBufferFraction(0.33f)
-            .setAdditionalConfiguration("fs.shouldExist", "fs")
-            .setAdditionalConfiguration("test.key.1", "key1")
-            .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
-                String.valueOf(false))
-            .setAdditionalConfiguration(additionalConf)
-            .setFromConfiguration(fromConf);
-
-    UnorderedKVInputConfigurer configuration = builder.build();
-
-    UnorderedKVInputConfigurer rebuilt = new UnorderedKVInputConfigurer();
-    rebuilt.fromUserPayload(configuration.toUserPayload());
-
-    Configuration conf = rebuilt.conf;
-
-    // Verify programmatic API usage
-    assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
-    assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
-    assertEquals("CustomCodec",
-        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-    assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
-        false));
-    assertEquals(0.11f, conf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT, 0.0f), 0.001f);
-    assertEquals(0.22f, conf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, 0.0f), 0.001f);
-    assertEquals(0.33f, conf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, 0.00f), 0.001f);
-
-    // Verify additional configs
-    assertEquals(false, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
-        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
-    assertEquals(1111, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES,
-        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT));
-    assertEquals(3, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR, -1));
-    assertEquals("io", conf.get("io.shouldExist"));
-    assertEquals("file", conf.get("file.shouldExist"));
-    assertEquals("fs", conf.get("fs.shouldExist"));
-    assertNull(conf.get("test.conf.key.1"));
-    assertNull(conf.get("test.key.1"));
-    assertNull(conf.get("test.key.2"));
-  }
-
-  @Test
-  public void testDefaultConfigsUsed() {
-    UnorderedKVInputConfigurer.Builder builder =
-        UnorderedKVInputConfigurer.newBuilder("KEY", "VALUE");
-    UnorderedKVInputConfigurer configuration = builder.build();
-
-    UnorderedKVInputConfigurer rebuilt = new UnorderedKVInputConfigurer();
-    rebuilt.fromUserPayload(configuration.toUserPayload());
-
-    Configuration conf = rebuilt.conf;
-
-    assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
-        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
-
-    // Default property present.
-    assertEquals("TestCodec",
-        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
-
-    // Verify whatever was configured
-    assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
-    assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
-  }
-}

http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestUnorderedKVEdgeConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestUnorderedKVEdgeConfig.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestUnorderedKVEdgeConfig.java
new file mode 100644
index 0000000..a25e5cf
--- /dev/null
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestUnorderedKVEdgeConfig.java
@@ -0,0 +1,173 @@
+/*
+ * *
+ *  * 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.tez.runtime.library.conf;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.CommonConfigurationKeys;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.junit.Test;
+
+public class TestUnorderedKVEdgeConfig {
+
+  @Test
+  public void testNullParams() {
+    try {
+      UnorderedKVEdgeConfig.newBuilder(null, "VALUE");
+      fail("Expecting a null parameter list to fail");
+    } catch (NullPointerException npe) {
+      assertTrue(npe.getMessage().contains("cannot be null"));
+    }
+
+    try {
+      UnorderedKVEdgeConfig.newBuilder("KEY", null);
+      fail("Expecting a null parameter list to fail");
+    } catch (NullPointerException npe) {
+      assertTrue(npe.getMessage().contains("cannot be null"));
+    }
+  }
+
+  @Test
+  public void testDefaultConfigsUsed() {
+    UnorderedKVEdgeConfig.Builder builder =
+        UnorderedKVEdgeConfig.newBuilder("KEY", "VALUE");
+    builder.setKeySerializationClass("SerClass1", null).setValueSerializationClass("SerClass2", null);
+
+    UnorderedKVEdgeConfig configuration = builder.build();
+
+
+    UnorderedKVOutputConfig rebuiltOutput =
+        new UnorderedKVOutputConfig();
+    rebuiltOutput.fromUserPayload(configuration.getOutputPayload());
+    UnorderedKVInputConfig rebuiltInput =
+        new UnorderedKVInputConfig();
+    rebuiltInput.fromUserPayload(configuration.getInputPayload());
+
+    Configuration outputConf = rebuiltOutput.conf;
+    assertEquals(true, outputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
+    assertEquals("TestCodec",
+        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+    assertTrue(outputConf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY).startsWith
+        ("SerClass2,SerClass1"));
+
+    Configuration inputConf = rebuiltInput.conf;
+    assertEquals(true, inputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
+    assertEquals("TestCodec",
+        inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+    assertTrue(inputConf.get(CommonConfigurationKeys.IO_SERIALIZATIONS_KEY).startsWith
+        ("SerClass2,SerClass1"));
+  }
+
+  @Test
+  public void testSpecificIOConfs() {
+    // Ensures that Output and Input confs are not mixed.
+    UnorderedKVEdgeConfig.Builder builder =
+        UnorderedKVEdgeConfig.newBuilder("KEY", "VALUE");
+
+    UnorderedKVEdgeConfig configuration = builder.build();
+
+    UnorderedKVOutputConfig rebuiltOutput =
+        new UnorderedKVOutputConfig();
+    rebuiltOutput.fromUserPayload(configuration.getOutputPayload());
+    UnorderedKVInputConfig rebuiltInput =
+        new UnorderedKVInputConfig();
+    rebuiltInput.fromUserPayload(configuration.getInputPayload());
+
+    Configuration outputConf = rebuiltOutput.conf;
+    assertEquals("TestCodec",
+        outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "DEFAULT"));
+
+    Configuration inputConf = rebuiltInput.conf;
+    assertEquals("TestCodec",
+        inputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, "DEFAULT"));
+  }
+
+  @Test
+  public void tetCommonConf() {
+
+    Configuration fromConf = new Configuration(false);
+    fromConf.set("test.conf.key.1", "confkey1");
+    fromConf.setBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD, false);
+    fromConf.setFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, 0.11f);
+    fromConf.set("io.shouldExist", "io");
+    Map<String, String> additionalConfs = new HashMap<String, String>();
+    additionalConfs.put("test.key.2", "key2");
+    additionalConfs.put(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, "1111");
+    additionalConfs.put(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT, "0.22f");
+    additionalConfs.put("file.shouldExist", "file");
+
+    UnorderedKVEdgeConfig.Builder builder = UnorderedKVEdgeConfig
+        .newBuilder("KEY",
+            "VALUE")
+        .setAdditionalConfiguration("fs.shouldExist", "fs")
+        .setAdditionalConfiguration("test.key.1", "key1")
+        .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE, "3333")
+        .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, "0.33f")
+        .setAdditionalConfiguration(additionalConfs)
+        .setFromConfiguration(fromConf);
+
+    UnorderedKVEdgeConfig configuration = builder.build();
+
+    UnorderedKVOutputConfig rebuiltOutput =
+        new UnorderedKVOutputConfig();
+    rebuiltOutput.fromUserPayload(configuration.getOutputPayload());
+    UnorderedKVInputConfig rebuiltInput =
+        new UnorderedKVInputConfig();
+    rebuiltInput.fromUserPayload(configuration.getInputPayload());
+
+    Configuration outputConf = rebuiltOutput.conf;
+    Configuration inputConf = rebuiltInput.conf;
+
+    assertEquals(false, outputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD, true));
+    assertEquals(1111, outputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 0));
+    assertEquals(3333, outputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE, 0));
+    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT));
+    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT));
+    assertNull(outputConf.get(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT));
+    assertEquals("io", outputConf.get("io.shouldExist"));
+    assertEquals("file", outputConf.get("file.shouldExist"));
+    assertEquals("fs", outputConf.get("fs.shouldExist"));
+
+
+    assertEquals(false, inputConf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD, true));
+    assertEquals(1111, inputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 0));
+    assertEquals(3333, inputConf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_FILE_BUFFER_SIZE, 0));
+    assertEquals(0.11f,
+        inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, 0.0f), 0.001f);
+    assertEquals(0.22f,
+        inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT, 0.0f), 0.001f);
+    assertEquals(0.33f, inputConf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, 0.0f),
+        0.001f);
+    assertEquals("io", inputConf.get("io.shouldExist"));
+    assertEquals("file", inputConf.get("file.shouldExist"));
+    assertEquals("fs", inputConf.get("fs.shouldExist"));
+
+  }
+}

http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestUnorderedKVInputConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestUnorderedKVInputConfig.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestUnorderedKVInputConfig.java
new file mode 100644
index 0000000..b5590f5
--- /dev/null
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestUnorderedKVInputConfig.java
@@ -0,0 +1,131 @@
+/*
+ * *
+ *  * 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.tez.runtime.library.conf;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.junit.Test;
+
+public class TestUnorderedKVInputConfig {
+
+  @Test
+  public void testNullParams() {
+    try {
+      UnorderedKVInputConfig.newBuilder(null, "VALUE");
+      fail("Expecting a null parameter list to fail");
+    } catch (NullPointerException npe) {
+      assertTrue(npe.getMessage().contains("cannot be null"));
+    }
+
+    try {
+      UnorderedKVInputConfig.newBuilder("KEY", null);
+      fail("Expecting a null parameter list to fail");
+    } catch (NullPointerException npe) {
+      assertTrue(npe.getMessage().contains("cannot be null"));
+    }
+  }
+
+  @Test
+  public void testSetters() {
+    Configuration fromConf = new Configuration(false);
+    fromConf.set("test.conf.key.1", "confkey1");
+    fromConf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 1111);
+    fromConf.set("io.shouldExist", "io");
+    Map<String, String> additionalConf = new HashMap<String, String>();
+    additionalConf.put("test.key.2", "key2");
+    additionalConf.put(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR, "3");
+    additionalConf.put("file.shouldExist", "file");
+    UnorderedKVInputConfig.Builder builder =
+        UnorderedKVInputConfig.newBuilder("KEY", "VALUE")
+            .setCompression(true, "CustomCodec", null)
+            .setMaxSingleMemorySegmentFraction(0.11f)
+            .setMergeFraction(0.22f)
+            .setShuffleBufferFraction(0.33f)
+            .setAdditionalConfiguration("fs.shouldExist", "fs")
+            .setAdditionalConfiguration("test.key.1", "key1")
+            .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+                String.valueOf(false))
+            .setAdditionalConfiguration(additionalConf)
+            .setFromConfiguration(fromConf);
+
+    UnorderedKVInputConfig configuration = builder.build();
+
+    UnorderedKVInputConfig rebuilt = new UnorderedKVInputConfig();
+    rebuilt.fromUserPayload(configuration.toUserPayload());
+
+    Configuration conf = rebuilt.conf;
+
+    // Verify programmatic API usage
+    assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
+    assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
+    assertEquals("CustomCodec",
+        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+    assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
+        false));
+    assertEquals(0.11f, conf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MEMORY_LIMIT_PERCENT, 0.0f), 0.001f);
+    assertEquals(0.22f, conf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_MERGE_PERCENT, 0.0f), 0.001f);
+    assertEquals(0.33f, conf.getFloat(TezRuntimeConfiguration.TEZ_RUNTIME_SHUFFLE_INPUT_BUFFER_PERCENT, 0.00f), 0.001f);
+
+    // Verify additional configs
+    assertEquals(false, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
+    assertEquals(1111, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT));
+    assertEquals(3, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IO_SORT_FACTOR, -1));
+    assertEquals("io", conf.get("io.shouldExist"));
+    assertEquals("file", conf.get("file.shouldExist"));
+    assertEquals("fs", conf.get("fs.shouldExist"));
+    assertNull(conf.get("test.conf.key.1"));
+    assertNull(conf.get("test.key.1"));
+    assertNull(conf.get("test.key.2"));
+  }
+
+  @Test
+  public void testDefaultConfigsUsed() {
+    UnorderedKVInputConfig.Builder builder =
+        UnorderedKVInputConfig.newBuilder("KEY", "VALUE");
+    UnorderedKVInputConfig configuration = builder.build();
+
+    UnorderedKVInputConfig rebuilt = new UnorderedKVInputConfig();
+    rebuilt.fromUserPayload(configuration.toUserPayload());
+
+    Configuration conf = rebuilt.conf;
+
+    assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
+
+    // Default property present.
+    assertEquals("TestCodec",
+        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+
+    // Verify whatever was configured
+    assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
+    assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
+  }
+}

http://git-wip-us.apache.org/repos/asf/tez/blob/b526ed5c/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestUnorderedKVOutputConfig.java
----------------------------------------------------------------------
diff --git a/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestUnorderedKVOutputConfig.java b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestUnorderedKVOutputConfig.java
new file mode 100644
index 0000000..8588ef1
--- /dev/null
+++ b/tez-runtime-library/src/test/java/org/apache/tez/runtime/library/conf/TestUnorderedKVOutputConfig.java
@@ -0,0 +1,129 @@
+/*
+ * *
+ *  * 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.tez.runtime.library.conf;
+
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertNull;
+import static org.junit.Assert.assertTrue;
+import static org.junit.Assert.fail;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.tez.runtime.library.api.TezRuntimeConfiguration;
+import org.junit.Test;
+
+public class TestUnorderedKVOutputConfig {
+
+  @Test
+  public void testNullParams() {
+    try {
+      UnorderedKVOutputConfig.newBuilder(
+          null, "VALUE");
+      fail("Expecting a null parameter list to fail");
+    } catch (NullPointerException npe) {
+      assertTrue(npe.getMessage().contains("cannot be null"));
+    }
+
+    try {
+      UnorderedKVOutputConfig.newBuilder(
+          "KEY", null);
+      fail("Expecting a null parameter list to fail");
+    } catch (NullPointerException npe) {
+      assertTrue(npe.getMessage().contains("cannot be null"));
+    }
+  }
+
+  @Test
+  public void testSetters() {
+    Configuration fromConf = new Configuration(false);
+    fromConf.set("test.conf.key.1", "confkey1");
+    fromConf.setInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES, 1111);
+    fromConf.set("io.shouldExist", "io");
+    Map<String, String> additionalConf = new HashMap<String, String>();
+    additionalConf.put("test.key.2", "key2");
+    additionalConf.put(TezRuntimeConfiguration.TEZ_RUNTIME_UNORDERED_OUTPUT_MAX_PER_BUFFER_SIZE_BYTES, "2222");
+    additionalConf.put("file.shouldExist", "file");
+    UnorderedKVOutputConfig.Builder builder =
+        UnorderedKVOutputConfig.newBuilder("KEY", "VALUE")
+            .setCompression(true, "CustomCodec", null)
+            .setAdditionalConfiguration("fs.shouldExist", "fs")
+            .setAdditionalConfiguration("test.key.1", "key1")
+            .setAdditionalConfiguration(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+                String.valueOf(false))
+            .setAdditionalConfiguration(additionalConf)
+            .setFromConfiguration(fromConf);
+
+    UnorderedKVOutputConfig configuration = builder.build();
+
+    UnorderedKVOutputConfig rebuilt =
+        new UnorderedKVOutputConfig();
+    rebuilt.fromUserPayload(configuration.toUserPayload());
+
+    Configuration conf = rebuilt.conf;
+
+    // Verify programmatic API usage
+    assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
+    assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
+    assertEquals("CustomCodec",
+        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+    assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS,
+        false));
+
+    // Verify additional configs
+    assertEquals(false, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
+    assertEquals(1111, conf.getInt(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_BYTES_DEFAULT));
+    assertEquals("io", conf.get("io.shouldExist"));
+    assertEquals("file", conf.get("file.shouldExist"));
+    assertEquals("fs", conf.get("fs.shouldExist"));
+    assertNull(conf.get("test.conf.key.1"));
+    assertNull(conf.get("test.key.1"));
+    assertNull(conf.get("test.key.2"));
+  }
+
+  @Test
+  public void testDefaultConfigsUsed() {
+    UnorderedKVOutputConfig.Builder builder =
+        UnorderedKVOutputConfig
+            .newBuilder("KEY", "VALUE");
+    UnorderedKVOutputConfig configuration = builder.build();
+
+    UnorderedKVOutputConfig rebuilt =
+        new UnorderedKVOutputConfig();
+    rebuilt.fromUserPayload(configuration.toUserPayload());
+
+    Configuration conf = rebuilt.conf;
+
+    assertEquals(true, conf.getBoolean(TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD,
+        TezRuntimeConfiguration.TEZ_RUNTIME_IFILE_READAHEAD_DEFAULT));
+
+    // Default property present.
+    assertEquals("TestCodec",
+        conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_COMPRESS_CODEC, ""));
+
+    // Verify whatever was configured
+    assertEquals("KEY", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_KEY_CLASS, ""));
+    assertEquals("VALUE", conf.get(TezRuntimeConfiguration.TEZ_RUNTIME_VALUE_CLASS, ""));
+  }
+}


Mime
View raw message