Filter extends PTransform<PCollection<T>, PCollection<T>> { public static <T extends Comparable<T>> Filter<T> equal(final T value) { return by((ProcessFunction<T, Boolean>) input -> input.compareTo(value) == 0) .described(String.format("x == %s", value)); } private  Filter(ProcessFunction<T, Boolean> predicate); private  Filter(ProcessFunction<T, Boolean> predicate, String predicateDescription); static Filter<T> by(
      PredicateT predicate); static Filter<T> by(
      PredicateT predicate); static Filter<T> lessThan(final T value); static Filter<T> greaterThan(final T value); static Filter<T> lessThanEq(final T value); static Filter<T> greaterThanEq(final T value); static Filter<T> equal(final T value); @Override PCollection<T> expand(PCollection<T> input); @Override void populateDisplayData(DisplayData.Builder builder);  }
@Test @Category(NeedsRunner.class) public void testFilterEqual() { PCollection<Integer> output = p.apply(Create.of(1, 2, 3, 4, 5, 6, 7)).apply(Filter.equal(4)); PAssert.that(output).containsInAnyOrder(4); p.run(); }