public static <T> WithTimestamps<T> of(SerializableFunction<T, Instant> fn) {
    return new WithTimestamps<>(fn, Duration.ZERO);
  }
@Test
  @Category(NeedsRunner.class)
  public void withTimestampsBackwardsInTimeShouldThrow() {

    SerializableFunction<String, Instant> timestampFn = input -> new Instant(Long.valueOf(input));
    SerializableFunction<String, Instant> backInTimeFn =
        input -> new Instant(Long.valueOf(input)).minus(Duration.millis(1000L));

    String yearTwoThousand = "946684800000";

    p.apply(Create.of("1234", "0", Integer.toString(Integer.MAX_VALUE), yearTwoThousand))
        .apply("WithTimestamps", WithTimestamps.of(timestampFn))
        .apply("AddSkew", WithTimestamps.of(backInTimeFn));

    thrown.expect(PipelineExecutionException.class);
    thrown.expectCause(isA(IllegalArgumentException.class));
    thrown.expectMessage("no earlier than the timestamp of the current input");

    p.run();
  }