public static <T, PredicateT extends ProcessFunction<T, Boolean>> Filter<T> by(
      PredicateT predicate) {
    return new Filter<>(predicate);
  }
@Test
  @Category(NeedsRunner.class)
  public void testIdentityFilterByPredicate() {
    PCollection<Integer> output =
        p.apply(Create.of(591, 11789, 1257, 24578, 24799, 307))
            .apply(Filter.by(new TrivialFn(true)));

    PAssert.that(output).containsInAnyOrder(591, 11789, 1257, 24578, 24799, 307);
    p.run();
  }