[BEAM-2535] Added control to set output timestamp independent of firing time for event time timers. (Direct Runner implementation) #4700

huygaa11 · 2018-02-16T19:23:08Z

This PR adds ability to control the output timestamp of event timers. This feature only works for the direct runner at the moment. Before this PR, the output timestamp of event timers are set to timer's delivery timestamp.

Example:
When an user sets timer with output timestamp 3 and delivery timestamp 5:

@ProcessElement()
public void processElement(ProcessContext c, @TimerId("foo") Timer timer) {
timer.withOutputTimestamp(new Instant(3)).set(new Instant(5));
}

@ontimer("foo")
public void myTimerCallback(OnTimerContext c, BoundedWindow w) {
c.output("some_value"); // outputs with timestamp 3.
}

The timer will deliver at 5 as usual. However, output() method of the OnTimerContext in myTimerCallback function uses 3 as the output timestamp. To make sure the element is not late for further transforms down the pipeline, output watermark will be held at the output timestamp of the timer (3 in the example) until myTimerCallback executes.

This WM hold feature is also useful for outputting values saved in state cells right before garbage collection. (BEAM-1589) GC timer fires after the window expiration, therefore, outputting elements with output timestamp past the window expiration results them to be dropped further down the pipeline. By setting onWindowExpiration timer as:
timer.withOutputTimestamp(windowExpiration).set(GCtime), we could output these elements inside the window.

Follow this checklist to help us incorporate your contribution quickly and easily:

…amp of the pending timers.

huygaa11 · 2018-02-16T19:27:25Z

@kennknowles Here is my initial try of implementing withOutputTimestamp() in DirectRunner.

I did not put that much effort into fixing nits and implementing tests yet. If this PR looks good in general, I will go ahead and add tests and polishing.

Let me know your initial review, until then I will be working on other tasks. Thanks.

kennknowles

Looks overall good! I think @tgroh would be a good source for also looking over direct runner changes.

kennknowles · 2018-02-16T20:37:32Z

runners/core-java/src/main/java/org/apache/beam/runners/core/ProcessFnRunner.java

