### Question:
HighWaterMarkInputFormat extends TextInputFormat { @Nonnull @Override public List<FileStatus> listStatus(@Nonnull final JobContext job) throws IOException { long highWaterMark = Math.max(lastHighWaterMark, getHighWaterMark(job)); final long maxAge = HighWaterMarkInputFormat.getMaxFileAge(job); final long minAge = HighWaterMarkInputFormat.getMinFileAge(job); if (minAge > maxAge) { throw new IOException(MIN_FILE_AGE + " cannot be greater than " + MAX_FILE_AGE); } final List<FileStatus> allFiles = super.listStatus(job); final List<FileStatus> jobFiles = new ArrayList<>(allFiles.size()); final long currentTime = currentTimeMillis(); for (final FileStatus file : allFiles) { final long fileTime = file.getModificationTime(); final long fileAge = currentTime - fileTime; if (!file.isDirectory() && fileAge >= minAge && fileAge <= maxAge && fileTime > highWaterMark) { jobFiles.add(file); if (fileTime > lastHighWaterMark) { lastHighWaterMark = fileTime; } } } lastHighWaterMark = Math.max(lastHighWaterMark, highWaterMark); return jobFiles; }  static long getHighWaterMark(@Nonnull final JobContext job); static void setHighWaterMark(@Nonnull final Job job, final long highWaterMark); static long getMaxFileAge(@Nonnull final JobContext job); static void setMaxFileAge(@Nonnull final Job job, final long maxFileAge); static long getMinFileAge(@Nonnull final JobContext job); static void setMinFileAge(@Nonnull final Job job, final long minFileAge); long getLastHighWaterMark(); @Nonnull @Override List<FileStatus> listStatus(@Nonnull final JobContext job); static final String HIGH_WATER_MARK; static final String MAX_FILE_AGE; static final String MIN_FILE_AGE; }### Answer:
@Test public void listStatus() throws IOException { final File file1 = tempFolder.newFile("file1"); Assert.assertTrue(file1.setLastModified(currentTimeMillis)); final File folder1 = tempFolder.newFolder("folder1"); final File file2 = tempFolder.newFile("folder1/file2"); Assert.assertTrue(file2.setLastModified(currentTimeMillis)); tempFolder.newFolder("folder1", "folder2"); final File file3 = tempFolder.newFile("folder1/folder2/file3"); Assert.assertTrue(file3.setLastModified(currentTimeMillis)); final Job job = Job.getInstance(new Configuration(false)); HighWaterMarkInputFormat.setInputPaths(job, file1.getAbsolutePath() + "," + folder1.getAbsolutePath()); final HighWaterMarkInputFormat inputFormat = new MockHighWaterMarkInputFormat(); final List<FileStatus> files = inputFormat.listStatus(job); Collections.sort(files, new FileStatusComparator()); Assert.assertEquals(new Path(file1.toURI()), files.get(0).getPath()); Assert.assertEquals(new Path(file2.toURI()), files.get(1).getPath()); Assert.assertEquals(2, files.size()); Assert.assertEquals(currentTimeMillis, inputFormat.getLastHighWaterMark()); Assert.assertEquals(0, inputFormat.listStatus(job).size()); }