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

xubii · 2019-09-27T07:57:11Z

It is the initial phase to make staled PR (#4700) compatible with the upstream master. The default implementation is done for Direct Runner.

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.

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.

…urrent 4700 pull request into my branch. And apply required changes

shehzaadn-vd · 2019-10-15T10:59:19Z

...apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexParDoOperator.java

+ timerData.getTimerId(),
+ window,
+ timerData.getTimestamp(),
+ timerData.getTimestamp(),


Why is this timerData.getTimestamp()

shehzaadn-vd · 2019-10-15T11:12:22Z

...ataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/SimpleParDoFn.java

+ timer.getTimerId(),
+ window,
+ timer.getTimestamp(),
+ timer.getTimestamp(),


this should be timer.getOutputTimestamp()

shehzaadn-vd · 2019-10-15T11:12:55Z

...ava/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillTimerInternals.java

+ StateNamespace namespace,
+ String timerId,
+ Instant timestamp,
+ Instant OutputTimestamp,


typo: case is wrong

OutputTimestamp should be outputTimestamp

shehzaadn-vd · 2019-10-15T11:13:30Z

...a/worker/src/main/java/org/apache/beam/runners/dataflow/worker/fn/control/TimerReceiver.java

@@ -119,7 +119,8 @@ public boolean receive(String pCollectionId, Object receivedElement) {
 String timerId = timerSpec.timerId();

 TimerInternals timerInternals = stepContext.namespacedToUser().timerInternals();
- timerInternals.setTimer(namespace, timerId, timer.getTimestamp(), timeDomain);
+ timerInternals.setTimer(
+ namespace, timerId, timer.getTimestamp(), timer.getTimestamp(), timeDomain);


the second timer.getTimestamp() should be timer.getOutputTimestamp()

Here timer is not belongs to TimerData.java so its don`t have getOutputTimestamp() method.

So we should add outputTimestamp to TimerInternals.

We need to make sure that the Timer object always gets constructed with an output timestamp so that we can plumb it through here.

shehzaadn-vd · 2019-10-15T11:14:17Z

runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkProcessContext.java

+ timer.getTimerId(),
+ window,
+ timer.getTimestamp(),
+ timer.getTimestamp(),


timer.getOutputTimestamp

shehzaadn-vd · 2019-10-15T11:14:46Z

runners/samza/src/main/java/org/apache/beam/runners/samza/runtime/KeyedInternals.java

+ Instant target,
+ Instant targetOutput,
+ TimeDomain timeDomain) {
+ getInternals().setTimer(namespace, timerId, target, target, timeDomain);


second "target" should be "targetOutput"

shehzaadn-vd · 2019-10-15T11:14:58Z

runners/samza/src/main/java/org/apache/beam/runners/samza/runtime/DoFnOp.java

+ timer.getTimerId(),
+ window,
+ timer.getTimestamp(),
+ timer.getTimestamp(),


timer.getOutputTimestamp()

shehzaadn-vd · 2019-10-15T11:15:18Z

runners/jet/src/main/java/org/apache/beam/runners/jet/processors/StatefulParDoP.java

@@ -92,7 +92,8 @@ private static void fireTimer(
 TimerInternals.TimerData timer, DoFnRunner<KV<?, ?>, ?> doFnRunner) {
 StateNamespace namespace = timer.getNamespace();
 BoundedWindow window = ((StateNamespaces.WindowNamespace) namespace).getWindow();
- doFnRunner.onTimer(timer.getTimerId(), window, timer.getTimestamp(), timer.getDomain());
+ doFnRunner.onTimer(
+ timer.getTimerId(), window, timer.getTimestamp(), timer.getTimestamp(), timer.getDomain());


second timer.getTimestamp() should be timer.getOutputTimestamp()

shehzaadn-vd · 2019-10-15T11:15:36Z

...main/java/org/apache/beam/runners/fnexecution/splittabledofn/SDFFeederViaStateAndTimers.java

@@ -160,7 +160,7 @@ public void commit() throws IOException {

 // Set a timer to continue processing this element.
 timerInternals.setTimer(
- stateNamespace, "sdfContinuation", wakeupTime, TimeDomain.PROCESSING_TIME);
+ stateNamespace, "sdfContinuation", wakeupTime, wakeupTime, TimeDomain.PROCESSING_TIME);


double check the second wakeupTime. Looks suspicious

shehzaadn-vd · 2019-10-15T11:16:34Z

...pex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexTimerInternals.java

+ String timerId,
+ Instant target,
+ Instant targetOutput,
+ TimeDomain timeDomain) {
 TimerData timerData = TimerData.of(timerId, namespace, target, timeDomain);


you are ignoring targetOutput

shehzaadn-vd · 2019-10-15T22:22:16Z

Any test results after commit 524d991?

xubii · 2019-10-16T05:54:46Z

Any test results after commit 524d991?

Some test cases still failing after changed

shehzaadn-vd · 2019-10-16T06:13:32Z

So how many previously failing tests are now passing? How many failures remain?

…

On Tue, Oct 15, 2019 at 10:55 PM xubii ***@***.***> wrote: Any test results after commit 524d991? Some test cases still failing after changed — You are receiving this because you commented. Reply to this email directly, view it on GitHub <#9677?email_source=notifications&email_token=AHYHLVWYBA7CUFAYDI7QWY3QO2UDPA5CNFSM4I3DCXKKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEBLF5JA#issuecomment-542531236>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AHYHLVV6H6PSQGP6VH74C5LQO2UDPANCNFSM4I3DCXKA> .

-- *Shehzaad Nakhoda* CTO USA/WhatsApp: +1 6502085107 PAK: +92 3082654179 Skype: shehzaad.nakhoda <http:https://venturedive.com/>

kennknowles · 2019-10-17T18:24:12Z

The JavaPortabilityApi test seems to have a compile error.

Test Result (4 failures / +4)

 - org.apache.beam.sdk.extensions.sql.impl.rel.BeamSortRelTest.testOrderBy_with_offset2
 - org.apache.beam.sdk.extensions.sql.zetasql.ZetaSQLDialectSpecTest.testZetaSQLSelectFromTableLimitOffset
 - org.apache.beam.sdk.io.hbase.HBaseIOTest.testReadingKeyRangeMiddleSDF
 - org.apache.beam.sdk.io.hbase.HBaseIOTest.testReadingSDF

The two SQL failures are the same: they are ZetaSQL and Calcite SQL versions of testing LIMIT with OFFSET. This uses stateful DoFn here: https://github.com/apache/beam/blob/master/sdks/java/extensions/sql/src/main/java/org/apache/beam/sdk/extensions/sql/impl/rel/BeamSortRel.java#L248

kennknowles · 2019-10-17T18:34:43Z

The HBase tests are using SplittableDoFn. It is getting a null range to the key range tracker: https://github.com/apache/beam/blob/master/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/splittabledofn/ByteKeyRangeTracker.java#L54

Since SDF is probably implemented using state & timers under the hood, this is probably a good bug to catch.

kennknowles · 2019-10-17T18:34:53Z

run dataflow validatesrunner

kennknowles · 2019-10-17T18:35:08Z

I expect that test suite to uncover some more instances of state/timers having trouble.

kennknowles · 2019-10-17T18:35:31Z

Notably, the thing that is null is never permitted to be null. So our static analysis has failed to detect a critical error somewhere.

xubii · 2019-10-18T05:47:54Z

run dataflow validatesrunner

reuvenlax · 2020-01-08T03:50:38Z

retest this please

reuvenlax · 2020-01-08T03:51:33Z

I added a test to this branch that should verify this behavior. It might require running through spotless

rehmanmuradali · 2020-01-08T11:15:54Z

Retest this please

rehmanmuradali · 2020-01-08T11:20:04Z

@reuvenlax , can you please run the jobs? Seems like I cannot run it according to new Jenkins job execution policy.

aromanenko-dev · 2020-01-08T12:48:57Z

retest this please

iemejia · 2020-01-08T13:00:46Z

retest this please

reuvenlax · 2020-01-08T16:59:37Z

retest this please

reuvenlax · 2020-01-08T17:19:10Z

Run Java PreCommit

reuvenlax · 2020-01-08T17:21:55Z

run spark validatesrunner

reuvenlax · 2020-01-08T17:22:05Z

run flink validatesrunner

reuvenlax · 2020-01-08T17:23:46Z

Run Flink ValidatesRunner

reuvenlax · 2020-01-08T17:24:21Z

Run Spark ValidatesRunner

reuvenlax · 2020-01-08T18:25:16Z

This is almost done, however you need to exclude the Flink runner from the new test, as it is failing. To do this you need to introduce a new testing tag (e.g. look at the UsesTestStream interface) and then add it to the excludeCategories list in flink_runner.gradle

rehmanmuradali · 2020-01-08T18:56:37Z

This is almost done, however you need to exclude the Flink runner from the new test, as it is failing. To do this you need to introduce a new testing tag (e.g. look at the UsesTestStream interface) and then add it to the excludeCategories list in flink_runner.gradle

@reuvenlax , done

reuvenlax · 2020-01-08T19:13:32Z

Run Flink ValidatesRunner

reuvenlax · 2020-01-08T19:15:50Z

Run Flink ValidatesRunner

boyuanzz · 2020-01-09T23:50:59Z

Seems like this PR is breaking:
Java_ValidatesRunner_Samza
Java_ValidatesRunner_Spark
Java_PVR_Flink_Streaming
Java_PVR_Flink_Batch

@xubii could you please help look into it?

boyuanzz · 2020-01-10T00:00:12Z

File JIAR here: https://issues.apache.org/jira/browse/BEAM-9083

boyuanzz · 2020-01-13T19:27:13Z

This PR also breaks dataflow post commit: https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/
https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_PortabilityApi_Dataflow/

reuvenlax · 2020-01-13T19:57:39Z

Flink tests should explicitly excluded, but maybe we didn't exclude the portable flink tests. There's a pending PR to add support for Flink. We should add exclusions for Samza, Spark, and the portableApi Dataflow runners.

…

On Mon, Jan 13, 2020 at 11:27 AM Boyuan Zhang ***@***.***> wrote: This PR also breaks dataflow post commit: https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/ https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_PortabilityApi_Dataflow/ — You are receiving this because you modified the open/close state. Reply to this email directly, view it on GitHub <#9677?email_source=notifications&email_token=AFAYJVKOLCBMPNP2RXYIEXTQ5S6BFA5CNFSM4I3DCXKKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEIZ7A3Q#issuecomment-573829230>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AFAYJVOJASD45WJPK3X3YKTQ5S6BFANCNFSM4I3DCXKA> .

boyuanzz · 2020-01-13T20:04:41Z

Flink tests should explicitly excluded, but maybe we didn't exclude the portable flink tests. There's a pending PR to add support for Flink. We should add exclusions for Samza, Spark, and the portableApi Dataflow runners.
…
On Mon, Jan 13, 2020 at 11:27 AM Boyuan Zhang @.***> wrote: This PR also breaks dataflow post commit: https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/ https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_PortabilityApi_Dataflow/ — You are receiving this because you modified the open/close state. Reply to this email directly, view it on GitHub <#9677?email_source=notifications&email_token=AFAYJVKOLCBMPNP2RXYIEXTQ5S6BFA5CNFSM4I3DCXKKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEIZ7A3Q#issuecomment-573829230>, or unsubscribe https://github.com/notifications/unsubscribe-auth/AFAYJVOJASD45WJPK3X3YKTQ5S6BFANCNFSM4I3DCXKA .

Thank Reuven for the confirmation! Is there anyone actively working on excluding these tests from the mentioned runner? If not, I'm going to work on a PR to add exclusions.

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

Nobody yet. The test is already marked with an test category (because we currently exclude the regular Flink runner). It also looks like the test should have been marked with UsesTestStream, which I think would fix the Spark problem (probably by excluding Spark). On Mon, Jan 13, 2020 at 12:04 PM Boyuan Zhang <[email protected]> wrote:

…

Flink tests should explicitly excluded, but maybe we didn't exclude the portable flink tests. There's a pending PR to add support for Flink. We should add exclusions for Samza, Spark, and the portableApi Dataflow runners. … <#m_3450155051590485810_> On Mon, Jan 13, 2020 at 11:27 AM Boyuan Zhang *@*.***> wrote: This PR also breaks dataflow post commit: https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/ https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_PortabilityApi_Dataflow/ — You are receiving this because you modified the open/close state. Reply to this email directly, view it on GitHub <#9677 <#9677>?email_source=notifications&email_token=AFAYJVKOLCBMPNP2RXYIEXTQ5S6BFA5CNFSM4I3DCXKKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEIZ7A3Q#issuecomment-573829230>, or unsubscribe https://github.com/notifications/unsubscribe-auth/AFAYJVOJASD45WJPK3X3YKTQ5S6BFANCNFSM4I3DCXKA . Thank Reuven for the confirmation! Is there anyone actively working on excluding these tests from the mentioned runner? If not, I'm going to work on a PR to add exclusions. — You are receiving this because you modified the open/close state. Reply to this email directly, view it on GitHub <#9677?email_source=notifications&email_token=AFAYJVMJA3CJYO4CZTLCXF3Q5TCNVA5CNFSM4I3DCXKKYY3PNVWWK3TUL52HS4DFVREXG43VMVBW63LNMVXHJKTDN5WW2ZLOORPWSZGOEI2DBAI#issuecomment-573845633>, or unsubscribe <https://github.com/notifications/unsubscribe-auth/AFAYJVM62OSOGTX4GIVUAFDQ5TCNVANCNFSM4I3DCXKA> .

xubii added 4 commits September 27, 2019 12:01

Beam-2535 : Pursue pull request 4700 with manual apply changes from c…

329c2e6

…urrent 4700 pull request into my branch. And apply required changes

Beam-2535 : Pursue pull request 4700 with manual apply changes from c…

f72a29a

…urrent 4700 pull request into my branch. And apply required changes

Merge https://github.com/apache/beam into time-output-timestamp

fd72944

Beam-2535 : Pursue pull request 4700 with manual apply changes from c…

81621be

…urrent 4700 pull request into my branch. And apply required changes

shehzaadn-vd reviewed Oct 15, 2019

View reviewed changes

Beam-2535 : Replace timeStamp with outputTimeStamp

5249dd1

Beam-2535 : Apply Spotless

cab84c2

Merge branch 'master' into time-output-timestamp

b0eb6f8

xubii force-pushed the time-output-timestamp branch from 8fb2714 to b0eb6f8 Compare October 18, 2019 13:30

xubii closed this Oct 18, 2019

xubii force-pushed the time-output-timestamp branch from b0eb6f8 to 7b67a92 Compare October 18, 2019 13:32

Beam-2535 : Pass outputTimestamp param in onTimer method

5aeeebf

Rehman added 3 commits January 8, 2020 15:22

Merge master into time-output-timestamp

2565959

Apply spotless

2ea7bab

Compilation Fix PardoTest

5e63afd

Reverting outputTimestamp validation

80862f2

CheckStyle Fix

149dec2

Adding Category to exclude Flink Runner

6fdb1e1

reuvenlax changed the title ~~[Beam-2535] : Apply changes from stale PR 4700 on latest code~~ [Beam-2535] : Support outputTimestamp and watermark holds in timers. Jan 8, 2020

reuvenlax merged commit 4b77225 into apache:master Jan 8, 2020

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

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

Conversation

xubii commented Sep 27, 2019 • edited Loading

Post-Commit Tests Status (on master branch)

Pre-Commit Tests Status (on master branch)

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

xubii Oct 16, 2019 • edited Loading

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

shehzaadn-vd commented Oct 15, 2019

xubii commented Oct 16, 2019

shehzaadn-vd commented Oct 16, 2019 via email

kennknowles commented Oct 17, 2019 • edited Loading

kennknowles commented Oct 17, 2019

kennknowles commented Oct 17, 2019

kennknowles commented Oct 17, 2019

kennknowles commented Oct 17, 2019

xubii commented Oct 18, 2019

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

rehmanmuradali commented Jan 8, 2020

rehmanmuradali commented Jan 8, 2020

aromanenko-dev commented Jan 8, 2020

iemejia commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

rehmanmuradali commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

reuvenlax commented Jan 8, 2020

boyuanzz commented Jan 9, 2020

boyuanzz commented Jan 10, 2020

boyuanzz commented Jan 13, 2020

reuvenlax commented Jan 13, 2020 via email

boyuanzz commented Jan 13, 2020

reuvenlax commented Jan 13, 2020 via email

xubii commented Sep 27, 2019 •

edited

Loading

xubii Oct 16, 2019 •

edited

Loading

kennknowles commented Oct 17, 2019 •

edited

Loading