[BEAM-2535] Support outputTimestamp and watermark holds in processing timers. #10627

rehmanmuradali · 2020-01-17T14:31:16Z

Please add a meaningful description for your change here

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Choose reviewer(s) and mention them in a comment (R: @username).
Format the pull request title like [BEAM-XXX] Fixes bug in ApproximateQuantiles, where you replace BEAM-XXX with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

Post-Commit Tests Status (on master branch)

Lang	SDK	Apex	Dataflow	Gearpump	Samza	Spark
Go		---	---	---	---
Java
Python		---		---	---
XLang	---	---	---	---	---	---

Pre-Commit Tests Status (on master branch)

---	Java	Python	Go	Website
Non-portable
Portable	---		---	---

See .test-infra/jenkins/README for trigger phrase, status and link of all Jenkins jobs.

rehmanmuradali · 2020-01-20T18:17:40Z

R: @reuvenlax

aaltay · 2020-01-23T17:10:38Z

Pinging to trigger tests.

amaliujia · 2020-01-23T18:09:44Z

Run Java PreCommit

amaliujia · 2020-01-23T18:09:54Z

Run JavaPortabilityApi PreCommit

rehmanmuradali · 2020-01-27T17:16:38Z

@reuvenlax I have completed outputTimestamp functionality. Need your input with the failed test cases. I think it is expected with the addition of this feature.

iemejia · 2020-01-28T13:26:42Z

retest this please

iemejia · 2020-01-28T13:28:08Z

Run Direct ValidatesRunner

rehmanmuradali · 2020-01-28T18:20:29Z

@reuvenlax really need your input with the failed test cases as the functionality is completed.

mwalenia · 2020-01-29T12:52:10Z

Run Java PreCommit

reuvenlax · 2020-01-29T20:18:35Z

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

+    public Timer withOutputTimestampOffset(Duration outputTimestampOffset) {
+      this.outputTimestampOffset = outputTimestampOffset;
+      return this;
+    }


I'm not sure that we need withOutputTimestampOffset - I think withOutputTimestamp is sufficient.

@reuvenlax done

reuvenlax · 2020-01-29T20:22:23Z

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

