@Override
  public List<InputSplit> getSplits(JobContext jobContext)
      throws IOException, InterruptedException {
    LOG.info("DynamicInputFormat: Getting splits for job:"
             + jobContext.getJobID());
    chunkContext = getChunkContext(jobContext.getConfiguration());
    return createSplits(jobContext,
                        splitCopyListingIntoChunksWithShuffle(jobContext));
  }
@Test
  public void testGetSplits() throws Exception {
    final DistCpContext context = new DistCpContext(getOptions());
    Configuration configuration = new Configuration();
    configuration.set("mapred.map.tasks",
                      String.valueOf(context.getMaxMaps()));
    CopyListing.getCopyListing(configuration, CREDENTIALS, context)
        .buildListing(new Path(cluster.getFileSystem().getUri().toString()
            +"/tmp/testDynInputFormat/fileList.seq"), context);

    JobContext jobContext = new JobContextImpl(configuration, new JobID());
    DynamicInputFormat<Text, CopyListingFileStatus> inputFormat =
        new DynamicInputFormat<Text, CopyListingFileStatus>();
    List<InputSplit> splits = inputFormat.getSplits(jobContext);

    int nFiles = 0;
    int taskId = 0;

    for (InputSplit split : splits) {
      StubContext stubContext = new StubContext(jobContext.getConfiguration(),
                                                null, taskId);
      final TaskAttemptContext taskAttemptContext
         = stubContext.getContext();

      RecordReader<Text, CopyListingFileStatus> recordReader =
          inputFormat.createRecordReader(split, taskAttemptContext);
      stubContext.setReader(recordReader);
      recordReader.initialize(splits.get(0), taskAttemptContext);
      float previousProgressValue = 0f;
      while (recordReader.nextKeyValue()) {
        CopyListingFileStatus fileStatus = recordReader.getCurrentValue();
        String source = fileStatus.getPath().toString();
        System.out.println(source);
        Assert.assertTrue(expectedFilePaths.contains(source));
        final float progress = recordReader.getProgress();
        Assert.assertTrue(progress >= previousProgressValue);
        Assert.assertTrue(progress >= 0.0f);
        Assert.assertTrue(progress <= 1.0f);
        previousProgressValue = progress;
        ++nFiles;
      }
      Assert.assertTrue(recordReader.getProgress() == 1.0f);

      ++taskId;
    }

    Assert.assertEquals(expectedFilePaths.size(), nFiles);
  }