laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r732207845
##########
File path:
runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -472,6 +663,37 @@ public Duration getAllowedTimestampSkew() {
}
}
+ /**
+ * A {@link DoFn} that creates/sets a timer with an output timestamp equal
to the input timestamp
+ * minus the input element's value. Keys are ignored but required for timers.
+ */
+ private static class TimerSkewingDoFn extends DoFn<KV<String, Duration>,
Duration> {
+ static final String TIMER_ID = "testTimerId";
+ private final Duration allowedSkew;
+
+ @TimerId(TIMER_ID)
+ private static final TimerSpec timer =
TimerSpecs.timer(TimeDomain.PROCESSING_TIME);
+
+ private TimerSkewingDoFn(Duration allowedSkew) {
+ this.allowedSkew = allowedSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context, @TimerId(TIMER_ID)
Timer timer) {
+ timer
+
.withOutputTimestamp(context.timestamp().minus(context.element().getValue()))
+ .set(new Instant(0));
+ }
+
+ @OnTimer(TIMER_ID)
+ public void onTimer() {}
Review comment:
If by this you mean setting the output timestamp via onTimer (e.g.
timer.set() in DoFn, timerContext.outputWithTimestamp in timer), then this
actually does not properly check for skew. Let's chat offline about how to
handle this. I'm not sure if we want to include in this CL as it's already
quite bit and since I think it's broken for both runners it could affect user
DoFns too.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
