### Question:
FileBasedDeadLetterQueueReconsumer extends PTransform<PBegin, PCollection<String>> { static PTransform<PCollection<Metadata>, PCollection<String>> moveAndConsumeMatches() { return new PTransform<PCollection<Metadata>, PCollection<String>>() { @Override public PCollection<String> expand(PCollection<Metadata> input) { return input.apply(Reshuffle.viaRandomKey()) .apply(ParDo.of(new MoveAndConsumeFn())); } }; } private  FileBasedDeadLetterQueueReconsumer(String dlqDirectory, Duration recheckPeriod); static FileBasedDeadLetterQueueReconsumer create(String dlqDirectory); PCollection<String> expand(PBegin in); static final Duration DEFAULT_RECHECK_PERIOD; static final String TEMPORARY_HOLD_SUBDIRECTORY; }### Answer:
@Test public void testFilesAreConsumed() throws IOException { String fileName = createJsonFile("dlqFile1.json", JSON_FILE_CONTENTS_1); folder.newFolder("tmp"); String folderPath = Paths.get(folder.getRoot().getAbsolutePath()).resolve("*").toString(); PCollection<String> jsonData = p .apply(FileIO.match() .filepattern(folderPath)) .apply(FileBasedDeadLetterQueueReconsumer.moveAndConsumeMatches()); PAssert.that(jsonData).containsInAnyOrder(JSON_RESULTS_1); p.run().waitUntilFinish(); assertFalse(new File(fileName).exists()); }