public static <T> WithTimestamps<T> of(SerializableFunction<T, Instant> fn) {
    return new WithTimestamps<>(fn, Duration.ZERO);
  }
@Test
  @Category(NeedsRunner.class)
  public void withTimestampsWithNullTimestampShouldThrow() {
    SerializableFunction<String, Instant> timestampFn = input -> null;

    String yearTwoThousand = "946684800000";
    p.apply(Create.of("1234", "0", Integer.toString(Integer.MAX_VALUE), yearTwoThousand))
        .apply(WithTimestamps.of(timestampFn));

    thrown.expect(PipelineExecutionException.class);
    thrown.expectCause(isA(NullPointerException.class));
    thrown.expectMessage("WithTimestamps");
    thrown.expectMessage("cannot be null");

    p.run();
  }