WithTimestamps extends PTransform<PCollection<T>, PCollection<T>> { public static <T> WithTimestamps<T> of(SerializableFunction<T, Instant> fn) { return new WithTimestamps<>(fn, Duration.ZERO); } private  WithTimestamps(SerializableFunction<T, Instant> fn, Duration allowedTimestampSkew); static WithTimestamps<T> of(SerializableFunction<T, Instant> fn); @Deprecated WithTimestamps<T> withAllowedTimestampSkew(Duration allowedTimestampSkew); @Deprecated Duration getAllowedTimestampSkew(); @Override PCollection<T> expand(PCollection<T> input);  }
@Test @Category(ValidatesRunner.class) public void withTimestampsShouldApplyTimestamps() { SerializableFunction<String, Instant> timestampFn = input -> new Instant(Long.valueOf(input)); String yearTwoThousand = "946684800000"; PCollection<String> timestamped = p.apply(Create.of("1234", "0", Integer.toString(Integer.MAX_VALUE), yearTwoThousand)) .apply(WithTimestamps.of(timestampFn)); PCollection<KV<String, Instant>> timestampedVals = timestamped.apply( ParDo.of( new DoFn<String, KV<String, Instant>>() { @ProcessElement public void processElement(DoFn<String, KV<String, Instant>>.ProcessContext c) throws Exception { c.output(KV.of(c.element(), c.timestamp())); } })); PAssert.that(timestamped) .containsInAnyOrder(yearTwoThousand, "0", "1234", Integer.toString(Integer.MAX_VALUE)); PAssert.that(timestampedVals) .containsInAnyOrder( KV.of("0", new Instant(0)), KV.of("1234", new Instant(1234L)), KV.of(Integer.toString(Integer.MAX_VALUE), new Instant(Integer.MAX_VALUE)), KV.of(yearTwoThousand, new Instant(Long.valueOf(yearTwoThousand)))); p.run(); }
@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(); }
@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(); }
@Test @Category(ValidatesRunner.class) public void withTimestampsWithNullFnShouldThrowOnConstruction() { SerializableFunction<String, Instant> timestampFn = null; thrown.expect(NullPointerException.class); thrown.expectMessage("WithTimestamps fn cannot be null"); p.apply(Create.of("1234", "0", Integer.toString(Integer.MAX_VALUE))) .apply(WithTimestamps.of(timestampFn)); p.run(); }
@Test @Category(ValidatesRunner.class) public void withTimestampsLambdaShouldApplyTimestamps() { final String yearTwoThousand = "946684800000"; PCollection<String> timestamped = p.apply(Create.of("1234", "0", Integer.toString(Integer.MAX_VALUE), yearTwoThousand)) .apply(WithTimestamps.of((String input) -> new Instant(Long.valueOf(input)))); PCollection<KV<String, Instant>> timestampedVals = timestamped.apply( ParDo.of( new DoFn<String, KV<String, Instant>>() { @ProcessElement public void processElement(ProcessContext c) throws Exception { c.output(KV.of(c.element(), c.timestamp())); } })); PAssert.that(timestamped) .containsInAnyOrder(yearTwoThousand, "0", "1234", Integer.toString(Integer.MAX_VALUE)); PAssert.that(timestampedVals) .containsInAnyOrder( KV.of("0", new Instant(0)), KV.of("1234", new Instant(Long.valueOf("1234"))), KV.of(Integer.toString(Integer.MAX_VALUE), new Instant(Integer.MAX_VALUE)), KV.of(yearTwoThousand, new Instant(Long.valueOf(yearTwoThousand)))); p.run(); }