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