+                : target.minus(offset.minus(outputTimestampOffset));
+      }
+
      if (TimeDomain.EVENT_TIME.equals(spec.getTimeDomain())) {


I think that we should verify that the output timestamp is > the timestamp of the input message (if in processElement) or the output timestamp of the firing timer (if in processTimer). The < check remains correct - even for processing-time timers.

@reuvenlax , does that mean we need to compare output timestamp with timerInternal.currentInputWatermarkTime()? I think it will fail our previous test case of output timestamp with event timer as we have set output timestamp as 5 while having a timestamp of input element as 9.

@reuvenlax done

reuvenlax · 2020-01-29T20:23:11Z

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

-      if (outputTimestamp == null) {
+      if (outputTimestamp == null && TimeDomain.EVENT_TIME.equals(spec.getTimeDomain())) {
        outputTimestamp = target;
      }


I think that if the timer is processing time, then then default outputTimestamp should be that of the input element (or the output time of the firing timer if in processTimer).

@reuvenlax done

…nd firing timer

reuvenlax · 2020-01-30T20:13:42Z

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

        TimerSpec spec = (TimerSpec) signature.timerDeclarations().get(timerId).field().get(fn);
        return new TimerInternalsTimer(
-            window, getNamespace(), timerId, spec, stepContext.timerInternals());
+            window, getNamespace(), timerId, spec, fireTimestamp(), stepContext.timerInternals());


I think this should be timestamp(), not fireTimestamp().

@reuvenlax done

reuvenlax · 2020-01-30T20:13:57Z

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

+            window(),
+            getNamespace(),
+            spec,
+            fireTimestamp(),


ditto - this should be timestamp() not fireTimestamp()

reuvenlax · 2020-01-30T20:15:39Z

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

+          outputTimestamp,
+          elementInputTimestamp);

      if (TimeDomain.EVENT_TIME.equals(spec.getTimeDomain())) {


check outputTimestamp is in the window for processing-timer as well.

@reuvenlax done

reuvenlax · 2020-01-30T20:17:46Z

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java

+          new DoFn<KV<String, Integer>, Integer>() {
+
+            @TimerId(timerId)
+            private final TimerSpec timer = TimerSpecs.timer(TimeDomain.PROCESSING_TIME);


I think you want the second DoFn to set an event-time timer. That way you can test that the watermark hold in the first DoFn prevents firing of the second DoFn's timer, which it will only do if the second DoFn's timer is even time.

@reuvenlax updated

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

reuvenlax · 2020-01-30T20:25:14Z

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

      Instant earliest = THE_END_OF_TIME.get();
      for (NavigableSet<TimerData> timers : processingTimers.values()) {
        if (!timers.isEmpty()) {
-          earliest = INSTANT_ORDERING.min(timers.first().getTimestamp(), earliest);


I wonder if this is sufficient. The NavigableSet is ordered by timer firing time, not by outputTimestamp. This means that the logic to just look at the first timer (which was correct before we had holds) is probably no longer enough. we probably need to iterate over the set (or we need to have a second data structure to hold the watermark holds and take the minimum of that data structure).

@reuvenlax added a new function to calculate the earliest outputTimestamp from NavigableSet.

…, firing time fix, test case updates

iemejia · 2020-01-31T13:09:17Z

retest this please

iemejia · 2020-01-31T13:09:25Z

Run Direct ValidatesRunner

reuvenlax · 2020-02-02T00:38:48Z

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

-            target,
-            windowExpiry);
+      // For processing timers
+      if (outputTimestamp == null) {


are we missing an else clause here?

@reuvenlax we only set outputTimestamp here. I think If that is already set before then we don't need anything. I think else clause is not required here.

reuvenlax · 2020-02-02T01:21:00Z

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

    private final TimerSpec spec;
    private Instant target;
    private Instant outputTimestamp;
+    private Instant elementInputTimestamp;


@reuvenlax done

reuvenlax · 2020-02-02T03:33:29Z

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

+          !outputTimestamp.isBefore(elementInputTimestamp),
+          "output timestamp %s should be after input message timestamp or output timestamp of firing timers %s",
+          outputTimestamp,
+          elementInputTimestamp);


check this before setting outputTimestamp above (and only if outputTimestamp != null)

@reuvenlax done

reuvenlax · 2020-02-02T07:47:58Z

retest this please

iemejia · 2020-02-02T14:16:03Z

retest this please

reuvenlax · 2020-02-02T18:16:32Z

Run Java PreCommit

kennknowles · 2020-02-05T04:06:54Z

run dataflow validatesrunner

kennknowles · 2020-02-05T04:07:05Z

run flink validatesrunner

reuvenlax · 2020-02-05T22:33:40Z

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java

+              // DoFn timer's watermark hold. This timer should not fire until the previous timer
+              // fires and removes
+              // the watermark hold.
+              timer.offset(Duration.standardSeconds(8)).setRelative();


just use timer.set, not timer.offset

@reuvenlax done

reuvenlax · 2020-02-05T22:40:49Z

sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java

+              .addElements(KV.of("key", 1))
+              // Normally this would case fn2's timer to expire, but it shouldn't here because of
+              // the output timestamp.
+              .advanceProcessingTime(Duration.standardSeconds(9))


I think you might have to advance processing time to at least 11.

You also need to advance the watermark to at least 10 to allow the timer to fire. Right now this isn't testing anything, because the input watermark is preventing the timer from firing.

@reuvenlax, I think if we advance processing time to 11, it will fire fn1's timer as we have set delivery time offset as 10. If we advance processing time by 9 and watermark by 11 then it will test the functionality?

yes, but you need to advance the watermark before calling addElements.

@reuvenlax done

reuvenlax · 2020-02-05T22:42:52Z

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

     */
    public synchronized Instant getEarliestTimerTimestamp() {
      Instant earliest = THE_END_OF_TIME.get();
      for (NavigableSet<TimerData> timers : processingTimers.values()) {


What about the other getEarliestTimerTimestamp function? Does that need to be updated?

@reuvenlax done

aaltay · 2020-02-07T18:09:45Z

Pinging to trigger tests.

reuvenlax · 2020-02-09T06:56:07Z

Run Python2_PVR_Flink PreCommit

rehmanmuradali and others added 3 commits January 17, 2020 19:29

Adding outputTimestamp for processing Timers

de6d3ab

Calculating outputTimestamp with outputTimestampOffset

695db95

Correcting calculation of outputtimestamp

41cbfb5

Updating test case

00dceb3

rehmanmuradali requested a review from reuvenlax January 22, 2020 07:49

rehmanmuradali added 2 commits January 23, 2020 21:50

Updating Input and Output Watermarks for processing timers

e41d391

Adding outputTimestamp category in test

a0b2e2e

rehmanmuradali added 3 commits January 24, 2020 13:03

Merge branch 'master' into feature/processing-time

a33f137

Add missing imports

87906d1

Updating firedTime for processingTimers

17f401e

rehmanmuradali requested a review from kennknowles January 27, 2020 17:34

rehmanmuradali added 2 commits January 28, 2020 16:09

Updating OnTimer, Removed equal checks in test cases

ef4407f

Merge branch 'master' into feature/processing-time

aa1a938

reuvenlax reviewed Jan 29, 2020

View reviewed changes

rehmanmuradali added 4 commits January 30, 2020 19:29

Removing outputTimestamp Offset

e27d620

Adding validation of output timestamp

1d1ee22

Adding validation of output timestamp with respect to input element a…

cac7b67

…nd firing timer

Minor fixes

54e8977

reuvenlax reviewed Jan 30, 2020

View reviewed changes

rehmanmuradali added 2 commits January 31, 2020 14:21

Input Element timestamp fix, processing timer after window validation…

311a38d

…, firing time fix, test case updates

Get earliest outputTimestamp

b7cd4a5

Adding outputtimestamp validation

4c6b602

reuvenlax reviewed Feb 2, 2020

View reviewed changes

rehmanmuradali added 2 commits February 2, 2020 12:38

Output Timestamp validation fix

d085688

Merge branch 'master' into feature/processing-time

c45ecc6

reuvenlax reviewed Feb 5, 2020

View reviewed changes

Test case fix, getMinimumOutputTimestamp fix

63f3a71

reuvenlax merged commit a005fd7 into apache:master Feb 9, 2020

[BEAM-2535] Support outputTimestamp and watermark holds in processing timers. #10627

[BEAM-2535] Support outputTimestamp and watermark holds in processing timers. #10627

Uh oh!

Conversation

rehmanmuradali commented Jan 17, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

Uh oh!

rehmanmuradali commented Jan 20, 2020

Uh oh!

aaltay commented Jan 23, 2020

Uh oh!

amaliujia commented Jan 23, 2020

Uh oh!

amaliujia commented Jan 23, 2020

Uh oh!

rehmanmuradali commented Jan 27, 2020

Uh oh!

iemejia commented Jan 28, 2020

Uh oh!

iemejia commented Jan 28, 2020

Uh oh!

rehmanmuradali commented Jan 28, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mwalenia commented Jan 29, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

rehmanmuradali Jan 30, 2020 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

iemejia commented Jan 31, 2020

Uh oh!

iemejia commented Jan 31, 2020

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

rehmanmuradali commented Jan 17, 2020 •

edited

Loading

rehmanmuradali commented Jan 28, 2020 •

edited

Loading

rehmanmuradali Jan 30, 2020 •

edited

Loading

rehmanmuradali Feb 6, 2020 •

edited

Loading