DynamicInputFormat extends InputFormat<K, V> { @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)); }  @Override List<InputSplit> getSplits(JobContext jobContext); DynamicInputChunkContext<K, V> getChunkContext(
      Configuration configuration); @Override RecordReader<K, V> createRecordReader(
          InputSplit inputSplit,
          TaskAttemptContext taskAttemptContext);  }
@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); }