{"action":"created","comment":{"url":"https://api.github.com/repos/ssisk/beam/pulls/comments/96997210","pull_request_review_id":17582215,"id":96997210,"diff_hunk":"@@ -0,0 +1,776 @@\n+/*\n+ * Licensed to the Apache Software Foundation (ASF) under one or more contributor license\n+ * agreements. See the NOTICE file distributed with this work for additional information regarding\n+ * copyright ownership. The ASF licenses this file to you under the Apache License, Version 2.0 (the\n+ * \"License\"); you may not use this file except in compliance with the License. You may obtain a\n+ * copy of the License at\n+ *\n+ * http:www.apache.org/licenses/LICENSE-2.0\n+ *\n+ * Unless required by applicable law or agreed to in writing, software distributed under the License\n+ * is distributed on an \"AS IS\" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express\n+ * or implied. See the License for the specific language governing permissions and limitations under\n+ * the License.\n+ */\n+package org.apache.beam.sdk.io.hadoop.inputformat.unit.tests;\n+\n+import static org.apache.beam.sdk.transforms.display.DisplayDataMatchers.hasDisplayItem;\n+import static org.hamcrest.Matchers.containsInAnyOrder;\n+import static org.junit.Assert.assertEquals;\n+import static org.junit.Assert.assertThat;\n+\n+import java.io.IOException;\n+import java.util.ArrayList;\n+import java.util.Iterator;\n+import java.util.List;\n+import java.util.Map.Entry;\n+\n+import org.apache.beam.sdk.coders.AvroCoder;\n+import org.apache.beam.sdk.coders.StringUtf8Coder;\n+import org.apache.beam.sdk.io.BoundedSource;\n+import org.apache.beam.sdk.io.BoundedSource.BoundedReader;\n+import org.apache.beam.sdk.io.hadoop.inputformat.HadoopInputFormatIO;\n+import org.apache.beam.sdk.io.hadoop.inputformat.HadoopInputFormatIO.HadoopInputFormatBoundedSource;\n+import org.apache.beam.sdk.io.hadoop.inputformat.HadoopInputFormatIO.SerializableConfiguration;\n+import org.apache.beam.sdk.io.hadoop.inputformat.HadoopInputFormatIOContants;\n+import org.apache.beam.sdk.io.hadoop.inputformat.coders.WritableCoder;\n+import org.apache.beam.sdk.io.hadoop.inputformat.unit.tests.inputs.BadCreateReaderInputFormat;\n+import org.apache.beam.sdk.io.hadoop.inputformat.unit.tests.inputs.BadEmptySplitsInputFormat;\n+import org.apache.beam.sdk.io.hadoop.inputformat.unit.tests.inputs.BadGetSplitsInputFormat;\n+import org.apache.beam.sdk.io.hadoop.inputformat.unit.tests.inputs.BadNoRecordsInputFormat;\n+import org.apache.beam.sdk.io.hadoop.inputformat.unit.tests.inputs.BadNullCreateReaderInputFormat;\n+import org.apache.beam.sdk.io.hadoop.inputformat.unit.tests.inputs.BadNullSplitsInputFormat;\n+import org.apache.beam.sdk.io.hadoop.inputformat.unit.tests.inputs.ConfigurableEmployeeInputFormat;\n+import org.apache.beam.sdk.io.hadoop.inputformat.unit.tests.inputs.Employee;\n+import org.apache.beam.sdk.io.hadoop.inputformat.unit.tests.inputs.NewObjectsEmployeeInputFormat;\n+import org.apache.beam.sdk.io.hadoop.inputformat.unit.tests.inputs.ReuseObjectsEmployeeInputFormat;\n+import org.apache.beam.sdk.io.hadoop.inputformat.unit.tests.inputs.UnitTestUtils;\n+import org.apache.beam.sdk.testing.PAssert;\n+import org.apache.beam.sdk.testing.SourceTestUtils;\n+import org.apache.beam.sdk.testing.TestPipeline;\n+import org.apache.beam.sdk.transforms.SimpleFunction;\n+import org.apache.beam.sdk.transforms.display.DisplayData;\n+import org.apache.beam.sdk.values.KV;\n+import org.apache.beam.sdk.values.PBegin;\n+import org.apache.beam.sdk.values.PCollection;\n+import org.apache.hadoop.conf.Configuration;\n+import org.apache.hadoop.io.LongWritable;\n+import org.apache.hadoop.io.Text;\n+import org.apache.hadoop.mapreduce.InputFormat;\n+import org.junit.BeforeClass;\n+import org.junit.Rule;\n+import org.junit.Test;\n+import org.junit.rules.ExpectedException;\n+import org.junit.runner.RunWith;\n+import org.junit.runners.JUnit4;\n+\n+/**\n+ * Unit tests for {@link HadoopInputFormatIO}.\n+ */\n+@RunWith(JUnit4.class)\n+public class HadoopInputFormatIOTest {\n+  static SerializableConfiguration serConf;\n+  static SimpleFunction<Text, String> myKeyTranslate;\n+  static SimpleFunction<Employee, String> myValueTranslate;\n+\n+  @Rule public final transient TestPipeline p = TestPipeline.create();\n+  @Rule public ExpectedException thrown = ExpectedException.none();\n+\n+  private PBegin input = PBegin.in(p);\n+\n+  @BeforeClass\n+  public static void setUp() {\n+    serConf = loadTestConfiguration(\n+                  NewObjectsEmployeeInputFormat.class, \n+                  Text.class, \n+                  Employee.class);\n+    myKeyTranslate = new SimpleFunction<Text, String>() {\n+      private static final long serialVersionUID = 1L;\n+      @Override\n+      public String apply(Text input) {\n+        return input.toString();\n+      }\n+    };\n+    myValueTranslate = new SimpleFunction<Employee, String>() {\n+      private static final long serialVersionUID = 1L;\n+      @Override\n+      public String apply(Employee input) {\n+        return input.getEmpName() + \"_\" + input.getEmpAddress();\n+      }\n+    };\n+  }\n+\n+  @Test\n+  public void testReadBuildsCorrectly() {\n+    HadoopInputFormatIO.Read<String, String> read = HadoopInputFormatIO.<String, String>read()\n+        .withConfiguration(serConf.getHadoopConfiguration())\n+        .withKeyTranslation(myKeyTranslate)\n+        .withValueTranslation(myValueTranslate);\n+    assertEquals(serConf.getHadoopConfiguration(),\n+        read.getConfiguration().getHadoopConfiguration());\n+    assertEquals(myKeyTranslate, read.getKeyTranslationFunction());\n+    assertEquals(myValueTranslate, read.getValueTranslationFunction());\n+    assertEquals(myValueTranslate.getOutputTypeDescriptor(), read.getValueClass());\n+    assertEquals(myKeyTranslate.getOutputTypeDescriptor(), read.getKeyClass());\n+  }\n+\n+  @Test\n+  public void testReadBuildsCorrectlyInDifferentOrder() {\n+    HadoopInputFormatIO.Read<String, String> read =\n+        HadoopInputFormatIO.<String, String>read()\n+            .withValueTranslation(myValueTranslate)\n+            .withConfiguration(serConf.getHadoopConfiguration())\n+            .withKeyTranslation(myKeyTranslate);\n+    assertEquals(serConf.getHadoopConfiguration(),\n+        read.getConfiguration().getHadoopConfiguration());\n+    assertEquals(myKeyTranslate, read.getKeyTranslationFunction());\n+    assertEquals(myValueTranslate, read.getValueTranslationFunction());\n+    assertEquals(myKeyTranslate.getOutputTypeDescriptor(), read.getKeyClass());\n+    assertEquals(myValueTranslate.getOutputTypeDescriptor(), read.getValueClass());\n+  }\n+\n+  @Test\n+  public void testReadDisplayData() {\n+    HadoopInputFormatIO.Read<String, String> read = HadoopInputFormatIO.<String, String>read()\n+        .withConfiguration(serConf.getHadoopConfiguration())\n+        .withKeyTranslation(myKeyTranslate)\n+        .withValueTranslation(myValueTranslate);\n+    read.validate(input);\n+    DisplayData displayData = DisplayData.from(read);\n+    Iterator<Entry<String, String>> propertyElement = serConf.getHadoopConfiguration().iterator();\n+    while (propertyElement.hasNext()) {\n+      Entry<String, String> element = propertyElement.next();\n+      assertThat(displayData, hasDisplayItem(element.getKey(), element.getValue()));\n+    }\n+    assertThat(displayData,\n+        hasDisplayItem(\"KeyTranslationSimpleFunction\", myKeyTranslate.toString()));\n+    assertThat(displayData,\n+        hasDisplayItem(\"ValueTranslationSimpleFunction\", myValueTranslate.toString()));\n+  }\n+\n+  /**\n+   * This test validates Read transform object creation fails with null configuration.\n+   * withConfiguration() method checks configuration is null and throws exception if it is null.\n+   */\n+  @Test\n+  public void testReadObjectCreationFailsIfConfigurationIsNull() {\n+    thrown.expect(NullPointerException.class);\n+    thrown.expectMessage(HadoopInputFormatIOContants.NULL_CONFIGURATION_ERROR_MSG);\n+    HadoopInputFormatIO.<Text, Employee>read()\n+          .withConfiguration(null);\n+  }\n+\n+  /**\n+   * This test validates Read transform object creation with only configuration.\n+   */\n+  @Test\n+  public void testReadObjectCreationWithConfiguration() {\n+    HadoopInputFormatIO.Read<Text, Employee> read = HadoopInputFormatIO.<Text, Employee>read()\n+        .withConfiguration(serConf.getHadoopConfiguration());\n+    read.validate(input);\n+    assertEquals(serConf.getHadoopConfiguration(),\n+        read.getConfiguration().getHadoopConfiguration());\n+    assertEquals(null, read.getKeyTranslationFunction());\n+    assertEquals(null, read.getValueTranslationFunction());\n+    assertEquals(serConf.getHadoopConfiguration().getClass(HadoopInputFormatIOContants.KEY_CLASS,\n+        Object.class), read.getKeyClass().getRawType());\n+    assertEquals(serConf.getHadoopConfiguration().getClass(HadoopInputFormatIOContants.VALUE_CLASS,\n+        Object.class), read.getValueClass().getRawType());\n+\n+  }\n+\n+  /**\n+   * This test validates Read transform object creation fails with configuration and null key\n+   * translation. withKeyTranslation() checks keyTranslation is null and throws exception if it null\n+   * value is passed.\n+   */\n+  @Test\n+  public void testReadObjectCreationFailsIfKeyTranslationFunctionIsNull() {\n+    thrown.expect(NullPointerException.class);\n+    thrown.expectMessage(HadoopInputFormatIOContants.NULL_KEY_TRANSLATIONFUNC_ERROR_MSG);\n+    HadoopInputFormatIO.<String, Employee>read()\n+        .withConfiguration(serConf.getHadoopConfiguration())\n+        .withKeyTranslation(null);\n+  }\n+\n+  /**\n+   * This test validates Read transform object creation with configuration and key translation.\n+   */\n+  @Test\n+  public void testReadObjectCreationWithConfigurationKeyTranslation() {\n+    HadoopInputFormatIO.Read<String, Employee> read = HadoopInputFormatIO.<String, Employee>read()\n+        .withConfiguration(serConf.getHadoopConfiguration())\n+        .withKeyTranslation(myKeyTranslate);\n+    read.validate(input);\n+    assertEquals(serConf.getHadoopConfiguration(),\n+        read.getConfiguration().getHadoopConfiguration());\n+    assertEquals(myKeyTranslate, read.getKeyTranslationFunction());\n+    assertEquals(null, read.getValueTranslationFunction());\n+    assertEquals(myKeyTranslate.getOutputTypeDescriptor().getRawType(),\n+        read.getKeyClass().getRawType());\n+    assertEquals(serConf.getHadoopConfiguration().getClass(\"value.class\", Object.class),\n+        read.getValueClass().getRawType());\n+  }\n+\n+  /**\n+   * This test validates Read transform object creation fails with configuration and null value\n+   * translation. withValueTranslation() checks valueTranslation is null and throws exception if\n+   * null value is passed.\n+   */\n+  @Test\n+  public void testReadObjectCreationFailsIfValueTranslationFunctionIsNull() {\n+    thrown.expect(NullPointerException.class);\n+    thrown.expectMessage(HadoopInputFormatIOContants.NULL_VALUE_TRANSLATIONFUNC_ERROR_MSG);\n+    HadoopInputFormatIO.<Text, String>read()\n+        .withConfiguration(serConf.getHadoopConfiguration())\n+        .withValueTranslation(null);\n+  }\n+\n+  /**\n+   * This test validates Read transform object creation with configuration and value translation.\n+   */\n+  @Test\n+  public void testReadObjectCreationWithConfigurationValueTranslation() {\n+    HadoopInputFormatIO.Read<Text, String> read = HadoopInputFormatIO.<Text, String>read()\n+        .withConfiguration(serConf.getHadoopConfiguration())\n+        .withValueTranslation(myValueTranslate);\n+    read.validate(input);\n+    assertEquals(serConf.getHadoopConfiguration(),\n+        read.getConfiguration().getHadoopConfiguration());\n+    assertEquals(null, read.getKeyTranslationFunction());\n+    assertEquals(myValueTranslate, read.getValueTranslationFunction());\n+    assertEquals(serConf.getHadoopConfiguration().getClass(\"key.class\", Object.class),\n+        read.getKeyClass().getRawType());\n+    assertEquals(myValueTranslate.getOutputTypeDescriptor().getRawType(),\n+        read.getValueClass().getRawType());\n+  }\n+\n+  /**\n+   * This test validates Read transform object creation with configuration , key translation and\n+   * value translation.\n+   */\n+  @Test\n+  public void testReadObjectCreationWithConfigurationKeyTranslationValueTranslation() {\n+    HadoopInputFormatIO.Read<String, String> read = HadoopInputFormatIO.<String, String>read()\n+        .withConfiguration(serConf.getHadoopConfiguration())\n+        .withKeyTranslation(myKeyTranslate)\n+        .withValueTranslation(myValueTranslate);\n+    read.validate(input);\n+    assertEquals(serConf.getHadoopConfiguration(),\n+        read.getConfiguration().getHadoopConfiguration());\n+    assertEquals(myKeyTranslate, read.getKeyTranslationFunction());\n+    assertEquals(myValueTranslate, read.getValueTranslationFunction());\n+    assertEquals(myKeyTranslate.getOutputTypeDescriptor().getRawType(),\n+        read.getKeyClass().getRawType());\n+    assertEquals(myValueTranslate.getOutputTypeDescriptor().getRawType(),\n+        read.getValueClass().getRawType());\n+  }\n+\n+  /**\n+   * This test validates functionality of Read.validate() function when Read transform is created\n+   * without calling withConfiguration().\n+   */\n+  @Test\n+  public void testReadValidationFailsMissingConfiguration() {\n+    HadoopInputFormatIO.Read<String, String> read = HadoopInputFormatIO.<String, String>read();\n+    thrown.expect(NullPointerException.class);\n+    thrown.expectMessage(HadoopInputFormatIOContants.MISSING_CONFIGURATION_ERROR_MSG);\n+    read.validate(input);\n+  }\n+\n+  /**\n+   * This test validates functionality of withConfiguration() function when Hadoop InputFormat class\n+   * is not provided by the user in configuration.\n+   */\n+  @Test\n+  public void testReadValidationFailsMissingInputFormatInConf() {\n+    Configuration configuration = new Configuration();\n+    configuration.setClass(HadoopInputFormatIOContants.KEY_CLASS, Text.class, Object.class);\n+    configuration.setClass(HadoopInputFormatIOContants.VALUE_CLASS, Employee.class, Object.class);\n+    thrown.expect(NullPointerException.class);\n+    thrown.expectMessage(HadoopInputFormatIOContants.MISSING_INPUTFORMAT_ERROR_MSG);\n+    HadoopInputFormatIO.Read<Text, Employee> read = HadoopInputFormatIO.<Text, Employee>read()\n+        .withConfiguration(configuration);\n+  }\n+\n+  /**\n+   * This test validates functionality of withConfiguration() function when key class is not\n+   * provided by the user in configuration.\n+   */\n+  @Test\n+  public void testReadValidationFailsMissingKeyClassInConf() {\n+    Configuration configuration = new Configuration();\n+    configuration.setClass(HadoopInputFormatIOContants.INPUTFORMAT_CLASSNAME,\n+        NewObjectsEmployeeInputFormat.class, InputFormat.class);\n+    configuration.setClass(HadoopInputFormatIOContants.VALUE_CLASS, Employee.class, Object.class);\n+    thrown.expect(NullPointerException.class);\n+    thrown.expectMessage(HadoopInputFormatIOContants.MISSING_INPUTFORMAT_KEY_CLASS_ERROR_MSG);\n+    HadoopInputFormatIO.Read<Text, Employee> read = HadoopInputFormatIO.<Text, Employee>read()\n+        .withConfiguration(configuration);\n+  }\n+\n+  /**\n+   * This test validates functionality of withConfiguration() function when value class is not\n+   * provided by the user in configuration.\n+   */\n+  @Test\n+  public void testReadValidationFailsMissingValueClassInConf() {\n+    Configuration configuration = new Configuration();\n+    configuration.setClass(HadoopInputFormatIOContants.INPUTFORMAT_CLASSNAME,\n+        NewObjectsEmployeeInputFormat.class, InputFormat.class);\n+    configuration.setClass(HadoopInputFormatIOContants.KEY_CLASS, Text.class, Object.class);\n+    thrown.expect(NullPointerException.class);\n+    thrown.expectMessage(HadoopInputFormatIOContants.MISSING_INPUTFORMAT_VALUE_CLASS_ERROR_MSG);\n+    HadoopInputFormatIO.Read<Text, Employee> read = HadoopInputFormatIO.<Text, Employee>read()\n+        .withConfiguration(configuration);\n+  }\n+\n+  /**\n+   * This test validates functionality of Read.validate() function when myKeyTranslate's (simple\n+   * function provided by user for key translation) input type is not same as Hadoop InputFormat's\n+   * keyClass(Which is property set in configuration as \"key.class\").\n+   */\n+  @Test\n+  public void testReadValidationFailsWithWrongInputTypeKeyTranslationFunction() {\n+    SimpleFunction<LongWritable, String> myKeyTranslateWithWrongInputType =\n+        new SimpleFunction<LongWritable, String>() {\n+          private static final long serialVersionUID = 1L;\n+          @Override\n+          public String apply(LongWritable input) {\n+            return input.toString();\n+          }\n+        };\n+    HadoopInputFormatIO.Read<String, Employee> read = HadoopInputFormatIO.<String, Employee>read()\n+        .withConfiguration(serConf.getHadoopConfiguration())\n+        .withKeyTranslation(myKeyTranslateWithWrongInputType);\n+    thrown.expect(IllegalArgumentException.class);\n+    thrown.expectMessage(\n+        String.format(HadoopInputFormatIOContants.WRONG_KEY_TRANSLATIONFUNC_ERROR_MSG,\n+            serConf.getHadoopConfiguration()\n+                .getClass(HadoopInputFormatIOContants.INPUTFORMAT_CLASSNAME, InputFormat.class),\n+            serConf.getHadoopConfiguration().getClass(HadoopInputFormatIOContants.KEY_CLASS,\n+                Object.class)));\n+    read.validate(input);\n+\n+  }\n+\n+  /**\n+   * This test validates functionality of Read.validate() function when myValueTranslate's (simple\n+   * function provided by user for value translation) input type is not same as Hadoop InputFormat's\n+   * valueClass(Which is property set in configuration as \"value.class\").\n+   */\n+  @Test\n+  public void testReadValidationFailsWithWrongInputTypeValueTranslationFunction() {\n+    SimpleFunction<LongWritable, String> myValueTranslateWithWrongInputType =\n+        new SimpleFunction<LongWritable, String>() {\n+          private static final long serialVersionUID = 1L;\n+          @Override\n+          public String apply(LongWritable input) {\n+            return input.toString();\n+          }\n+        };\n+    HadoopInputFormatIO.Read<Text, String> read = HadoopInputFormatIO.<Text, String>read()\n+            .withConfiguration(serConf.getHadoopConfiguration())\n+            .withValueTranslation(myValueTranslateWithWrongInputType);\n+    String expectedMessage =\n+        String.format(HadoopInputFormatIOContants.WRONG_VALUE_TRANSLATIONFUNC_ERROR_MSG,\n+            serConf.getHadoopConfiguration()\n+                .getClass(HadoopInputFormatIOContants.INPUTFORMAT_CLASSNAME, InputFormat.class),\n+            serConf.getHadoopConfiguration()\n+                .getClass(HadoopInputFormatIOContants.VALUE_CLASS, Object.class));\n+    thrown.expect(IllegalArgumentException.class);\n+    thrown.expectMessage(expectedMessage);\n+    read.validate(input);\n+  }\n+\n+  @Test\n+  public void testReadingData() throws Exception {\n+    HadoopInputFormatIO.Read<Text, Employee> read = HadoopInputFormatIO.<Text, Employee>read()\n+        .withConfiguration(serConf.getHadoopConfiguration());\n+    List<KV<Text, Employee>> expected = UnitTestUtils.getEmployeeData();\n+    PCollection<KV<Text, Employee>> actual = p.apply(\"ReadTest\", read);\n+    PAssert.that(actual).containsInAnyOrder(expected);\n+    p.run();\n+  }\n+\n+  /**\n+   * This test validates behavior of HadoopInputFormatSource if RecordReader object creation fails\n+   * in start() method.\n+   */\n+  @Test\n+  public void testReadersStartIfCreateRecordReaderFails() throws Exception {\n+    SerializableConfiguration serConf = \n+      loadTestConfiguration(\n+            BadCreateReaderInputFormat.class, \n+            Text.class, \n+            Employee.class);\n+    HadoopInputFormatBoundedSource<Text, Employee> parentHIFSource =\n+        new HadoopInputFormatBoundedSource<Text, Employee>(serConf, \n+            WritableCoder.of(Text.class),\n+            AvroCoder.of(Employee.class));\n+    List<BoundedSource<KV<Text, Employee>>> boundedSourceList =\n+        (List<BoundedSource<KV<Text, Employee>>>) parentHIFSource.splitIntoBundles(0,\n+            p.getOptions());\n+    for (BoundedSource<KV<Text, Employee>> source : boundedSourceList) {\n+      BoundedReader<KV<Text, Employee>> reader = source.createReader(p.getOptions());\n+      thrown.expect(Exception.class);\n+      thrown\n+          .expectMessage(\"Exception in creating RecordReader in BadCreateRecordReaderInputFormat\");\n+      reader.start();\n+    }\n+  }\n+\n+  /**\n+   * This test validates behavior of HadoopInputFormatSource if createRecordReader() of InputFormat\n+   * returns null.\n+   */\n+  @Test\n+  public void testReadersStartWithNullCreateRecordReader() throws Exception {\n+    SerializableConfiguration serConf =\n+      loadTestConfiguration(\n+          BadNullCreateReaderInputFormat.class, \n+          Text.class, \n+          Employee.class);\n+    HadoopInputFormatBoundedSource<Text, Employee> parentHIFSource =\n+        new HadoopInputFormatBoundedSource<Text, Employee>(\n+            serConf, \n+            WritableCoder.of(Text.class),\n+            AvroCoder.of(Employee.class));\n+    List<BoundedSource<KV<Text, Employee>>> boundedSourceList = \n+        (List<BoundedSource<KV<Text, Employee>>>) parentHIFSource\n+              .splitIntoBundles(0, p.getOptions());\n+    for (BoundedSource<KV<Text, Employee>> source : boundedSourceList) {\n+      BoundedReader<KV<Text, Employee>> reader = source.createReader(p.getOptions());\n+      thrown.expect(IOException.class);\n+      thrown.expectMessage(\n+          String.format(HadoopInputFormatIOContants.NULL_CREATE_RECORDREADER_ERROR_MSG,\n+              new BadNullCreateReaderInputFormat().getClass()));\n+      reader.start();\n+    }\n+  }\n+\n+  /**\n+   * This test validates behavior of createReader() and start() methods if InputFormat's getSplits()\n+   * returns InputSplitList having having no records.\n+   */\n+  @Test\n+  public void testReadersCreateReaderAndStartWithZeroRecords() throws Exception {\n+    SerializableConfiguration serConf = \n+     loadTestConfiguration(\n+        BadNoRecordsInputFormat.class,\n+        java.lang.String.class, \n+        java.lang.String.class);\n+    HadoopInputFormatBoundedSource<Text, Employee> parentHIFSource =\n+        new HadoopInputFormatBoundedSource<Text, Employee>(\n+            serConf, \n+            WritableCoder.of(Text.class),\n+            AvroCoder.of(Employee.class));\n+    List<BoundedSource<KV<Text, Employee>>> boundedSourceList =\n+        (List<BoundedSource<KV<Text, Employee>>>) parentHIFSource\n+              .splitIntoBundles(0, p.getOptions());\n+    for (BoundedSource<KV<Text, Employee>> source : boundedSourceList) {\n+      BoundedReader<KV<Text, Employee>> reader = source.createReader(p.getOptions());\n+      assertEquals(false, reader.start());\n+    }\n+  }\n+\n+  /**\n+   * This test validates fractions consumed while reading data.\n+   */\n+  @Test\n+  public void testReadersGetFractionConsumed() throws Exception {\n+    List<KV<Text, Employee>> referenceRecords = UnitTestUtils.getEmployeeData();\n+    HadoopInputFormatBoundedSource<Text, Employee> parentHIFSource =\n+        new HadoopInputFormatBoundedSource<Text, Employee>(\n+            serConf, \n+            WritableCoder.of(Text.class),\n+            AvroCoder.of(Employee.class));\n+    long estimatedSize = parentHIFSource.getEstimatedSizeBytes(p.getOptions());\n+    assertEquals(referenceRecords.size(), estimatedSize);\n+    List<BoundedSource<KV<Text, Employee>>> boundedSourceList =\n+        (List<BoundedSource<KV<Text, Employee>>>) parentHIFSource\n+            .splitIntoBundles(0, p.getOptions());\n+    assertEquals(UnitTestUtils.NUMBER_OF_SPLITS, boundedSourceList.size());\n+    List<KV<Text, Employee>> bundleRecords = new ArrayList<>();\n+    for (BoundedSource<KV<Text, Employee>> source : boundedSourceList) {\n+      List<KV<Text, Employee>> elements = new ArrayList<KV<Text, Employee>>();\n+      BoundedReader<KV<Text, Employee>> reader = source.createReader(p.getOptions());\n+      assertEquals(new Double((float) 0), reader.getFractionConsumed());\n+      int recordsRead = 0;","path":"sdks/java/io/hadoop-input-format/src/test/java/org/apache/beam/sdk/io/hadoop/inputformat/unit/tests/HadoopInputFormatIOTest.java","position":499,"original_position":499,"commit_id":"d66ee45403c38af1d59a289333bbdd96c7ac3a51","original_commit_id":"d66ee45403c38af1d59a289333bbdd96c7ac3a51","user":{"login":"ssisk","id":707137,"avatar_url":"https://avatars.githubusercontent.com/u/707137?v=3","gravatar_id":"","url":"https://api.github.com/users/ssisk","html_url":"https://github.com/ssisk","followers_url":"https://api.github.com/users/ssisk/followers","following_url":"https://api.github.com/users/ssisk/following{/other_user}","gists_url":"https://api.github.com/users/ssisk/gists{/gist_id}","starred_url":"https://api.github.com/users/ssisk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ssisk/subscriptions","organizations_url":"https://api.github.com/users/ssisk/orgs","repos_url":"https://api.github.com/users/ssisk/repos","events_url":"https://api.github.com/users/ssisk/events{/privacy}","received_events_url":"https://api.github.com/users/ssisk/received_events","type":"User","site_admin":false},"body":"should recordsRead be a float rather than an int?","created_at":"2017-01-20T01:27:22Z","updated_at":"2017-01-20T01:28:42Z","html_url":"https://github.com/ssisk/beam/pull/1#discussion_r96997210","pull_request_url":"https://api.github.com/repos/ssisk/beam/pulls/1","_links":{"self":{"href":"https://api.github.com/repos/ssisk/beam/pulls/comments/96997210"},"html":{"href":"https://github.com/ssisk/beam/pull/1#discussion_r96997210"},"pull_request":{"href":"https://api.github.com/repos/ssisk/beam/pulls/1"}}},"pull_request":{"url":"https://api.github.com/repos/ssisk/beam/pulls/1","id":100210814,"html_url":"https://github.com/ssisk/beam/pull/1","diff_url":"https://github.com/ssisk/beam/pull/1.diff","patch_url":"https://github.com/ssisk/beam/pull/1.patch","issue_url":"https://api.github.com/repos/ssisk/beam/issues/1","number":1,"state":"open","locked":false,"title":"First pass review for Hadoop changes. ","user":{"login":"ssisk","id":707137,"avatar_url":"https://avatars.githubusercontent.com/u/707137?v=3","gravatar_id":"","url":"https://api.github.com/users/ssisk","html_url":"https://github.com/ssisk","followers_url":"https://api.github.com/users/ssisk/followers","following_url":"https://api.github.com/users/ssisk/following{/other_user}","gists_url":"https://api.github.com/users/ssisk/gists{/gist_id}","starred_url":"https://api.github.com/users/ssisk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ssisk/subscriptions","organizations_url":"https://api.github.com/users/ssisk/orgs","repos_url":"https://api.github.com/users/ssisk/repos","events_url":"https://api.github.com/users/ssisk/events{/privacy}","received_events_url":"https://api.github.com/users/ssisk/received_events","type":"User","site_admin":false},"body":"\r\n","created_at":"2017-01-04T23:50:25Z","updated_at":"2017-01-20T01:28:42Z","closed_at":null,"merged_at":null,"merge_commit_sha":"578de561e1b2ce48fee330bb079fffb8523c751f","assignee":{"login":"ssisk","id":707137,"avatar_url":"https://avatars.githubusercontent.com/u/707137?v=3","gravatar_id":"","url":"https://api.github.com/users/ssisk","html_url":"https://github.com/ssisk","followers_url":"https://api.github.com/users/ssisk/followers","following_url":"https://api.github.com/users/ssisk/following{/other_user}","gists_url":"https://api.github.com/users/ssisk/gists{/gist_id}","starred_url":"https://api.github.com/users/ssisk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ssisk/subscriptions","organizations_url":"https://api.github.com/users/ssisk/orgs","repos_url":"https://api.github.com/users/ssisk/repos","events_url":"https://api.github.com/users/ssisk/events{/privacy}","received_events_url":"https://api.github.com/users/ssisk/received_events","type":"User","site_admin":false},"assignees":[{"login":"ssisk","id":707137,"avatar_url":"https://avatars.githubusercontent.com/u/707137?v=3","gravatar_id":"","url":"https://api.github.com/users/ssisk","html_url":"https://github.com/ssisk","followers_url":"https://api.github.com/users/ssisk/followers","following_url":"https://api.github.com/users/ssisk/following{/other_user}","gists_url":"https://api.github.com/users/ssisk/gists{/gist_id}","starred_url":"https://api.github.com/users/ssisk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ssisk/subscriptions","organizations_url":"https://api.github.com/users/ssisk/orgs","repos_url":"https://api.github.com/users/ssisk/repos","events_url":"https://api.github.com/users/ssisk/events{/privacy}","received_events_url":"https://api.github.com/users/ssisk/received_events","type":"User","site_admin":false}],"milestone":null,"commits_url":"https://api.github.com/repos/ssisk/beam/pulls/1/commits","review_comments_url":"https://api.github.com/repos/ssisk/beam/pulls/1/comments","review_comment_url":"https://api.github.com/repos/ssisk/beam/pulls/comments{/number}","comments_url":"https://api.github.com/repos/ssisk/beam/issues/1/comments","statuses_url":"https://api.github.com/repos/ssisk/beam/statuses/d66ee45403c38af1d59a289333bbdd96c7ac3a51","head":{"label":"diptikul:master","ref":"master","sha":"d66ee45403c38af1d59a289333bbdd96c7ac3a51","user":{"login":"diptikul","id":24399965,"avatar_url":"https://avatars.githubusercontent.com/u/24399965?v=3","gravatar_id":"","url":"https://api.github.com/users/diptikul","html_url":"https://github.com/diptikul","followers_url":"https://api.github.com/users/diptikul/followers","following_url":"https://api.github.com/users/diptikul/following{/other_user}","gists_url":"https://api.github.com/users/diptikul/gists{/gist_id}","starred_url":"https://api.github.com/users/diptikul/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/diptikul/subscriptions","organizations_url":"https://api.github.com/users/diptikul/orgs","repos_url":"https://api.github.com/users/diptikul/repos","events_url":"https://api.github.com/users/diptikul/events{/privacy}","received_events_url":"https://api.github.com/users/diptikul/received_events","type":"User","site_admin":false},"repo":{"id":75679495,"name":"incubator-beam","full_name":"diptikul/incubator-beam","owner":{"login":"diptikul","id":24399965,"avatar_url":"https://avatars.githubusercontent.com/u/24399965?v=3","gravatar_id":"","url":"https://api.github.com/users/diptikul","html_url":"https://github.com/diptikul","followers_url":"https://api.github.com/users/diptikul/followers","following_url":"https://api.github.com/users/diptikul/following{/other_user}","gists_url":"https://api.github.com/users/diptikul/gists{/gist_id}","starred_url":"https://api.github.com/users/diptikul/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/diptikul/subscriptions","organizations_url":"https://api.github.com/users/diptikul/orgs","repos_url":"https://api.github.com/users/diptikul/repos","events_url":"https://api.github.com/users/diptikul/events{/privacy}","received_events_url":"https://api.github.com/users/diptikul/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/diptikul/incubator-beam","description":"Mirror of Apache Beam (Incubating)","fork":true,"url":"https://api.github.com/repos/diptikul/incubator-beam","forks_url":"https://api.github.com/repos/diptikul/incubator-beam/forks","keys_url":"https://api.github.com/repos/diptikul/incubator-beam/keys{/key_id}","collaborators_url":"https://api.github.com/repos/diptikul/incubator-beam/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/diptikul/incubator-beam/teams","hooks_url":"https://api.github.com/repos/diptikul/incubator-beam/hooks","issue_events_url":"https://api.github.com/repos/diptikul/incubator-beam/issues/events{/number}","events_url":"https://api.github.com/repos/diptikul/incubator-beam/events","assignees_url":"https://api.github.com/repos/diptikul/incubator-beam/assignees{/user}","branches_url":"https://api.github.com/repos/diptikul/incubator-beam/branches{/branch}","tags_url":"https://api.github.com/repos/diptikul/incubator-beam/tags","blobs_url":"https://api.github.com/repos/diptikul/incubator-beam/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/diptikul/incubator-beam/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/diptikul/incubator-beam/git/refs{/sha}","trees_url":"https://api.github.com/repos/diptikul/incubator-beam/git/trees{/sha}","statuses_url":"https://api.github.com/repos/diptikul/incubator-beam/statuses/{sha}","languages_url":"https://api.github.com/repos/diptikul/incubator-beam/languages","stargazers_url":"https://api.github.com/repos/diptikul/incubator-beam/stargazers","contributors_url":"https://api.github.com/repos/diptikul/incubator-beam/contributors","subscribers_url":"https://api.github.com/repos/diptikul/incubator-beam/subscribers","subscription_url":"https://api.github.com/repos/diptikul/incubator-beam/subscription","commits_url":"https://api.github.com/repos/diptikul/incubator-beam/commits{/sha}","git_commits_url":"https://api.github.com/repos/diptikul/incubator-beam/git/commits{/sha}","comments_url":"https://api.github.com/repos/diptikul/incubator-beam/comments{/number}","issue_comment_url":"https://api.github.com/repos/diptikul/incubator-beam/issues/comments{/number}","contents_url":"https://api.github.com/repos/diptikul/incubator-beam/contents/{+path}","compare_url":"https://api.github.com/repos/diptikul/incubator-beam/compare/{base}...{head}","merges_url":"https://api.github.com/repos/diptikul/incubator-beam/merges","archive_url":"https://api.github.com/repos/diptikul/incubator-beam/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/diptikul/incubator-beam/downloads","issues_url":"https://api.github.com/repos/diptikul/incubator-beam/issues{/number}","pulls_url":"https://api.github.com/repos/diptikul/incubator-beam/pulls{/number}","milestones_url":"https://api.github.com/repos/diptikul/incubator-beam/milestones{/number}","notifications_url":"https://api.github.com/repos/diptikul/incubator-beam/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/diptikul/incubator-beam/labels{/name}","releases_url":"https://api.github.com/repos/diptikul/incubator-beam/releases{/id}","deployments_url":"https://api.github.com/repos/diptikul/incubator-beam/deployments","created_at":"2016-12-06T00:43:59Z","updated_at":"2016-12-06T00:44:05Z","pushed_at":"2017-01-19T16:02:51Z","git_url":"git://github.com/diptikul/incubator-beam.git","ssh_url":"git@github.com:diptikul/incubator-beam.git","clone_url":"https://github.com/diptikul/incubator-beam.git","svn_url":"https://github.com/diptikul/incubator-beam","homepage":null,"size":26683,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":0,"forks":0,"open_issues":0,"watchers":0,"default_branch":"master"}},"base":{"label":"ssisk:master","ref":"master","sha":"49aeef92ae201f357c06d338ee93b81738e3bf52","user":{"login":"ssisk","id":707137,"avatar_url":"https://avatars.githubusercontent.com/u/707137?v=3","gravatar_id":"","url":"https://api.github.com/users/ssisk","html_url":"https://github.com/ssisk","followers_url":"https://api.github.com/users/ssisk/followers","following_url":"https://api.github.com/users/ssisk/following{/other_user}","gists_url":"https://api.github.com/users/ssisk/gists{/gist_id}","starred_url":"https://api.github.com/users/ssisk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ssisk/subscriptions","organizations_url":"https://api.github.com/users/ssisk/orgs","repos_url":"https://api.github.com/users/ssisk/repos","events_url":"https://api.github.com/users/ssisk/events{/privacy}","received_events_url":"https://api.github.com/users/ssisk/received_events","type":"User","site_admin":false},"repo":{"id":71084459,"name":"beam","full_name":"ssisk/beam","owner":{"login":"ssisk","id":707137,"avatar_url":"https://avatars.githubusercontent.com/u/707137?v=3","gravatar_id":"","url":"https://api.github.com/users/ssisk","html_url":"https://github.com/ssisk","followers_url":"https://api.github.com/users/ssisk/followers","following_url":"https://api.github.com/users/ssisk/following{/other_user}","gists_url":"https://api.github.com/users/ssisk/gists{/gist_id}","starred_url":"https://api.github.com/users/ssisk/starred{/owner}{/repo}","subscriptions_url":"https://api.github.com/users/ssisk/subscriptions","organizations_url":"https://api.github.com/users/ssisk/orgs","repos_url":"https://api.github.com/users/ssisk/repos","events_url":"https://api.github.com/users/ssisk/events{/privacy}","received_events_url":"https://api.github.com/users/ssisk/received_events","type":"User","site_admin":false},"private":false,"html_url":"https://github.com/ssisk/beam","description":"Mirror of Apache Beam","fork":true,"url":"https://api.github.com/repos/ssisk/beam","forks_url":"https://api.github.com/repos/ssisk/beam/forks","keys_url":"https://api.github.com/repos/ssisk/beam/keys{/key_id}","collaborators_url":"https://api.github.com/repos/ssisk/beam/collaborators{/collaborator}","teams_url":"https://api.github.com/repos/ssisk/beam/teams","hooks_url":"https://api.github.com/repos/ssisk/beam/hooks","issue_events_url":"https://api.github.com/repos/ssisk/beam/issues/events{/number}","events_url":"https://api.github.com/repos/ssisk/beam/events","assignees_url":"https://api.github.com/repos/ssisk/beam/assignees{/user}","branches_url":"https://api.github.com/repos/ssisk/beam/branches{/branch}","tags_url":"https://api.github.com/repos/ssisk/beam/tags","blobs_url":"https://api.github.com/repos/ssisk/beam/git/blobs{/sha}","git_tags_url":"https://api.github.com/repos/ssisk/beam/git/tags{/sha}","git_refs_url":"https://api.github.com/repos/ssisk/beam/git/refs{/sha}","trees_url":"https://api.github.com/repos/ssisk/beam/git/trees{/sha}","statuses_url":"https://api.github.com/repos/ssisk/beam/statuses/{sha}","languages_url":"https://api.github.com/repos/ssisk/beam/languages","stargazers_url":"https://api.github.com/repos/ssisk/beam/stargazers","contributors_url":"https://api.github.com/repos/ssisk/beam/contributors","subscribers_url":"https://api.github.com/repos/ssisk/beam/subscribers","subscription_url":"https://api.github.com/repos/ssisk/beam/subscription","commits_url":"https://api.github.com/repos/ssisk/beam/commits{/sha}","git_commits_url":"https://api.github.com/repos/ssisk/beam/git/commits{/sha}","comments_url":"https://api.github.com/repos/ssisk/beam/comments{/number}","issue_comment_url":"https://api.github.com/repos/ssisk/beam/issues/comments{/number}","contents_url":"https://api.github.com/repos/ssisk/beam/contents/{+path}","compare_url":"https://api.github.com/repos/ssisk/beam/compare/{base}...{head}","merges_url":"https://api.github.com/repos/ssisk/beam/merges","archive_url":"https://api.github.com/repos/ssisk/beam/{archive_format}{/ref}","downloads_url":"https://api.github.com/repos/ssisk/beam/downloads","issues_url":"https://api.github.com/repos/ssisk/beam/issues{/number}","pulls_url":"https://api.github.com/repos/ssisk/beam/pulls{/number}","milestones_url":"https://api.github.com/repos/ssisk/beam/milestones{/number}","notifications_url":"https://api.github.com/repos/ssisk/beam/notifications{?since,all,participating}","labels_url":"https://api.github.com/repos/ssisk/beam/labels{/name}","releases_url":"https://api.github.com/repos/ssisk/beam/releases{/id}","deployments_url":"https://api.github.com/repos/ssisk/beam/deployments","created_at":"2016-10-16T23:48:31Z","updated_at":"2017-01-04T17:29:32Z","pushed_at":"2017-01-19T16:02:52Z","git_url":"git://github.com/ssisk/beam.git","ssh_url":"git@github.com:ssisk/beam.git","clone_url":"https://github.com/ssisk/beam.git","svn_url":"https://github.com/ssisk/beam","homepage":"","size":25812,"stargazers_count":0,"watchers_count":0,"language":"Java","has_issues":false,"has_downloads":true,"has_wiki":false,"has_pages":false,"forks_count":0,"mirror_url":null,"open_issues_count":1,"forks":0,"open_issues":1,"watchers":0,"default_branch":"master"}},"_links":{"self":{"href":"https://api.github.com/repos/ssisk/beam/pulls/1"},"html":{"href":"https://github.com/ssisk/beam/pull/1"},"issue":{"href":"https://api.github.com/repos/ssisk/beam/issues/1"},"comments":{"href":"https://api.github.com/repos/ssisk/beam/issues/1/comments"},"review_comments":{"href":"https://api.github.com/repos/ssisk/beam/pulls/1/comments"},"review_comment":{"href":"https://api.github.com/repos/ssisk/beam/pulls/comments{/number}"},"commits":{"href":"https://api.github.com/repos/ssisk/beam/pulls/1/commits"},"statuses":{"href":"https://api.github.com/repos/ssisk/beam/statuses/d66ee45403c38af1d59a289333bbdd96c7ac3a51"}}}}
{"actor":{"display_login":"ssisk"}}