[BEAM-8543] Dataflow streaming timers are not strictly time ordered when set earlier mid-bundle #11924

rehmanmuradali · 2020-06-04T19:58:45Z

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.
Update CHANGES.md with noteworthy changes.
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
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-06-04T20:08:07Z

R: @reuvenlax
Could you please take a look that I am on right track?

reuvenlax · 2020-06-05T20:24:25Z

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java

+      if (userTimerInternals.hasTimerBefore(currentInputWatermark)) {
+        while (!toBeFiredTimersOrdered.isEmpty()) {
+          userTimerInternals.setTimer(toBeFiredTimersOrdered.poll());
+        }
      }


@kennknowles for comment. This doesn't look right to me, as I don't think we should be modifying the WindmillTimerInternals here. I think we just want to merge the timer modifications from processing the workitem into this priority queue; note that if timers are deleted, we need to detect that as well and remove from the priority queue.

Yea I don't actually understand what this block is for.

FWIW to do timer deletion/reset cheaply without building a bespoke data structure just keep a map from id to firing time or tombstone. This way, whenever a timer comes up in the prio queue you pull out the actual time for it from the map. If it is actually set for another time, don't fire it. If it is obsolete, don't fire it.

@reuvenlax done

kennknowles · 2020-06-05T21:27:10Z

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java

@@ -577,12 +583,21 @@ public void flushState() {
                        WindmillTimerInternals.windmillTimerToTimerData(
                            WindmillNamespacePrefix.USER_NAMESPACE_PREFIX, timer, windowCoder))
                .iterator();
+
+        cachedFiredUserTimers.forEachRemaining(toBeFiredTimersOrdered::add);


Do we even need cachedFiredUserTimers? It seems obsolete if we populate the priority queue. The name is also wrong - even before this PR it wasn't a cache. It is a lazily initialized iterator. Instead, we should have a lazily initialized priority queue (like you do) and just a flag to say whether the incoming timers have been loaded yet.

kennknowles · 2020-06-05T21:31:27Z

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java

+      if (userTimerInternals.hasTimerBefore(currentInputWatermark)) {
+        while (!toBeFiredTimersOrdered.isEmpty()) {
+          userTimerInternals.setTimer(toBeFiredTimersOrdered.poll());
+        }
      }


Yea I don't actually understand what this block is for.

FWIW to do timer deletion/reset cheaply without building a bespoke data structure just keep a map from id to firing time or tombstone. This way, whenever a timer comes up in the prio queue you pull out the actual time for it from the map. If it is actually set for another time, don't fire it. If it is obsolete, don't fire it.

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

ajamato · 2020-06-18T23:47:42Z

@kennknowles Would you mind following up on this PR?

kennknowles · 2020-07-23T21:09:50Z

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java

-import java.util.List;
-import java.util.Map;
-import java.util.Set;
+import java.util.*;


Please keep the imports all explicit.

kennknowles · 2020-07-23T21:11:06Z

...ker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java

-    private Iterator<TimerData> cachedFiredUserTimers = null;
+    private PriorityQueue<TimerData> toBeFiredTimersOrdered = null;
+
+    // to track if timer is reset earlier mid-bundle.


Can you add a comment about what are the keys and values of this map?

kennknowles · 2020-07-23T21:17:33Z

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

    }

    private void testEventTimeTimerOrderingWithInputPTransform(
        Instant now,
        int numTestElements,
-        PTransform<PBegin, PCollection<KV<String, String>>> transform)
+        PTransform<PBegin, PCollection<KV<String, String>>> transform,
+        boolean isStreaming)


This doesn't depend on streaming or not, but just controls whether the pcollection should be bounded or unbounded. For clarity, you can just make this parameter IsBounded isBounded

Changed to IsBounded

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

kennknowles · 2020-07-23T21:18:04Z

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

      }

      private final Instant start;
      private final Instant end;
+      private final boolean isStreaming;


kennknowles · 2020-07-23T21:18:11Z

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

