DynamicInputFormat extends InputFormat<K, V> { private static int getSplitRatio(Configuration conf) { int splitRatio = conf.getInt( DistCpConstants.CONF_LABEL_SPLIT_RATIO, DistCpConstants.SPLIT_RATIO_DEFAULT); if (splitRatio <= 0) { LOG.warn(DistCpConstants.CONF_LABEL_SPLIT_RATIO + " should be positive. Fall back to default value: " + DistCpConstants.SPLIT_RATIO_DEFAULT); splitRatio = DistCpConstants.SPLIT_RATIO_DEFAULT; } return splitRatio; }  @Override List<InputSplit> getSplits(JobContext jobContext); DynamicInputChunkContext<K, V> getChunkContext(
      Configuration configuration); @Override RecordReader<K, V> createRecordReader(
          InputSplit inputSplit,
          TaskAttemptContext taskAttemptContext);  }
@Test public void testGetSplitRatio() throws Exception { Assert.assertEquals(1, DynamicInputFormat.getSplitRatio(1, 1000000000)); Assert.assertEquals(2, DynamicInputFormat.getSplitRatio(11000000, 10)); Assert.assertEquals(4, DynamicInputFormat.getSplitRatio(30, 700)); Assert.assertEquals(2, DynamicInputFormat.getSplitRatio(30, 200)); Configuration conf = new Configuration(); conf.setInt(DistCpConstants.CONF_LABEL_MAX_CHUNKS_TOLERABLE, -1); conf.setInt(DistCpConstants.CONF_LABEL_MAX_CHUNKS_IDEAL, -1); conf.setInt(DistCpConstants.CONF_LABEL_MIN_RECORDS_PER_CHUNK, -1); conf.setInt(DistCpConstants.CONF_LABEL_SPLIT_RATIO, -1); Assert.assertEquals(1, DynamicInputFormat.getSplitRatio(1, 1000000000, conf)); Assert.assertEquals(2, DynamicInputFormat.getSplitRatio(11000000, 10, conf)); Assert.assertEquals(4, DynamicInputFormat.getSplitRatio(30, 700, conf)); Assert.assertEquals(2, DynamicInputFormat.getSplitRatio(30, 200, conf)); conf.setInt(DistCpConstants.CONF_LABEL_MAX_CHUNKS_TOLERABLE, 100); conf.setInt(DistCpConstants.CONF_LABEL_MAX_CHUNKS_IDEAL, 30); conf.setInt(DistCpConstants.CONF_LABEL_MIN_RECORDS_PER_CHUNK, 10); conf.setInt(DistCpConstants.CONF_LABEL_SPLIT_RATIO, 53); Assert.assertEquals(53, DynamicInputFormat.getSplitRatio(3, 200, conf)); }