DebuggingWordCount { static void runDebuggingWordCount(WordCountOptions options) { Pipeline p = Pipeline.create(options); PCollection<KV<String, Long>> filteredWords = p.apply("ReadLines", TextIO.read().from(options.getInputFile())) .apply(new WordCount.CountWords()) .apply(ParDo.of(new FilterTextFn(options.getFilterPattern()))); List<KV<String, Long>> expectedResults = Arrays.asList(KV.of("Flourish", 3L), KV.of("stomach", 1L)); PAssert.that(filteredWords).containsInAnyOrder(expectedResults); p.run().waitUntilFinish(); }  static void main(String[] args);  }
@Test public void testDebuggingWordCount() throws Exception { File inputFile = tmpFolder.newFile(); File outputFile = tmpFolder.newFile(); Files.write( "stomach secret Flourish message Flourish here Flourish", inputFile, StandardCharsets.UTF_8); WordCountOptions options = TestPipeline.testingPipelineOptions().as(WordCountOptions.class); options.setInputFile(getFilePath(inputFile.getAbsolutePath())); options.setOutput(getFilePath(outputFile.getAbsolutePath())); DebuggingWordCount.runDebuggingWordCount(options); }