+          Instant start,
+          Instant end,
+          PTransform<PBegin, PCollection<KV<Void, Void>>> input,
+          boolean isStreaming) {


kennknowles · 2020-07-23T21:19:37Z

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

@@ -4331,6 +4378,7 @@ public PDone expand(PBegin input) {
        PCollection<String> result =
            input
                .apply(inputPTransform)
+                .setIsBoundedInternal(isStreaming ? IsBounded.UNBOUNDED : IsBounded.BOUNDED)


Another way to do this that might be better is to use TestStream in the unbounded case. This will probably give best coverage. Even for an unbounded PCollection the watermark might instantly move to infinity.

# Conflicts: # runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingModeExecutionContext.java # runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillTimerInternals.java

rehmanmuradali · 2020-07-29T14:10:35Z

R: @kennknowles @reuvenlax

kennknowles

I'm ready to merge this. Thanks for sticking with us through the review. I will squash the commits since they are all incremental and the overall PR is one change.

kennknowles · 2020-07-29T16:58:36Z

run dataflow validatesrunner

lukecwik · 2020-08-07T22:15:08Z

This PR enabled new tests which are failing with some of the validates runner tests. for example:
https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/7773/testReport/
https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/7805/testReport/

Expanded existing JIRA: https://issues.apache.org/jira/browse/BEAM-8460, opened #12503 to disable failing test category.

…rdered when set earlier mid-bundle (apache#11924)" This reverts commit 88acc52. There is a bug in this commit that causes deleted timers to not clear their watermark holds, resulting in stuck pipelines. See https://issues.apache.org/jira/browse/BEAM-10991

…streaming timers are not strictly time ordered when set earlier mid-bundle (#11924)"

…rdered when set earlier mid-bundle (apache#11924)" This reverts commit 88acc52. There is a bug in this commit that causes deleted timers to not clear their watermark holds, resulting in stuck pipelines. See https://issues.apache.org/jira/browse/BEAM-10991

[BEAM-10991] Cherrypick #12980 to 2.25.0: Revert "[BEAM-8543] Dataflow streaming timers are not strictly time ordered when set earlier mid-bundle (#11924)"

validated timer ordering

9c1dd6a

probot-autolabeler bot added dataflow java runners labels Jun 4, 2020

update changes.md

b73c770

reuvenlax requested a review from kennknowles June 5, 2020 19:54

reuvenlax reviewed Jun 5, 2020

View reviewed changes

kennknowles requested changes Jun 5, 2020

View reviewed changes

rehmanmuradali added 3 commits June 9, 2020 18:53

Strict timer ordering without resetting timer

fba4c3d

maintain strict ordering + timer reset

032bde5

adding comments

abd93ec

kennknowles self-requested a review July 14, 2020 22:34

kennknowles reviewed Jul 23, 2020

View reviewed changes

rehmanmuradali added 2 commits July 27, 2020 13:47

Replace isStreaming with isBounded

10487f0

kennknowles approved these changes Jul 29, 2020

View reviewed changes

kennknowles merged commit 88acc52 into apache:master Jul 29, 2020

lukecwik mentioned this pull request Aug 7, 2020

[BEAM-8460] Exclude category containing failing tests for spark/flink to restore green test signal. #12503

Merged

4 tasks

kennknowles added a commit that referenced this pull request Oct 1, 2020

Merge pull request #12980: [BEAM-10991] Revert "[BEAM-8543] Dataflow …

75e38f7

…streaming timers are not strictly time ordered when set earlier mid-bundle (#11924)"

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[BEAM-8543] Dataflow streaming timers are not strictly time ordered when set earlier mid-bundle #11924

[BEAM-8543] Dataflow streaming timers are not strictly time ordered when set earlier mid-bundle #11924

rehmanmuradali commented Jun 4, 2020

rehmanmuradali commented Jun 4, 2020

reuvenlax Jun 5, 2020

kennknowles Jun 5, 2020

rehmanmuradali Jun 10, 2020

kennknowles Jun 5, 2020

rehmanmuradali Jun 10, 2020

kennknowles Jun 5, 2020

ajamato commented Jun 18, 2020

kennknowles Jul 23, 2020

kennknowles Jul 23, 2020

rehmanmuradali Jul 29, 2020

kennknowles Jul 23, 2020

rehmanmuradali Jul 29, 2020

kennknowles Jul 23, 2020

rehmanmuradali Jul 29, 2020

kennknowles Jul 23, 2020

rehmanmuradali Jul 29, 2020

kennknowles Jul 23, 2020

rehmanmuradali commented Jul 29, 2020 •

edited

kennknowles left a comment

kennknowles commented Jul 29, 2020

lukecwik commented Aug 7, 2020

[BEAM-8543] Dataflow streaming timers are not strictly time ordered when set earlier mid-bundle #11924

[BEAM-8543] Dataflow streaming timers are not strictly time ordered when set earlier mid-bundle #11924

Conversation

rehmanmuradali commented Jun 4, 2020

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

rehmanmuradali commented Jun 4, 2020

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

ajamato commented Jun 18, 2020

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

rehmanmuradali commented Jul 29, 2020 • edited

kennknowles left a comment

Choose a reason for hiding this comment

kennknowles commented Jul 29, 2020

lukecwik commented Aug 7, 2020

rehmanmuradali commented Jul 29, 2020 •

edited