code: private Map<Long, List<TimestampedFileInputSplit>> getInputSplitsSortedByModTime(
  				Map<Path, FileStatus> eligibleFiles) throws IOException {
  
  		Map<Long, List<TimestampedFileInputSplit>> splitsByModTime = new TreeMap<>();
  		if (eligibleFiles.isEmpty()) {
  			return splitsByModTime;
  		}
  
  		for (FileInputSplit split: format.createInputSplits(readerParallelism)) {
  			FileStatus fileStatus = eligibleFiles.get(split.getPath());
  			if (fileStatus != null) {
  				Long modTime = fileStatus.getModificationTime();
  				List<TimestampedFileInputSplit> splitsToForward = splitsByModTime.get(modTime);
  				if (splitsToForward == null) {
  					splitsToForward = new ArrayList<>();
  					splitsByModTime.put(modTime, splitsToForward);
  				}
  				splitsToForward.add(new TimestampedFileInputSplit(
  					modTime, split.getSplitNumber(), split.getPath(),
  					split.getStart(), split.getLength(), split.getHostnames()));
  			}
  		}
  		return splitsByModTime;
  	}
docs: Creates the input splits to be forwarded to the downstream tasks of the
  {@link ContinuousFileReaderOperator}. Splits are sorted <b>by modification time</b> before
  being forwarded and only splits belonging to files in the {@code eligibleFiles}
  list will be processed.
  @param eligibleFiles The files to process.