@@ -77,7 +77,8 @@ public void finishBundle() {

  @Override
  public void onTimer(
-      String timerId, BoundedWindow window, Instant timestamp, TimeDomain timeDomain) {
+      String timerId, BoundedWindow window, Instant timestamp, Instant outputTimestamp,


I would also change timestamp to deliveryTimestamp in all these methods, to make the difference very clear.

kennknowles · 2018-02-16T20:37:43Z

runners/core-java/src/main/java/org/apache/beam/runners/core/PushbackSideInputDoFnRunner.java

-  void onTimer(String timerId, BoundedWindow window, Instant timestamp,
-               TimeDomain timeDomain);
+  void onTimer(
+      String timerId, BoundedWindow window, Instant timestamp, Instant outputTimestamp,


nit: google-java-format

kennknowles · 2018-02-16T20:40:08Z

runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java

@@ -698,23 +701,23 @@ public TimerInternalsTimer(

    @Override
    public void set(Instant target) {
+      this.target = target;


It would be nice that the IllegalArgumentException is thrown before the wrong value is set. Here it is not likely to cause a problem, but it is a good practice. You never know how this class might be used in the future.

what would be a considered as illegal deliveryTimestamp?

Same as would be thrown in setAndVerify - and as an aside, I'd much prefer the timestamp be returned by verify and the fields be final if that's possible (I think it is)

kennknowles · 2018-02-16T20:41:11Z

runners/core-java/src/main/java/org/apache/beam/runners/core/TimerInternals.java

-     * generated.
+     * Construct a {@link TimerData} for the given parameters.
+     */
+    public static TimerData of(


Now that we have two Instant value parameters, it is a good time to switch this to use @AutoValue.Builder.

kennknowles · 2018-02-16T20:41:51Z

runners/core-java/src/main/java/org/apache/beam/runners/core/TimerInternals.java

@@ -225,6 +258,30 @@ public int compareTo(TimerData that) {
    }
  }

+  /**
+   * Used for sorting {@link TimerData} by output timestamp of the timer.


Javadoc that ordering other than the output timestamp is stable but unspecified.

kennknowles · 2018-02-16T20:44:45Z

runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java

@@ -400,6 +418,8 @@ public Instant get() {
     *     <ul>
     *       <li>the current input watermark</li>
     *       <li>the current watermark holds</li>
+     *       <li>the earliest delivery timestamp of pending timers</li>


I believe now that you have output timestamp, you don't need to take the minimum of the delivery timestamp. That is how you can use this for GC / OnWindowExpiration.

kennknowles · 2018-02-16T20:45:21Z

runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java

+    private final SortedMultiset<TimerData> pendingTimersSortedByTs;
+
+    // Keeps track of pending timers sorted by output timestamp.
+    private final SortedMultiset<TimerData> pendingTimersSortedByOutputTs;


And for processing time timers, we just let them expire, because we don't want to have a pipeline just stall, right?

kennknowles · 2018-02-16T20:46:19Z

Definitely put in some ValidatesRunner tests in ParDoTest.

tgroh · 2018-02-16T22:47:11Z

runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java

+    private void setAndVerifyOutputTimestamp() {
+      // Output timestamp is currently not supported in processing time timers.
+      if (outputTimestamp != null && !TimeDomain.EVENT_TIME.equals(spec.getTimeDomain())) {
+        throw new IllegalStateException("Cannot set outputTimestamp in processing time domain.");


"...domains other than %s: got %s", TimeDomain.EVENT_TIME, spec.getTimeDomain()

tgroh · 2018-02-16T22:47:13Z

runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java

+        outputTimestamp = target;
+      }
+
+      System.out.println(outputTimestamp);


tgroh · 2018-02-16T22:48:26Z

runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java

+      System.out.println(outputTimestamp);
+      if (TimeDomain.EVENT_TIME.equals(spec.getTimeDomain())) {
+        Instant windowExpiry = window.maxTimestamp().plus(allowedLateness);
+        checkArgument(!outputTimestamp.isAfter(windowExpiry),


This may be an even stricter check (probably should be even) - we should enforce that the outputTimestamp is no later than the maxTimestamp of the window, if I understand windows properly

tgroh

You've got a lot of printlns and commented out code - let me know when this is ready for review (at least, having cleaned those up)

tgroh · 2018-02-21T23:01:06Z

runners/core-java/src/main/java/org/apache/beam/runners/core/DoFnRunner.java

@@ -42,7 +42,9 @@
   * Calls a {@link DoFn DoFn's} {@link DoFn.OnTimer @OnTimer} method for the given timer
   * in the given window.
   */
-  void onTimer(String timerId, BoundedWindow window, Instant timestamp, TimeDomain timeDomain);
+  void onTimer(
+      String timerId, BoundedWindow window, Instant timestamp, Instant outputTimestamp,


Formatting, here and other changed lines?

tgroh · 2018-03-21T15:51:33Z

.../direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java


    public StatefulParDoEvaluator(
-        DoFnLifecycleManagerRemovingTransformEvaluator<KV<K, InputT>> delegateEvaluator) {
+        DoFnLifecycleManagerRemovingTransformEvaluator<KV<K, InputT>> delegateEvaluator,
+        DirectStepContext stepContext,


It's pretty important to document that these are owned by the StatefulParDoEvaluator - otherwise the mutations aren't ok.

stale · 2018-06-07T04:01:24Z

This pull request has been marked as stale due to 60 days of inactivity. It will be closed in 1 week if no further activity occurs. If you think that’s incorrect or this pull request requires a review, please simply write any comment. If closed, you can revive the PR at any time and @mention a reviewer or discuss it on the dev@beam.apache.org list. Thank you for your contributions.

stale · 2018-06-14T04:53:50Z

This pull request has been closed due to lack of activity. If you think that is incorrect, or the pull request requires review, you can revive the PR at any time.

Huygaa Batsaikhan added 4 commits February 16, 2018 10:57

Added outputTimestamp to TimerData

e82d4d9

AppliedPTransformOutputWatermark is held at the minimum output timest…

8d025e3

…amp of the pending timers.

Added output timestamp to DoFnRunner's OnTimer method.

0991a1a

Implemented withOutputTimestamp

8f9d5ee

huygaa11 changed the title ~~[BEAM-2535] Added control to set output timestamp independent of firing time for event time timers.~~ [BEAM-2535] Added control to set output timestamp independent of firing time for event time timers. (Direct Runner implementation) Feb 16, 2018

kennknowles requested changes Feb 16, 2018

View reviewed changes

tgroh reviewed Feb 16, 2018

View reviewed changes

lukecwik assigned tgroh and kennknowles Feb 21, 2018

tgroh assigned huygaa11 and unassigned tgroh and kennknowles Feb 21, 2018

watermark hold state

538d683

tgroh reviewed Mar 21, 2018

View reviewed changes

stale bot added the wontfix label Jun 7, 2018

stale bot closed this Jun 14, 2018

kennknowles added stale and removed wontfix labels Jun 25, 2018

xubii mentioned this pull request Oct 25, 2019

[Beam-2535] : Support outputTimestamp and watermark holds in timers. #9677

Merged

3 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-2535] Added control to set output timestamp independent of firing time for event time timers. (Direct Runner implementation) #4700

[BEAM-2535] Added control to set output timestamp independent of firing time for event time timers. (Direct Runner implementation) #4700

huygaa11 commented Feb 16, 2018 •

edited

huygaa11 commented Feb 16, 2018 •

edited

kennknowles left a comment

kennknowles Feb 16, 2018

kennknowles Feb 16, 2018

kennknowles Feb 16, 2018

huygaa11 Feb 20, 2018

tgroh Feb 20, 2018

kennknowles Feb 16, 2018

kennknowles Feb 16, 2018

kennknowles Feb 16, 2018

kennknowles Feb 16, 2018

kennknowles commented Feb 16, 2018

tgroh Feb 16, 2018

huygaa11 Feb 20, 2018

tgroh Feb 16, 2018

huygaa11 Feb 20, 2018

tgroh Feb 16, 2018

huygaa11 Feb 20, 2018

tgroh left a comment

tgroh Feb 21, 2018

tgroh Mar 21, 2018

stale bot commented Jun 7, 2018

stale bot commented Jun 14, 2018

[BEAM-2535] Added control to set output timestamp independent of firing time for event time timers. (Direct Runner implementation) #4700

[BEAM-2535] Added control to set output timestamp independent of firing time for event time timers. (Direct Runner implementation) #4700

Conversation

huygaa11 commented Feb 16, 2018 • edited

huygaa11 commented Feb 16, 2018 • edited

kennknowles left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kennknowles commented Feb 16, 2018

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tgroh left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

stale bot commented Jun 7, 2018

stale bot commented Jun 14, 2018

huygaa11 commented Feb 16, 2018 •

edited

huygaa11 commented Feb 16, 2018 •

edited