[BEAM-10329] Dataflow runner does not reserve timestamp of Create.Timestamped() in batch stateful dofn

• Type: Bug
• Status: Resolved
• Priority: P2
• Resolution: Fixed
• Fix Version/s: 2.23.0
• Component/s: runner-dataflow
• Labels: stale-assigned
• Assignee: Yichi Zhang
• Reporter: Yichi Zhang
• Created: 26/Jun/20 06:41
• Updated: 30/Jan/21 05:22
• Resolved: 30/Jan/21 05:22

Description

When run a test pipeline such as

public void testTimestampedValue() throws Exception {
      final String timerId = "foo";

      DoFn<KV<String, Long>, KV<Long, Instant>> statefn =
          new DoFn<KV<String, Long>, KV<Long, Instant>>() {

            @TimerId(timerId)
            private final TimerSpec spec = TimerSpecs.timer(TimeDomain.EVENT_TIME);

            @ProcessElement
            public void processElement(
                @TimerId(timerId) Timer timer,
                @Timestamp Instant timestamp,
                OutputReceiver<KV<Long, Instant>> r) {
              r.output(KV.of(3L, timestamp));
            }

            @OnTimer(timerId)
            public void onTimer(@Timestamp Instant timestamp, OutputReceiver<KV<Long, Instant>> r) {
              // do nothing. Since whether timer is involved doesn’t make difference
            }
          };

      PCollection<KV<Long, Instant>> output =
          pipeline
              .apply(Create.timestamped(TimestampedValue.of(KV.of("hello", 37L), new Instant(123L))))
              .apply(ParDo.of(statefn));

      PAssert.that(output).containsInAnyOrder(KV.of(3L, new Instant(123L)));
      pipeline.run();
}

On dataflow with fnapi, the timestamp of output in PAssert won't match.

Issue Links

• links to: GitHub Pull Request #12096