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())));

    /*
     * Concept #3: PAssert is a set of convenient PTransforms in the style of
     * Hamcrest's collection matchers that can be used when writing Pipeline level tests
     * to validate the contents of PCollections. PAssert is best used in unit tests
     * with small data sets but is demonstrated here as a teaching tool.
     *
     * <p>Below we verify that the set of filtered words matches our expected counts. Note
     * that PAssert does not provide any output and that successful completion of the
     * Pipeline implies that the expectations were met. Learn more at
     * https://beam.apache.org/documentation/pipelines/test-your-pipeline/ on how to test
     * your Pipeline and see {@link DebuggingWordCountTest} for an example unit test.
     */
    List<KV<String, Long>> expectedResults =
        Arrays.asList(KV.of("Flourish", 3L), KV.of("stomach", 1L));
    PAssert.that(filteredWords).containsInAnyOrder(expectedResults);

    p.run().waitUntilFinish();
  }
@Test
  public void testDebuggingWordCount() throws Exception {
    File inputFile = tmpFolder.newFile();
    File outputFile = tmpFolder.newFile();
    Files.asCharSink(inputFile, StandardCharsets.UTF_8)
        .write("stomach secret Flourish message Flourish here Flourish");
    WordCountOptions options = TestPipeline.testingPipelineOptions().as(WordCountOptions.class);
    options.setInputFile(getFilePath(inputFile.getAbsolutePath()));
    options.setOutput(getFilePath(outputFile.getAbsolutePath()));
    DebuggingWordCount.runDebuggingWordCount(options);
  }