You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2021/09/20 23:21:03 UTC
[GitHub] [beam] laraschmidt opened a new pull request #15540: Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
laraschmidt opened a new pull request #15540:
URL: https://github.com/apache/beam/pull/15540
A DoFn may emit elements with a timestamp up to DoFn#getAllowedTimestampSkew() before the current element's timestamp. This change implements this change for timer's as well. Now a timer may have an output timestamp up to DoFn#getAllowedTimestampSkew() before the current element's timestamp. Before this change a timer's output timestamp could not be before the current output element.
Additional Context: https://lists.apache.org/thread.html/r7554658114ddde86c5d82e1c39fe7e1ef587fe926b8e406d1130d501%40%3Cdev.beam.apache.org%3E
@reuvenlax
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) 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](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
`ValidatesRunner` compliance status (on master branch)
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_ULR/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Streaming/lastCompletedBuild/badge/icon?subject=V1+Streaming">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/badge/icon?subject=V1+Java+11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2_Streaming/lastCompletedBuild/badge/icon?subject=V2+Streaming">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/badge/icon?subject=Java+8">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/badge/icon?subject=Java+11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/badge/icon?subject=Portable+Streaming">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Samza/lastCompletedBuild/badge/icon?subject=Portable">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/badge/icon">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/badge/icon?subject=Structured+Streaming">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Python</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/badge/icon?subject=ValCont">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/badge/icon?subject=Portable">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Samza/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Samza/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
</tr>
</tbody>
</table>
Examples testing status on various runners
--------------------------------------------------------
<table>
<thead>
<tr>
<th>Lang</th>
<th>ULR</th>
<th>Dataflow</th>
<th>Flink</th>
<th>Samza</th>
<th>Spark</th>
<th>Twister2</th>
</tr>
</thead>
<tbody>
<tr>
<td>Go</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Java</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Cron/lastCompletedBuild/badge/icon?subject=V1">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Examples_Dataflow_Java11_Cron/lastCompletedBuild/badge/icon?subject=V1+Java11">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java_Examples_Dataflow_V2/lastCompletedBuild/badge/icon?subject=V2">
</a><br>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>Python</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
<tr>
<td>XLang</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
Post-Commit SDK/Transform Integration Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>Go</th>
<th>Java</th>
<th>Python</th>
</tr>
</thead>
<tbody>
<tr>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/badge/icon?subject=3.6">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/badge/icon?subject=3.7">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/badge/icon?subject=3.8">
</a>
</td>
</tr>
</tbody>
</table>
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
<table>
<thead>
<tr>
<th>---</th>
<th>Java</th>
<th>Python</th>
<th>Go</th>
<th>Website</th>
<th>Whitespace</th>
<th>Typescript</th>
</tr>
</thead>
<tbody>
<tr>
<td>Non-portable</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/badge/icon">
</a><br>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/badge/icon?subject=Tests">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/badge/icon?subject=Lint">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/badge/icon?subject=Docker">
</a><br>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/badge/icon?subject=Docs">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
</tr>
<tr>
<td>Portable</td>
<td>---</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>
<a href="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/">
<img alt="Build Status" src="https://ci-beam.apache.org/job/beam_PreCommit_GoPortable_Cron/lastCompletedBuild/badge/icon">
</a>
</td>
<td>---</td>
<td>---</td>
<td>---</td>
</tr>
</tbody>
</table>
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r719816647
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -472,6 +659,40 @@ 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 = "testTimerFamily";
+ private final Duration allowedSkew;
+
+ @TimerFamily(TIMER_ID)
+ private final TimerSpec timers = TimerSpecs.timerMap(TimeDomain.PROCESSING_TIME);
Review comment:
The elements need to share a timer then but I suppose the same logic will hit whenever we set the output timestamp. I removed the output of elements though since this would be harder. Wasn't really needed anyway.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r731267716
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -386,6 +397,186 @@ public void testInfiniteSkew() {
BoundedWindow.TIMESTAMP_MAX_VALUE));
}
+ @Test
+ @Category({ValidatesRunner.class, UsesTimersInParDo.class})
+ public void testRunnerNoSkew() {
+ PCollection<Duration> input =
+ p.apply(
+ "create",
+ Create.timestamped(
+ Arrays.asList(new Duration(0L), new Duration(1L)), Arrays.asList(0L, 1L)));
+ input.apply(ParDo.of(new SkewingDoFn(Duration.ZERO)));
+ // The errors differ between runners but at least check that the output timestamp is printed.
+ thrown.expectMessage(String.format("%s", new Instant(0L)));
+ thrown.expectMessage("output");
+ p.run();
+ }
+
+ @Test
+ @Category({UsesTimersInParDo.class, ValidatesRunner.class})
+ public void testRunnerAllowedSkew() {
+ PCollection<Duration> input =
+ p.apply(
+ "create",
+ Create.timestamped(
+ Arrays.asList(new Duration(0L), new Duration(1L)), Arrays.asList(0L, 1L)));
+ input.apply(ParDo.of(new SkewingDoFn(new Duration(2L))));
+ p.run();
+ }
+
+ /**
+ * Demonstrates that attempting to set a timer with an output timestamp before the timestamp of
+ * the current element with zero {@link DoFn#getAllowedTimestampSkew() allowed timestamp skew}
+ * throws.
+ */
+ @Test
+ public void testTimerBackwardsInTimeNoSkew() {
+ TimerSkewingDoFn fn = new TimerSkewingDoFn(Duration.ZERO);
+ DoFnRunner<KV<String, Duration>, Duration> runner =
+ new SimpleDoFnRunner<>(
+ null,
+ fn,
+ NullSideInputReader.empty(),
+ new ListOutputManager(),
+ new TupleTag<>(),
+ Collections.emptyList(),
+ mockStepContext,
+ null,
+ Collections.emptyMap(),
+ WindowingStrategy.of(new GlobalWindows()),
+ DoFnSchemaInformation.create(),
+ Collections.emptyMap());
+
+ runner.startBundle();
+ // A timer with output timestamp at the current timestamp is fine.
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(KV.of("1", Duration.ZERO), new Instant(0)));
+
+ thrown.expect(UserCodeException.class);
+ thrown.expectCause(isA(IllegalArgumentException.class));
+ thrown.expectMessage("output timestamp of firing timers");
+ thrown.expectMessage(
+ String.format("output timestamp %s", new Instant(0).minus(Duration.millis(1L))));
+ thrown.expectMessage(
+ String.format(
+ "allowed skew %s", PeriodFormat.getDefault().print(Duration.ZERO.toPeriod())));
+
+ // A timer with output timestamp before (current time - skew) is forbidden
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of("2", Duration.millis(1L)), new Instant(0)));
+ }
+
+ /**
+ * Demonstrates that attempting to have a timer with output timestamp before the timestamp of the
+ * current element plus the value of {@link DoFn#getAllowedTimestampSkew()} throws, but between
+ * that value and the current timestamp succeeds.
+ */
+ @Test
+ public void testTimerSkew() {
+ TimerSkewingDoFn fn = new TimerSkewingDoFn(Duration.standardMinutes(10L));
+ DoFnRunner<KV<String, Duration>, Duration> runner =
+ new SimpleDoFnRunner<>(
+ null,
+ fn,
+ NullSideInputReader.empty(),
+ new ListOutputManager(),
+ new TupleTag<>(),
+ Collections.emptyList(),
+ mockStepContext,
+ null,
+ Collections.emptyMap(),
+ WindowingStrategy.of(new GlobalWindows()),
+ DoFnSchemaInformation.create(),
+ Collections.emptyMap());
+
+ runner.startBundle();
+ // Timer with output timestamp between "now" and "now - allowed skew" succeeds.
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of("1", Duration.standardMinutes(5L)), new Instant(0)));
+ thrown.expect(UserCodeException.class);
+ thrown.expectCause(isA(IllegalArgumentException.class));
+ thrown.expectMessage("output timestamp of firing timers");
+ thrown.expectMessage(
+ String.format("output timestamp %s", new Instant(0).minus(Duration.standardHours(1L))));
+ thrown.expectMessage(
+ String.format(
+ "allowed skew %s",
+ PeriodFormat.getDefault().print(Duration.standardMinutes(10L).toPeriod())));
+ // Timer with output timestamp before "now - allowed skew" fails.
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of("2", Duration.standardHours(1L)), new Instant(0)));
+ }
+
+ /**
+ * Demonstrates that attempting to output an element with a timestamp before the current one
+ * always succeeds when {@link DoFn#getAllowedTimestampSkew()} is equal to {@link Long#MAX_VALUE}
+ * milliseconds.
+ */
+ @Test
+ public void testTimerInfiniteSkew() {
+ TimerSkewingDoFn fn = new TimerSkewingDoFn(Duration.millis(Long.MAX_VALUE));
+ DoFnRunner<KV<String, Duration>, Duration> runner =
+ new SimpleDoFnRunner<>(
+ null,
+ fn,
+ NullSideInputReader.empty(),
+ new ListOutputManager(),
+ new TupleTag<>(),
+ Collections.emptyList(),
+ mockStepContext,
+ null,
+ Collections.emptyMap(),
+ WindowingStrategy.of(new GlobalWindows()),
+ DoFnSchemaInformation.create(),
+ Collections.emptyMap());
+
+ runner.startBundle();
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of("1", Duration.millis(1L)), new Instant(0)));
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of("2", Duration.millis(1L)),
+ BoundedWindow.TIMESTAMP_MIN_VALUE.plus(Duration.millis(1))));
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of(
+ "3",
+ // This is the maximum amount a timestamp in beam can move (from the maximum
+ // timestamp
+ // to the minimum timestamp).
+ Duration.millis(BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis())
+ .minus(Duration.millis(BoundedWindow.TIMESTAMP_MIN_VALUE.getMillis()))),
+ BoundedWindow.TIMESTAMP_MAX_VALUE));
+ }
+
+ @Test
+ @Category({UsesTimersInParDo.class, ValidatesRunner.class})
+ public void testRunnerTimerNoSkew() {
+ List<KV<String, Duration>> durations =
+ Arrays.asList(KV.of("0", new Duration(0L)), KV.of("2", new Duration(1L)));
+ PCollection<KV<String, Duration>> input =
+ p.apply("create", Create.timestamped(durations, Arrays.asList(0L, 2L)));
+ input.apply(ParDo.of(new TimerSkewingDoFn(Duration.ZERO)));
+ // The errors differ between runners but at least check that the output timestamp is printed.
+ thrown.expectMessage(String.format("%s", new Instant(1L)));
Review comment:
Swapped to using PAssert and checking the PCollection output for the validates runner tests.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r719818746
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -386,6 +399,180 @@ public void testInfiniteSkew() {
BoundedWindow.TIMESTAMP_MAX_VALUE));
}
+ @Test
+ @Category({ValidatesRunner.class})
Review comment:
Done. Assuming just the validates runner tests need this because I don't see it elsewhere in the file. Let me know if not.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r728302215
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -437,19 +437,24 @@ public Instant timestamp() {
@SuppressWarnings("deprecation") // Allowed Skew is deprecated for users, but must be respected
private void checkTimestamp(Instant timestamp) {
- // The documentation of getAllowedTimestampSkew explicitly permits Long.MAX_VALUE to be used
- // for infinite skew. Defend against underflow in that case for timestamps before the epoch
- if (fn.getAllowedTimestampSkew().getMillis() != Long.MAX_VALUE
- && timestamp.isBefore(elem.getTimestamp().minus(fn.getAllowedTimestampSkew()))) {
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elem.getTimestamp().minus(fn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (timestamp.isBefore(lowerBound) || timestamp.isAfter(upperBound)) {
Review comment:
done
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,13 +1195,24 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
- checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
- "output timestamp %s should be after input message timestamp or output timestamp of firing timers %s",
- outputTimestamp,
- elementInputTimestamp);
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elementInputTimestamp.minus(fn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (outputTimestamp.isBefore(lowerBound) || outputTimestamp.isAfter(upperBound)) {
Review comment:
done.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r746172154
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1791,6 +1791,119 @@ public void finishBundle(FinishBundleContext c) {
/** Tests to validate output timestamps. */
@RunWith(JUnit4.class)
public static class TimestampTests extends SharedTestBase implements Serializable {
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<Duration, String> {
Review comment:
done.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-925394649
Added checks in FnApiDoFnRunner which was not checking for skew on timers and not checking timestamps at all for normal emit. PTAL and make sure it makes sense and I can prob add some tests in FnApiRunner.
PTAL @reuvenlax @lukecwik Thanks!
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #15540: Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r713213665
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -386,6 +388,89 @@ public void testInfiniteSkew() {
BoundedWindow.TIMESTAMP_MAX_VALUE));
}
+ /**
+ * Demonstrates that attempting to set a timer with an output timestamp before the timestamp of
+ * the current element with zero {@link DoFn#getAllowedTimestampSkew() allowed timestamp skew}
+ * throws.
+ */
+ @Test
+ public void testTimerBackwardsInTimeNoSkew() {
Review comment:
It would make more sense to make these tests `@ValidatesRunner` tests then use the deprecated `DoFnRunner`. This will allow us to get coverage across multiple implementations.
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,12 +1190,12 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
Review comment:
Do we need to make a similar change in other locations like `FnApiDoFnRunner.java`?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] edited a comment on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
codecov[bot] edited a comment on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-953406612
# [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#15540](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7b0de72) into [master](https://codecov.io/gh/apache/beam/commit/92ea5f547e2ac3dfd46e20b7c965f8a4850f2238?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (92ea5f5) will **decrease** coverage by `0.29%`.
> The diff coverage is `n/a`.
> :exclamation: Current head 7b0de72 differs from pull request most recent head bfaac5a. Consider uploading reports for the commit bfaac5a to get more accurate results
[](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #15540 +/- ##
==========================================
- Coverage 83.82% 83.53% -0.30%
==========================================
Files 441 445 +4
Lines 59727 61210 +1483
==========================================
+ Hits 50068 51131 +1063
- Misses 9659 10079 +420
```
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [sdks/python/apache\_beam/io/gcp/bigquery.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vZ2NwL2JpZ3F1ZXJ5LnB5) | `62.72% <0.00%> (-13.66%)` | :arrow_down: |
| [...n/apache\_beam/ml/gcp/recommendations\_ai\_test\_it.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWwvZ2NwL3JlY29tbWVuZGF0aW9uc19haV90ZXN0X2l0LnB5) | `56.81% <0.00%> (-12.95%)` | :arrow_down: |
| [...thon/apache\_beam/runners/worker/operation\_specs.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvb3BlcmF0aW9uX3NwZWNzLnB5) | `40.67% <0.00%> (-4.90%)` | :arrow_down: |
| [...ython/apache\_beam/io/gcp/bigquery\_read\_internal.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vZ2NwL2JpZ3F1ZXJ5X3JlYWRfaW50ZXJuYWwucHk=) | `53.92% <0.00%> (-4.55%)` | :arrow_down: |
| [sdks/python/apache\_beam/io/gcp/bigquery\_tools.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vZ2NwL2JpZ3F1ZXJ5X3Rvb2xzLnB5) | `82.91% <0.00%> (-4.52%)` | :arrow_down: |
| [...ks/python/apache\_beam/runners/worker/data\_plane.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvZGF0YV9wbGFuZS5weQ==) | `87.50% <0.00%> (-3.11%)` | :arrow_down: |
| [...ache\_beam/runners/interactive/recording\_manager.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9pbnRlcmFjdGl2ZS9yZWNvcmRpbmdfbWFuYWdlci5weQ==) | `96.55% <0.00%> (-2.43%)` | :arrow_down: |
| [sdks/python/apache\_beam/coders/coder\_impl.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vY29kZXJzL2NvZGVyX2ltcGwucHk=) | `94.04% <0.00%> (-1.25%)` | :arrow_down: |
| [.../apache\_beam/io/gcp/datastore/v1new/datastoreio.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vZ2NwL2RhdGFzdG9yZS92MW5ldy9kYXRhc3RvcmVpby5weQ==) | `86.45% <0.00%> (-0.99%)` | :arrow_down: |
| [...che\_beam/runners/interactive/augmented\_pipeline.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9pbnRlcmFjdGl2ZS9hdWdtZW50ZWRfcGlwZWxpbmUucHk=) | `69.38% <0.00%> (-0.62%)` | :arrow_down: |
| ... and [100 more](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [92ea5f5...bfaac5a](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-931773284
> Won't this allow for infinite skew since if have a timer at `X` and skew of `-1` then the first time the timer is processed you can output at time `X-1` and when it gets scheduled again you can now output at `X-2` since the the new timers timestamp is `X-1`?
So my understanding of the reason for these checks is to stop people from doing the wrong thing without realizing it. We don't even take any different action based on this variable. It seems okay to apply this to each specific output timestamp and let you skew more if you chain timers in this fashion.
On a more practical note, there's reasons why you might want a timer to output an earlier element if you've properly set up watermark holds. There's currently no way to do that so we need some allowance. It would probably be better if we could constrain skew from the first output timestamp but I don't think that's available in the later timers, right?
If you disagree with the approach, I can bring this up on the email thread for others to chime in in case they are not checking here.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r719826325
##########
File path: sdks/java/harness/src/test/java/org/apache/beam/fn/harness/FnApiDoFnRunnerTest.java
##########
@@ -164,7 +166,7 @@
@SuppressWarnings({
"rawtypes", // TODO(https://issues.apache.org/jira/browse/BEAM-10556)
})
-public class FnApiDoFnRunnerTest implements Serializable {
Review comment:
Was messing with some stuff attempting to get the test working. Forgot to re-add. Though as an aside I'm not sure why this is necessary since everything seems fine with out.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r727598137
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1897,14 +1898,26 @@ private Instant minTargetAndGcTime(Instant target) {
return Timer.cleared(userKey, dynamicTimerTag, Collections.singletonList(boundedWindow));
}
+ @SuppressWarnings("deprecation") // Allowed Skew is deprecated for users, but must be respected
private Timer<K> getTimerForTime(Instant scheduledTime) {
if (outputTimestamp != null) {
- checkArgument(
- !outputTimestamp.isBefore(elementTimestampOrTimerHoldTimestamp),
- "output timestamp %s should be after input message timestamp or output timestamp of"
- + " firing timers %s",
- outputTimestamp,
- elementTimestampOrTimerHoldTimestamp);
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elementTimestampOrTimerHoldTimestamp.minus(doFn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (outputTimestamp.isBefore(lowerBound) || outputTimestamp.isAfter(upperBound)) {
Review comment:
```suggestion
try {
lowerBound = elementTimestampOrTimerHoldTimestamp.minus(doFn.getAllowedTimestampSkew());
} catch (ArithmeticException e) {
lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
}
if (outputTimestamp.isBefore(lowerBound) || outputTimestamp.isAfter(BoundedWindow.TIMESTAMP_MAX_VALUE)) {
```
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1999,6 +2012,28 @@ public void set(String dynamicTimerTag, Instant absoluteTime) {
}
}
+ @SuppressWarnings("deprecation") // Allowed Skew is deprecated for users, but must be respected
+ private void checkTimestamp(Instant timestamp) {
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = currentElement.getTimestamp().minus(doFn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (timestamp.isBefore(lowerBound) || timestamp.isAfter(upperBound)) {
Review comment:
```suggestion
try {
lowerBound = currentElement.getTimestamp().minus(doFn.getAllowedTimestampSkew());
} catch (ArithmeticException e) {
lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
}
if (timestamp.isBefore(lowerBound) || timestamp.isAfter(BoundedWindow.TIMESTAMP_MAX_VALUE)) {
```
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,13 +1195,24 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
- checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
- "output timestamp %s should be after input message timestamp or output timestamp of firing timers %s",
- outputTimestamp,
- elementInputTimestamp);
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elementInputTimestamp.minus(fn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (outputTimestamp.isBefore(lowerBound) || outputTimestamp.isAfter(upperBound)) {
Review comment:
```suggestion
try {
lowerBound = elementInputTimestamp.minus(fn.getAllowedTimestampSkew());
} catch (ArithmeticException e) {
lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
}
if (outputTimestamp.isBefore(lowerBound) || outputTimestamp.isAfter(BoundedWindow.TIMESTAMP_MAX_VALUE)) {
```
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -386,6 +397,186 @@ public void testInfiniteSkew() {
BoundedWindow.TIMESTAMP_MAX_VALUE));
}
+ @Test
+ @Category({ValidatesRunner.class, UsesTimersInParDo.class})
+ public void testRunnerNoSkew() {
+ PCollection<Duration> input =
+ p.apply(
+ "create",
+ Create.timestamped(
+ Arrays.asList(new Duration(0L), new Duration(1L)), Arrays.asList(0L, 1L)));
+ input.apply(ParDo.of(new SkewingDoFn(Duration.ZERO)));
+ // The errors differ between runners but at least check that the output timestamp is printed.
+ thrown.expectMessage(String.format("%s", new Instant(0L)));
+ thrown.expectMessage("output");
+ p.run();
+ }
+
+ @Test
+ @Category({UsesTimersInParDo.class, ValidatesRunner.class})
+ public void testRunnerAllowedSkew() {
+ PCollection<Duration> input =
+ p.apply(
+ "create",
+ Create.timestamped(
+ Arrays.asList(new Duration(0L), new Duration(1L)), Arrays.asList(0L, 1L)));
+ input.apply(ParDo.of(new SkewingDoFn(new Duration(2L))));
+ p.run();
+ }
+
+ /**
+ * Demonstrates that attempting to set a timer with an output timestamp before the timestamp of
+ * the current element with zero {@link DoFn#getAllowedTimestampSkew() allowed timestamp skew}
+ * throws.
+ */
+ @Test
+ public void testTimerBackwardsInTimeNoSkew() {
+ TimerSkewingDoFn fn = new TimerSkewingDoFn(Duration.ZERO);
+ DoFnRunner<KV<String, Duration>, Duration> runner =
+ new SimpleDoFnRunner<>(
+ null,
+ fn,
+ NullSideInputReader.empty(),
+ new ListOutputManager(),
+ new TupleTag<>(),
+ Collections.emptyList(),
+ mockStepContext,
+ null,
+ Collections.emptyMap(),
+ WindowingStrategy.of(new GlobalWindows()),
+ DoFnSchemaInformation.create(),
+ Collections.emptyMap());
+
+ runner.startBundle();
+ // A timer with output timestamp at the current timestamp is fine.
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(KV.of("1", Duration.ZERO), new Instant(0)));
+
+ thrown.expect(UserCodeException.class);
+ thrown.expectCause(isA(IllegalArgumentException.class));
+ thrown.expectMessage("output timestamp of firing timers");
+ thrown.expectMessage(
+ String.format("output timestamp %s", new Instant(0).minus(Duration.millis(1L))));
+ thrown.expectMessage(
+ String.format(
+ "allowed skew %s", PeriodFormat.getDefault().print(Duration.ZERO.toPeriod())));
+
+ // A timer with output timestamp before (current time - skew) is forbidden
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of("2", Duration.millis(1L)), new Instant(0)));
+ }
+
+ /**
+ * Demonstrates that attempting to have a timer with output timestamp before the timestamp of the
+ * current element plus the value of {@link DoFn#getAllowedTimestampSkew()} throws, but between
+ * that value and the current timestamp succeeds.
+ */
+ @Test
+ public void testTimerSkew() {
+ TimerSkewingDoFn fn = new TimerSkewingDoFn(Duration.standardMinutes(10L));
+ DoFnRunner<KV<String, Duration>, Duration> runner =
+ new SimpleDoFnRunner<>(
+ null,
+ fn,
+ NullSideInputReader.empty(),
+ new ListOutputManager(),
+ new TupleTag<>(),
+ Collections.emptyList(),
+ mockStepContext,
+ null,
+ Collections.emptyMap(),
+ WindowingStrategy.of(new GlobalWindows()),
+ DoFnSchemaInformation.create(),
+ Collections.emptyMap());
+
+ runner.startBundle();
+ // Timer with output timestamp between "now" and "now - allowed skew" succeeds.
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of("1", Duration.standardMinutes(5L)), new Instant(0)));
+ thrown.expect(UserCodeException.class);
+ thrown.expectCause(isA(IllegalArgumentException.class));
+ thrown.expectMessage("output timestamp of firing timers");
+ thrown.expectMessage(
+ String.format("output timestamp %s", new Instant(0).minus(Duration.standardHours(1L))));
+ thrown.expectMessage(
+ String.format(
+ "allowed skew %s",
+ PeriodFormat.getDefault().print(Duration.standardMinutes(10L).toPeriod())));
+ // Timer with output timestamp before "now - allowed skew" fails.
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of("2", Duration.standardHours(1L)), new Instant(0)));
+ }
+
+ /**
+ * Demonstrates that attempting to output an element with a timestamp before the current one
+ * always succeeds when {@link DoFn#getAllowedTimestampSkew()} is equal to {@link Long#MAX_VALUE}
+ * milliseconds.
+ */
+ @Test
+ public void testTimerInfiniteSkew() {
+ TimerSkewingDoFn fn = new TimerSkewingDoFn(Duration.millis(Long.MAX_VALUE));
+ DoFnRunner<KV<String, Duration>, Duration> runner =
+ new SimpleDoFnRunner<>(
+ null,
+ fn,
+ NullSideInputReader.empty(),
+ new ListOutputManager(),
+ new TupleTag<>(),
+ Collections.emptyList(),
+ mockStepContext,
+ null,
+ Collections.emptyMap(),
+ WindowingStrategy.of(new GlobalWindows()),
+ DoFnSchemaInformation.create(),
+ Collections.emptyMap());
+
+ runner.startBundle();
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of("1", Duration.millis(1L)), new Instant(0)));
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of("2", Duration.millis(1L)),
+ BoundedWindow.TIMESTAMP_MIN_VALUE.plus(Duration.millis(1))));
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of(
+ "3",
+ // This is the maximum amount a timestamp in beam can move (from the maximum
+ // timestamp
+ // to the minimum timestamp).
+ Duration.millis(BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis())
+ .minus(Duration.millis(BoundedWindow.TIMESTAMP_MIN_VALUE.getMillis()))),
+ BoundedWindow.TIMESTAMP_MAX_VALUE));
+ }
+
+ @Test
+ @Category({UsesTimersInParDo.class, ValidatesRunner.class})
+ public void testRunnerTimerNoSkew() {
+ List<KV<String, Duration>> durations =
+ Arrays.asList(KV.of("0", new Duration(0L)), KV.of("2", new Duration(1L)));
+ PCollection<KV<String, Duration>> input =
+ p.apply("create", Create.timestamped(durations, Arrays.asList(0L, 2L)));
+ input.apply(ParDo.of(new TimerSkewingDoFn(Duration.ZERO)));
+ // The errors differ between runners but at least check that the output timestamp is printed.
+ thrown.expectMessage(String.format("%s", new Instant(1L)));
Review comment:
I don't think this will work for Dataflow streaming.
This passes on Jenkins because all of these tests don't run in both streaming and batch modes but will not work when we run tests inside Google.
##########
File path: sdks/java/harness/src/test/java/org/apache/beam/fn/harness/FnApiDoFnRunnerTest.java
##########
@@ -3878,6 +3885,182 @@ public void testProcessElementForWindowedTruncateAndSizeRestriction() throws Exc
}
}
+ @RunWith(JUnit4.class)
+ public static class ExceptionThrowingExecutionTest {
+ @Rule public final ExpectedException thrown = ExpectedException.none();
+
+ public static final String TEST_TRANSFORM_ID = "pTransformId";
+
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<String, String> {
+ private final Duration allowedSkew;
+
+ private SkewingDoFn(Duration allowedSkew) {
+ this.allowedSkew = allowedSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ Duration duration = new Duration(Long.valueOf(context.element()));
+ context.outputWithTimestamp(context.element(), context.timestamp().minus(duration));
+ }
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ @Test
+ public void testDoFnSkewNotAllowed() throws Exception {
+ Pipeline p = Pipeline.create();
+ PCollection<String> valuePCollection = p.apply(Create.of("0", "1"));
+ PCollection<String> outputPCollection =
+ valuePCollection.apply(TEST_TRANSFORM_ID, ParDo.of(new SkewingDoFn(Duration.ZERO)));
+
+ SdkComponents sdkComponents = SdkComponents.create(p.getOptions());
+ RunnerApi.Pipeline pProto = PipelineTranslation.toProto(p, sdkComponents);
+ String inputPCollectionId = sdkComponents.registerPCollection(valuePCollection);
+ String outputPCollectionId = sdkComponents.registerPCollection(outputPCollection);
+ RunnerApi.PTransform pTransform =
+ pProto
+ .getComponents()
+ .getTransformsOrThrow(
+ pProto
+ .getComponents()
+ .getTransformsOrThrow(TEST_TRANSFORM_ID)
+ .getSubtransforms(0));
+
+ List<WindowedValue<String>> mainOutputValues = new ArrayList<>();
+ MetricsContainerStepMap metricsContainerRegistry = new MetricsContainerStepMap();
+ PCollectionConsumerRegistry consumers =
+ new PCollectionConsumerRegistry(
+ metricsContainerRegistry, mock(ExecutionStateTracker.class));
+
+ consumers.register(
+ outputPCollectionId,
+ TEST_TRANSFORM_ID,
+ (FnDataReceiver) (FnDataReceiver<WindowedValue<String>>) mainOutputValues::add,
+ StringUtf8Coder.of());
+ PTransformFunctionRegistry startFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "start");
+ PTransformFunctionRegistry finishFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "finish");
+ List<ThrowingRunnable> teardownFunctions = new ArrayList<>();
+
+ new FnApiDoFnRunner.Factory<>()
+ .createRunnerForPTransform(
+ PipelineOptionsFactory.create(),
+ null /* beamFnDataClient */,
+ null /* beamFnStateClient */,
+ null /* beamFnTimerClient */,
+ TEST_TRANSFORM_ID,
+ pTransform,
+ Suppliers.ofInstance("57L")::get,
+ pProto.getComponents().getPcollectionsMap(),
+ pProto.getComponents().getCodersMap(),
+ pProto.getComponents().getWindowingStrategiesMap(),
+ consumers,
+ startFunctionRegistry,
+ finishFunctionRegistry,
+ null /* addResetFunction */,
+ teardownFunctions::add,
+ null /* addProgressRequestCallback */,
+ null /* splitListener */,
+ null /* bundleFinalizer */);
+
+ thrown.expect(UserCodeException.class);
+ thrown.expectMessage(String.format("timestamp %s", new Instant(0).minus(new Duration(1L))));
+ thrown.expectMessage(
+ String.format(
+ "allowed skew (%s)", PeriodFormat.getDefault().print(Duration.ZERO.toPeriod())));
+
+ Iterables.getOnlyElement(startFunctionRegistry.getFunctions()).run();
+ mainOutputValues.clear();
+
+ FnDataReceiver<WindowedValue<?>> mainInput =
+ consumers.getMultiplexingConsumer(inputPCollectionId);
+ mainInput.accept(valueInGlobalWindow("0"));
+ mainInput.accept(timestampedValueInGlobalWindow("1", new Instant(0L)));
Review comment:
Please swap to use `assertThrows` since `thrown.expect` is deprecated because it doesn't do a good job at showing which line of code is causing the exception.
##########
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:
It would be nice to be able to test the onTimer variant as well and not just when timers are set within processElement.
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -386,6 +399,180 @@ public void testInfiniteSkew() {
BoundedWindow.TIMESTAMP_MAX_VALUE));
}
+ @Test
+ @Category({ValidatesRunner.class})
+ public void testRunnerNoSkew() {
+ PCollection<Duration> input =
+ p.apply("create", Create.timestamped(Arrays.asList(new Duration(0L), new Duration(1L)), Arrays.asList(0L, 1L)));
+ input.apply(ParDo.of(new SkewingDoFn(Duration.ZERO)));
+ // The errors differ between runners but at least check that the output timestamp is printed.
+ thrown.expectMessage(String.format("%s", new Instant(0L)));
Review comment:
The issue is that `thrown.expect` is going to rely on Pipeline.run throwing the underlying cause for the job failure. I don't think all runners properly propagate the error (e.g. Dataflow streaming) that caused the pipeline to fail. The other failure tests in `ParDoTest.java` use `NeedsRunner` which run with the DirectRunner and not all runners. So if you want to only test DirectRunner mark as `NeedsRunner` instead of `ValidatesRunner`
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -437,19 +437,24 @@ public Instant timestamp() {
@SuppressWarnings("deprecation") // Allowed Skew is deprecated for users, but must be respected
private void checkTimestamp(Instant timestamp) {
- // The documentation of getAllowedTimestampSkew explicitly permits Long.MAX_VALUE to be used
- // for infinite skew. Defend against underflow in that case for timestamps before the epoch
- if (fn.getAllowedTimestampSkew().getMillis() != Long.MAX_VALUE
- && timestamp.isBefore(elem.getTimestamp().minus(fn.getAllowedTimestampSkew()))) {
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elem.getTimestamp().minus(fn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (timestamp.isBefore(lowerBound) || timestamp.isAfter(upperBound)) {
Review comment:
```suggestion
try {
lowerBound = elem.getTimestamp().minus(fn.getAllowedTimestampSkew());
} catch (ArithmeticException e) {
lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
}
if (timestamp.isBefore(lowerBound) || timestamp.isAfter(BoundedWindow.TIMESTAMP_MAX_VALUE)) {
```
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -437,19 +437,24 @@ public Instant timestamp() {
@SuppressWarnings("deprecation") // Allowed Skew is deprecated for users, but must be respected
private void checkTimestamp(Instant timestamp) {
- // The documentation of getAllowedTimestampSkew explicitly permits Long.MAX_VALUE to be used
- // for infinite skew. Defend against underflow in that case for timestamps before the epoch
- if (fn.getAllowedTimestampSkew().getMillis() != Long.MAX_VALUE
- && timestamp.isBefore(elem.getTimestamp().minus(fn.getAllowedTimestampSkew()))) {
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elem.getTimestamp().minus(fn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (timestamp.isBefore(lowerBound) || timestamp.isAfter(upperBound)) {
throw new IllegalArgumentException(
String.format(
"Cannot output with timestamp %s. Output timestamps must be no earlier than the "
- + "timestamp of the current input (%s) minus the allowed skew (%s). See the "
- + "DoFn#getAllowedTimestampSkew() Javadoc for details on changing the allowed "
- + "skew.",
+ + "timestamp of the current input (%s) minus the allowed skew (%s) and no "
+ + "later than %s. See the DoFn#getAllowedTimestampSkew() Javadoc for details "
+ + "on changing the allowed skew.",
timestamp,
elem.getTimestamp(),
- PeriodFormat.getDefault().print(fn.getAllowedTimestampSkew().toPeriod())));
+ PeriodFormat.getDefault().print(fn.getAllowedTimestampSkew().toPeriod()),
+ upperBound));
Review comment:
```suggestion
BoundedWindow.TIMESTAMP_MAX_VALUE));
```
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,13 +1195,24 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
- checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
- "output timestamp %s should be after input message timestamp or output timestamp of firing timers %s",
- outputTimestamp,
- elementInputTimestamp);
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elementInputTimestamp.minus(fn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (outputTimestamp.isBefore(lowerBound) || outputTimestamp.isAfter(upperBound)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "output timestamp %s (allowed skew %s) should be after input message timestamp or"
+ + " output timestamp of firing timers %s and before %s",
+ outputTimestamp,
+ PeriodFormat.getDefault().print(fn.getAllowedTimestampSkew().toPeriod()),
+ elementInputTimestamp,
+ upperBound));
Review comment:
The previous message isn't great. Can we improve this to be the similar to the other message like:
```
Cannot output timer with timestamp %s. Output timestamps must be no earlier than the timestamp of the current input (%s) minus the allowed skew (%s) and no later than %s. See the DoFn#getAllowedTimestampSkew() Javadoc for details on changing the allowed skew.
```
Ditto here and in `FnApiDoFnRunner.java`
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -386,6 +397,186 @@ public void testInfiniteSkew() {
BoundedWindow.TIMESTAMP_MAX_VALUE));
}
+ @Test
+ @Category({ValidatesRunner.class, UsesTimersInParDo.class})
+ public void testRunnerNoSkew() {
Review comment:
Please move the `ValidatesRunner` tests to `ParDoTest.java` since that is where the bulk of the other ParDo specific `ValidatesRunner` and `NeedsRunner` test exist.
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1999,6 +2012,28 @@ public void set(String dynamicTimerTag, Instant absoluteTime) {
}
}
+ @SuppressWarnings("deprecation") // Allowed Skew is deprecated for users, but must be respected
+ private void checkTimestamp(Instant timestamp) {
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = currentElement.getTimestamp().minus(doFn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (timestamp.isBefore(lowerBound) || timestamp.isAfter(upperBound)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Cannot output with timestamp %s. Output timestamps must be no earlier than the "
+ + "timestamp of the current input (%s) minus the allowed skew (%s). See the "
Review comment:
This message differs from the one in SimpleDoFnRunner as it is missing the upper bound.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-936865620
> > > Won't this allow for infinite skew since if have a timer at `X` and skew of `-1` then the first time the timer is processed you can output at time `X-1` and when it gets scheduled again you can now output at `X-2` since the the new timers timestamp is `X-1`?
> >
> >
> > So my understanding of the reason for these checks is to stop people from doing the wrong thing without realizing it. We don't even take any different action based on this variable. It seems okay to apply this to each specific output timestamp and let you skew more if you chain timers in this fashion.
> > On a more practical note, there's reasons why you might want a timer to output an earlier element if you've properly set up watermark holds. There's currently no way to do that so we need some allowance. It would probably be better if we could constrain skew from the first output timestamp but I don't think that's available in the later timers, right?
> > If you disagree with the approach, I can bring this up on the email thread for others to chime in in case they are not checking here.
>
> I think users will be surprised that their data will be dropped as late once they pass the watermark skew bound if they output past it. The existing logic had guards for this explicitly since it would be surprising for users so I do believe it is important enough to discuss whether there is another approach to solve this or we are ok with this happening.
We chatted a bit about this offline. There's actually no guarantee that the watermark is held back when using DoFn#getAllowedTimestampSkew. The allowedTimestampSkew just removes the check that we have to avoid accidentally dropping late data. See the javadoc [1] and relevant reply from Jan [2].
[1] https://beam.apache.org/releases/javadoc/2.5.0/org/apache/beam/sdk/transforms/DoFn.html#getAllowedTimestampSkew--
[2] https://lists.apache.org/thread.html/r34c70d8a5f213f7bd2f4557019e27b7f07f5120d0a8794512c88568c%40%3Cdev.beam.apache.org%3E
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r714359098
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -386,6 +388,89 @@ public void testInfiniteSkew() {
BoundedWindow.TIMESTAMP_MAX_VALUE));
}
+ /**
+ * Demonstrates that attempting to set a timer with an output timestamp before the timestamp of
+ * the current element with zero {@link DoFn#getAllowedTimestampSkew() allowed timestamp skew}
+ * throws.
+ */
+ @Test
+ public void testTimerBackwardsInTimeNoSkew() {
Review comment:
I added ValidatesRunner tests for both the timer case and the normal DoFn case as additional tests. Unit tests still make sense for this specific DoFn and the validates runner ones cannot check for the exact error without being overly restrictive. This seems like the wrong place for tests that validate a runner but it seems like that's standard practice in beam so I just put them here.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] edited a comment on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
codecov[bot] edited a comment on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-953406612
# [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#15540](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7b0de72) into [master](https://codecov.io/gh/apache/beam/commit/8da177f64d314cf72e89a51e51fb0915f706a784?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (8da177f) will **increase** coverage by `0.00%`.
> The diff coverage is `n/a`.
> :exclamation: Current head 7b0de72 differs from pull request most recent head 5a95bf6. Consider uploading reports for the commit 5a95bf6 to get more accurate results
[](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #15540 +/- ##
========================================
Coverage 83.53% 83.53%
========================================
Files 445 445
Lines 61385 61210 -175
========================================
- Hits 51276 51131 -145
+ Misses 10109 10079 -30
```
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...hon/apache\_beam/runners/direct/test\_stream\_impl.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9kaXJlY3QvdGVzdF9zdHJlYW1faW1wbC5weQ==) | `94.02% <0.00%> (-2.24%)` | :arrow_down: |
| [...python/apache\_beam/runners/worker/worker\_status.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvd29ya2VyX3N0YXR1cy5weQ==) | `78.26% <0.00%> (-1.45%)` | :arrow_down: |
| [sdks/python/apache\_beam/internal/metrics/metric.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW50ZXJuYWwvbWV0cmljcy9tZXRyaWMucHk=) | `90.00% <0.00%> (-1.00%)` | :arrow_down: |
| [...apache\_beam/runners/dataflow/internal/apiclient.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9kYXRhZmxvdy9pbnRlcm5hbC9hcGljbGllbnQucHk=) | `76.46% <0.00%> (-0.73%)` | :arrow_down: |
| [sdks/python/apache\_beam/io/textio.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vdGV4dGlvLnB5) | `97.10% <0.00%> (-0.26%)` | :arrow_down: |
| [...ks/python/apache\_beam/runners/worker/sdk\_worker.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvc2RrX3dvcmtlci5weQ==) | `89.25% <0.00%> (-0.17%)` | :arrow_down: |
| [...dks/python/apache\_beam/metrics/monitoring\_infos.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWV0cmljcy9tb25pdG9yaW5nX2luZm9zLnB5) | `96.93% <0.00%> (-0.07%)` | :arrow_down: |
| [sdks/python/apache\_beam/transforms/util.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdHJhbnNmb3Jtcy91dGlsLnB5) | `95.81% <0.00%> (-0.05%)` | :arrow_down: |
| [sdks/python/apache\_beam/dataframe/frames.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vZGF0YWZyYW1lL2ZyYW1lcy5weQ==) | `94.87% <0.00%> (-0.05%)` | :arrow_down: |
| [sdks/python/apache\_beam/coders/row\_coder.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vY29kZXJzL3Jvd19jb2Rlci5weQ==) | `95.07% <0.00%> (ø)` | |
| ... and [7 more](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [0cff1ff...5a95bf6](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] edited a comment on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
codecov[bot] edited a comment on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-953406612
# [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#15540](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7b0de72) into [master](https://codecov.io/gh/apache/beam/commit/03d4e42520070b304eed05da6f076c758653ca7f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (03d4e42) will **decrease** coverage by `0.00%`.
> The diff coverage is `76.92%`.
> :exclamation: Current head 7b0de72 differs from pull request most recent head a720c16. Consider uploading reports for the commit a720c16 to get more accurate results
[](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #15540 +/- ##
==========================================
- Coverage 83.53% 83.53% -0.01%
==========================================
Files 445 445
Lines 61385 61210 -175
==========================================
- Hits 51279 51131 -148
+ Misses 10106 10079 -27
```
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [sdks/python/apache\_beam/transforms/combiners.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdHJhbnNmb3Jtcy9jb21iaW5lcnMucHk=) | `93.54% <50.00%> (ø)` | |
| [sdks/python/apache\_beam/coders/coder\_impl.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vY29kZXJzL2NvZGVyX2ltcGwucHk=) | `94.04% <88.88%> (ø)` | |
| [...hon/apache\_beam/runners/direct/test\_stream\_impl.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9kaXJlY3QvdGVzdF9zdHJlYW1faW1wbC5weQ==) | `94.02% <0.00%> (-2.24%)` | :arrow_down: |
| [...python/apache\_beam/runners/worker/worker\_status.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvd29ya2VyX3N0YXR1cy5weQ==) | `78.26% <0.00%> (-1.45%)` | :arrow_down: |
| [sdks/python/apache\_beam/internal/metrics/metric.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW50ZXJuYWwvbWV0cmljcy9tZXRyaWMucHk=) | `90.00% <0.00%> (-1.00%)` | :arrow_down: |
| [...apache\_beam/runners/dataflow/internal/apiclient.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9kYXRhZmxvdy9pbnRlcm5hbC9hcGljbGllbnQucHk=) | `76.46% <0.00%> (-0.73%)` | :arrow_down: |
| [sdks/python/apache\_beam/io/textio.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vdGV4dGlvLnB5) | `97.10% <0.00%> (-0.26%)` | :arrow_down: |
| [...ks/python/apache\_beam/runners/worker/sdk\_worker.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvc2RrX3dvcmtlci5weQ==) | `89.25% <0.00%> (-0.17%)` | :arrow_down: |
| [...dks/python/apache\_beam/metrics/monitoring\_infos.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWV0cmljcy9tb25pdG9yaW5nX2luZm9zLnB5) | `96.93% <0.00%> (-0.07%)` | :arrow_down: |
| [sdks/python/apache\_beam/transforms/util.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdHJhbnNmb3Jtcy91dGlsLnB5) | `95.81% <0.00%> (-0.05%)` | :arrow_down: |
| ... and [9 more](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [0cff1ff...a720c16](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] edited a comment on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
codecov[bot] edited a comment on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-953406612
# [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#15540](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7b0de72) into [master](https://codecov.io/gh/apache/beam/commit/bee56a65ead45c765d4695cfc7f019eb0d073f64?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (bee56a6) will **increase** coverage by `37.44%`.
> The diff coverage is `n/a`.
> :exclamation: Current head 7b0de72 differs from pull request most recent head 34de17f. Consider uploading reports for the commit 34de17f to get more accurate results
[](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #15540 +/- ##
===========================================
+ Coverage 46.08% 83.53% +37.44%
===========================================
Files 195 445 +250
Lines 19457 61210 +41753
===========================================
+ Hits 8967 51131 +42164
- Misses 9521 10079 +558
+ Partials 969 0 -969
```
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [sdks/go/pkg/beam/core/graph/coder/time.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL2dyYXBoL2NvZGVyL3RpbWUuZ28=) | | |
| [sdks/go/pkg/beam/core/runtime/xlangx/payload.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUveGxhbmd4L3BheWxvYWQuZ28=) | | |
| [sdks/go/pkg/beam/transforms/top/top.shims.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS90cmFuc2Zvcm1zL3RvcC90b3Auc2hpbXMuZ28=) | | |
| [sdks/go/pkg/beam/core/runtime/options.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvb3B0aW9ucy5nbw==) | | |
| [sdks/go/pkg/beam/core/funcx/fn.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL2Z1bmN4L2ZuLmdv) | | |
| [sdks/go/pkg/beam/util/gcsx/gcs.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS91dGlsL2djc3gvZ2NzLmdv) | | |
| [sdks/go/pkg/beam/util/starcgenx/starcgenx.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS91dGlsL3N0YXJjZ2VueC9zdGFyY2dlbnguZ28=) | | |
| [sdks/go/pkg/beam/transforms/filter/distinct.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS90cmFuc2Zvcm1zL2ZpbHRlci9kaXN0aW5jdC5nbw==) | | |
| [sdks/go/pkg/beam/io/textio/textio.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9pby90ZXh0aW8vdGV4dGlvLmdv) | | |
| [sdks/go/pkg/beam/core/graph/coder/varint.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL2dyYXBoL2NvZGVyL3ZhcmludC5nbw==) | | |
| ... and [630 more](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [bee56a6...34de17f](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-953203098
PTAL @lukecwik
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] commented on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
codecov[bot] commented on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-953406612
# [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#15540](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7b0de72) into [master](https://codecov.io/gh/apache/beam/commit/92ea5f547e2ac3dfd46e20b7c965f8a4850f2238?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (92ea5f5) will **decrease** coverage by `0.29%`.
> The diff coverage is `n/a`.
> :exclamation: Current head 7b0de72 differs from pull request most recent head 7cdde98. Consider uploading reports for the commit 7cdde98 to get more accurate results
[](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #15540 +/- ##
==========================================
- Coverage 83.82% 83.53% -0.30%
==========================================
Files 441 445 +4
Lines 59727 61210 +1483
==========================================
+ Hits 50068 51131 +1063
- Misses 9659 10079 +420
```
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [sdks/python/apache\_beam/io/gcp/bigquery.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vZ2NwL2JpZ3F1ZXJ5LnB5) | `62.72% <0.00%> (-13.66%)` | :arrow_down: |
| [...n/apache\_beam/ml/gcp/recommendations\_ai\_test\_it.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWwvZ2NwL3JlY29tbWVuZGF0aW9uc19haV90ZXN0X2l0LnB5) | `56.81% <0.00%> (-12.95%)` | :arrow_down: |
| [...thon/apache\_beam/runners/worker/operation\_specs.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvb3BlcmF0aW9uX3NwZWNzLnB5) | `40.67% <0.00%> (-4.90%)` | :arrow_down: |
| [...ython/apache\_beam/io/gcp/bigquery\_read\_internal.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vZ2NwL2JpZ3F1ZXJ5X3JlYWRfaW50ZXJuYWwucHk=) | `53.92% <0.00%> (-4.55%)` | :arrow_down: |
| [sdks/python/apache\_beam/io/gcp/bigquery\_tools.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vZ2NwL2JpZ3F1ZXJ5X3Rvb2xzLnB5) | `82.91% <0.00%> (-4.52%)` | :arrow_down: |
| [...ks/python/apache\_beam/runners/worker/data\_plane.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvZGF0YV9wbGFuZS5weQ==) | `87.50% <0.00%> (-3.11%)` | :arrow_down: |
| [...ache\_beam/runners/interactive/recording\_manager.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9pbnRlcmFjdGl2ZS9yZWNvcmRpbmdfbWFuYWdlci5weQ==) | `96.55% <0.00%> (-2.43%)` | :arrow_down: |
| [sdks/python/apache\_beam/coders/coder\_impl.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vY29kZXJzL2NvZGVyX2ltcGwucHk=) | `94.04% <0.00%> (-1.25%)` | :arrow_down: |
| [.../apache\_beam/io/gcp/datastore/v1new/datastoreio.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vZ2NwL2RhdGFzdG9yZS92MW5ldy9kYXRhc3RvcmVpby5weQ==) | `86.45% <0.00%> (-0.99%)` | :arrow_down: |
| [...che\_beam/runners/interactive/augmented\_pipeline.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9pbnRlcmFjdGl2ZS9hdWdtZW50ZWRfcGlwZWxpbmUucHk=) | `69.38% <0.00%> (-0.62%)` | :arrow_down: |
| ... and [100 more](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [92ea5f5...7cdde98](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik merged pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
lukecwik merged pull request #15540:
URL: https://github.com/apache/beam/pull/15540
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] edited a comment on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
codecov[bot] edited a comment on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-953406612
# [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#15540](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7b0de72) into [master](https://codecov.io/gh/apache/beam/commit/163ac6a3c10c26898ad89ca8bedde8ef78ee7ee2?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (163ac6a) will **increase** coverage by `37.44%`.
> The diff coverage is `n/a`.
> :exclamation: Current head 7b0de72 differs from pull request most recent head 8954610. Consider uploading reports for the commit 8954610 to get more accurate results
[](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #15540 +/- ##
===========================================
+ Coverage 46.08% 83.53% +37.44%
===========================================
Files 195 445 +250
Lines 19457 61210 +41753
===========================================
+ Hits 8967 51131 +42164
- Misses 9521 10079 +558
+ Partials 969 0 -969
```
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [sdks/go/pkg/beam/core/runtime/exec/hash.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvZXhlYy9oYXNoLmdv) | | |
| [sdks/go/pkg/beam/core/runtime/exec/fn\_arity.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvZXhlYy9mbl9hcml0eS5nbw==) | | |
| [...o/pkg/beam/io/rtrackers/offsetrange/offsetrange.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9pby9ydHJhY2tlcnMvb2Zmc2V0cmFuZ2Uvb2Zmc2V0cmFuZ2UuZ28=) | | |
| [sdks/go/pkg/beam/core/runtime/xlangx/resolve.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUveGxhbmd4L3Jlc29sdmUuZ28=) | | |
| [sdks/go/pkg/beam/core/util/reflectx/call.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3V0aWwvcmVmbGVjdHgvY2FsbC5nbw==) | | |
| [sdks/go/pkg/beam/io/databaseio/mapper.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9pby9kYXRhYmFzZWlvL21hcHBlci5nbw==) | | |
| [sdks/go/pkg/beam/transforms/stats/quantiles.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS90cmFuc2Zvcm1zL3N0YXRzL3F1YW50aWxlcy5nbw==) | | |
| [sdks/go/pkg/beam/core/runtime/exec/fn.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvZXhlYy9mbi5nbw==) | | |
| [...pkg/beam/runners/dataflow/dataflowlib/translate.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9ydW5uZXJzL2RhdGFmbG93L2RhdGFmbG93bGliL3RyYW5zbGF0ZS5nbw==) | | |
| [sdks/go/pkg/beam/transforms/stats/sum\_switch.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS90cmFuc2Zvcm1zL3N0YXRzL3N1bV9zd2l0Y2guZ28=) | | |
| ... and [630 more](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [163ac6a...8954610](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] edited a comment on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
codecov[bot] edited a comment on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-953406612
# [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#15540](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7b0de72) into [master](https://codecov.io/gh/apache/beam/commit/bee56a65ead45c765d4695cfc7f019eb0d073f64?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (bee56a6) will **increase** coverage by `8.93%`.
> The diff coverage is `n/a`.
> :exclamation: Current head 7b0de72 differs from pull request most recent head b373e0a. Consider uploading reports for the commit b373e0a to get more accurate results
[](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #15540 +/- ##
==========================================
+ Coverage 74.59% 83.53% +8.93%
==========================================
Files 642 445 -197
Lines 80958 61210 -19748
==========================================
- Hits 60394 51131 -9263
+ Misses 19595 10079 -9516
+ Partials 969 0 -969
```
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [sdks/python/apache\_beam/io/avroio.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vYXZyb2lvLnB5) | `60.60% <0.00%> (-37.36%)` | :arrow_down: |
| [...python/apache\_beam/runners/worker/worker\_status.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvd29ya2VyX3N0YXR1cy5weQ==) | `78.26% <0.00%> (-1.45%)` | :arrow_down: |
| [sdks/python/apache\_beam/internal/metrics/metric.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW50ZXJuYWwvbWV0cmljcy9tZXRyaWMucHk=) | `90.00% <0.00%> (-1.00%)` | :arrow_down: |
| [...apache\_beam/runners/dataflow/internal/apiclient.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9kYXRhZmxvdy9pbnRlcm5hbC9hcGljbGllbnQucHk=) | `76.46% <0.00%> (-0.73%)` | :arrow_down: |
| [...thon/apache\_beam/runners/worker/sdk\_worker\_main.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvc2RrX3dvcmtlcl9tYWluLnB5) | `72.35% <0.00%> (-0.45%)` | :arrow_down: |
| [...s/python/apache\_beam/runners/portability/stager.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9wb3J0YWJpbGl0eS9zdGFnZXIucHk=) | `88.15% <0.00%> (-0.10%)` | :arrow_down: |
| [sdks/python/apache\_beam/typehints/typehints.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdHlwZWhpbnRzL3R5cGVoaW50cy5weQ==) | `93.34% <0.00%> (-0.10%)` | :arrow_down: |
| [sdks/python/apache\_beam/io/gcp/bigquery.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vZ2NwL2JpZ3F1ZXJ5LnB5) | `62.72% <0.00%> (-0.09%)` | :arrow_down: |
| [sdks/python/apache\_beam/pipeline.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcGlwZWxpbmUucHk=) | `91.73% <0.00%> (-0.08%)` | :arrow_down: |
| [sdks/python/apache\_beam/typehints/schemas.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vdHlwZWhpbnRzL3NjaGVtYXMucHk=) | `94.44% <0.00%> (-0.08%)` | :arrow_down: |
| ... and [244 more](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [bee56a6...b373e0a](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] edited a comment on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
codecov[bot] edited a comment on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-953406612
# [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#15540](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7b0de72) into [master](https://codecov.io/gh/apache/beam/commit/cd4b7f3b1af4f51bdab1a0b1a98f94b5288c09ec?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (cd4b7f3) will **increase** coverage by `0.02%`.
> The diff coverage is `n/a`.
> :exclamation: Current head 7b0de72 differs from pull request most recent head 9794db7. Consider uploading reports for the commit 9794db7 to get more accurate results
[](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #15540 +/- ##
==========================================
+ Coverage 83.50% 83.53% +0.02%
==========================================
Files 445 445
Lines 61299 61210 -89
==========================================
- Hits 51187 51131 -56
+ Misses 10112 10079 -33
```
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...hon/apache\_beam/runners/direct/test\_stream\_impl.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9kaXJlY3QvdGVzdF9zdHJlYW1faW1wbC5weQ==) | `94.02% <0.00%> (-2.24%)` | :arrow_down: |
| [...python/apache\_beam/runners/worker/worker\_status.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvd29ya2VyX3N0YXR1cy5weQ==) | `78.26% <0.00%> (-1.45%)` | :arrow_down: |
| [sdks/python/apache\_beam/internal/metrics/metric.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW50ZXJuYWwvbWV0cmljcy9tZXRyaWMucHk=) | `90.00% <0.00%> (-1.00%)` | :arrow_down: |
| [...ks/python/apache\_beam/runners/worker/sdk\_worker.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvc2RrX3dvcmtlci5weQ==) | `89.25% <0.00%> (-0.17%)` | :arrow_down: |
| [...dks/python/apache\_beam/metrics/monitoring\_infos.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWV0cmljcy9tb25pdG9yaW5nX2luZm9zLnB5) | `96.93% <0.00%> (-0.07%)` | :arrow_down: |
| [.../python/apache\_beam/io/gcp/resource\_identifiers.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vZ2NwL3Jlc291cmNlX2lkZW50aWZpZXJzLnB5) | `100.00% <0.00%> (ø)` | |
| [.../python/apache\_beam/portability/api/metrics\_pb2.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcG9ydGFiaWxpdHkvYXBpL21ldHJpY3NfcGIyLnB5) | `100.00% <0.00%> (ø)` | |
| [...on/apache\_beam/portability/api/metrics\_pb2\_urns.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcG9ydGFiaWxpdHkvYXBpL21ldHJpY3NfcGIyX3VybnMucHk=) | `100.00% <0.00%> (ø)` | |
| [...hon/apache\_beam/runners/worker/bundle\_processor.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvYnVuZGxlX3Byb2Nlc3Nvci5weQ==) | `93.64% <0.00%> (+0.12%)` | :arrow_up: |
| [...runners/interactive/display/pcoll\_visualization.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9pbnRlcmFjdGl2ZS9kaXNwbGF5L3Bjb2xsX3Zpc3VhbGl6YXRpb24ucHk=) | `86.36% <0.00%> (+0.50%)` | :arrow_up: |
| ... and [3 more](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [cd4b7f3...9794db7](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] reuvenlax commented on pull request #15540: Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
reuvenlax commented on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-924157443
I think you have to update the similar check in FnApiDoFnRunner
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] edited a comment on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
codecov[bot] edited a comment on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-953406612
# [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#15540](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7b0de72) into [master](https://codecov.io/gh/apache/beam/commit/cd4b7f3b1af4f51bdab1a0b1a98f94b5288c09ec?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (cd4b7f3) will **increase** coverage by `0.02%`.
> The diff coverage is `n/a`.
> :exclamation: Current head 7b0de72 differs from pull request most recent head fbcfd0b. Consider uploading reports for the commit fbcfd0b to get more accurate results
[](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #15540 +/- ##
==========================================
+ Coverage 83.50% 83.53% +0.02%
==========================================
Files 445 445
Lines 61299 61210 -89
==========================================
- Hits 51187 51131 -56
+ Misses 10112 10079 -33
```
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...hon/apache\_beam/runners/direct/test\_stream\_impl.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9kaXJlY3QvdGVzdF9zdHJlYW1faW1wbC5weQ==) | `94.02% <0.00%> (-2.24%)` | :arrow_down: |
| [...python/apache\_beam/runners/worker/worker\_status.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvd29ya2VyX3N0YXR1cy5weQ==) | `78.26% <0.00%> (-1.45%)` | :arrow_down: |
| [sdks/python/apache\_beam/internal/metrics/metric.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW50ZXJuYWwvbWV0cmljcy9tZXRyaWMucHk=) | `90.00% <0.00%> (-1.00%)` | :arrow_down: |
| [...ks/python/apache\_beam/runners/worker/sdk\_worker.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvc2RrX3dvcmtlci5weQ==) | `89.25% <0.00%> (-0.17%)` | :arrow_down: |
| [...dks/python/apache\_beam/metrics/monitoring\_infos.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWV0cmljcy9tb25pdG9yaW5nX2luZm9zLnB5) | `96.93% <0.00%> (-0.07%)` | :arrow_down: |
| [.../python/apache\_beam/io/gcp/resource\_identifiers.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vZ2NwL3Jlc291cmNlX2lkZW50aWZpZXJzLnB5) | `100.00% <0.00%> (ø)` | |
| [.../python/apache\_beam/portability/api/metrics\_pb2.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcG9ydGFiaWxpdHkvYXBpL21ldHJpY3NfcGIyLnB5) | `100.00% <0.00%> (ø)` | |
| [...on/apache\_beam/portability/api/metrics\_pb2\_urns.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcG9ydGFiaWxpdHkvYXBpL21ldHJpY3NfcGIyX3VybnMucHk=) | `100.00% <0.00%> (ø)` | |
| [...hon/apache\_beam/runners/worker/bundle\_processor.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvYnVuZGxlX3Byb2Nlc3Nvci5weQ==) | `93.64% <0.00%> (+0.12%)` | :arrow_up: |
| [...runners/interactive/display/pcoll\_visualization.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9pbnRlcmFjdGl2ZS9kaXNwbGF5L3Bjb2xsX3Zpc3VhbGl6YXRpb24ucHk=) | `86.36% <0.00%> (+0.50%)` | :arrow_up: |
| ... and [3 more](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [cd4b7f3...fbcfd0b](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r728302215
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -437,19 +437,24 @@ public Instant timestamp() {
@SuppressWarnings("deprecation") // Allowed Skew is deprecated for users, but must be respected
private void checkTimestamp(Instant timestamp) {
- // The documentation of getAllowedTimestampSkew explicitly permits Long.MAX_VALUE to be used
- // for infinite skew. Defend against underflow in that case for timestamps before the epoch
- if (fn.getAllowedTimestampSkew().getMillis() != Long.MAX_VALUE
- && timestamp.isBefore(elem.getTimestamp().minus(fn.getAllowedTimestampSkew()))) {
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elem.getTimestamp().minus(fn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (timestamp.isBefore(lowerBound) || timestamp.isAfter(upperBound)) {
Review comment:
done
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,13 +1195,24 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
- checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
- "output timestamp %s should be after input message timestamp or output timestamp of firing timers %s",
- outputTimestamp,
- elementInputTimestamp);
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elementInputTimestamp.minus(fn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (outputTimestamp.isBefore(lowerBound) || outputTimestamp.isAfter(upperBound)) {
Review comment:
done.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r719825698
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -386,6 +399,180 @@ public void testInfiniteSkew() {
BoundedWindow.TIMESTAMP_MAX_VALUE));
}
+ @Test
+ @Category({ValidatesRunner.class})
+ public void testRunnerNoSkew() {
+ PCollection<Duration> input =
+ p.apply("create", Create.timestamped(Arrays.asList(new Duration(0L), new Duration(1L)), Arrays.asList(0L, 1L)));
+ input.apply(ParDo.of(new SkewingDoFn(Duration.ZERO)));
+ // The errors differ between runners but at least check that the output timestamp is printed.
+ thrown.expectMessage(String.format("%s", new Instant(0L)));
Review comment:
That's a neat idea. though it still can't make any extra validations on the error message thrown from the system though and would be limited to the same check. I guess we could check more specifically where it failed, but not sure of the overall value there. Though if left as is it keeps the error checking in the test code instead of the DoFn which I like. So I'll probably leave it as is unless you have a strong opinion.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-931774284
PTAL, @lukecwik @reuvenlax
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r746171750
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1791,6 +1791,119 @@ public void finishBundle(FinishBundleContext c) {
/** Tests to validate output timestamps. */
@RunWith(JUnit4.class)
public static class TimestampTests extends SharedTestBase implements Serializable {
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<Duration, String> {
+ private final Duration allowedSkew;
+ static final String CORRECT_ELEMENT = "element";
+
+ private SkewingDoFn(Duration allowedSkew) {
+ this.allowedSkew = allowedSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ Instant outputTimestamp = context.timestamp().minus(context.element());
+ try {
+ context.outputWithTimestamp(CORRECT_ELEMENT, outputTimestamp);
+ } catch (IllegalArgumentException e) {
+ // Errors differ between runners but at least check that the output timestamp is printed.
Review comment:
Yeah though I still feel a bit odd checking specific error messages in multiple runners. I updated it to check the relevant pieces but not the specific wording.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] edited a comment on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
codecov[bot] edited a comment on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-953406612
# [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#15540](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7b0de72) into [master](https://codecov.io/gh/apache/beam/commit/bc82f6f84e72c520be397977edf4bf645782ac8f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (bc82f6f) will **increase** coverage by `37.42%`.
> The diff coverage is `n/a`.
> :exclamation: Current head 7b0de72 differs from pull request most recent head b3192dc. Consider uploading reports for the commit b3192dc to get more accurate results
[](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #15540 +/- ##
===========================================
+ Coverage 46.10% 83.53% +37.42%
===========================================
Files 196 445 +249
Lines 19498 61210 +41712
===========================================
+ Hits 8990 51131 +42141
- Misses 9538 10079 +541
+ Partials 970 0 -970
```
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [sdks/go/pkg/beam/option.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9vcHRpb24uZ28=) | | |
| [sdks/go/pkg/beam/core/runtime/exec/encode.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvZXhlYy9lbmNvZGUuZ28=) | | |
| [sdks/go/pkg/beam/impulse.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9pbXB1bHNlLmdv) | | |
| [sdks/go/pkg/beam/create.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jcmVhdGUuZ28=) | | |
| [sdks/go/pkg/beam/metrics.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9tZXRyaWNzLmdv) | | |
| [sdks/go/pkg/beam/core/metrics/metrics.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL21ldHJpY3MvbWV0cmljcy5nbw==) | | |
| [sdks/go/pkg/beam/io/filesystem/memfs/memory.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9pby9maWxlc3lzdGVtL21lbWZzL21lbW9yeS5nbw==) | | |
| [sdks/go/pkg/beam/core/runtime/exec/multiplex.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvZXhlYy9tdWx0aXBsZXguZ28=) | | |
| [sdks/go/pkg/beam/core/runtime/harness/statemgr.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvaGFybmVzcy9zdGF0ZW1nci5nbw==) | | |
| [sdks/go/pkg/beam/core/runtime/graphx/coder.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvZ3JhcGh4L2NvZGVyLmdv) | | |
| ... and [631 more](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [bc82f6f...b3192dc](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] edited a comment on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
codecov[bot] edited a comment on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-953406612
# [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#15540](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7b0de72) into [master](https://codecov.io/gh/apache/beam/commit/163ac6a3c10c26898ad89ca8bedde8ef78ee7ee2?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (163ac6a) will **increase** coverage by `37.44%`.
> The diff coverage is `n/a`.
> :exclamation: Current head 7b0de72 differs from pull request most recent head 3249a0c. Consider uploading reports for the commit 3249a0c to get more accurate results
[](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #15540 +/- ##
===========================================
+ Coverage 46.08% 83.53% +37.44%
===========================================
Files 195 445 +250
Lines 19457 61210 +41753
===========================================
+ Hits 8967 51131 +42164
- Misses 9521 10079 +558
+ Partials 969 0 -969
```
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [sdks/go/pkg/beam/transforms/stats/sum.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS90cmFuc2Zvcm1zL3N0YXRzL3N1bS5nbw==) | | |
| [sdks/go/pkg/beam/core/util/reflectx/call.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3V0aWwvcmVmbGVjdHgvY2FsbC5nbw==) | | |
| [sdks/go/pkg/beam/validate.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS92YWxpZGF0ZS5nbw==) | | |
| [sdks/go/pkg/beam/core/graph/coder/iterable.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL2dyYXBoL2NvZGVyL2l0ZXJhYmxlLmdv) | | |
| [sdks/go/pkg/beam/core/graph/coder/windows.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL2dyYXBoL2NvZGVyL3dpbmRvd3MuZ28=) | | |
| [sdks/go/pkg/beam/core/runtime/xlangx/payload.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUveGxhbmd4L3BheWxvYWQuZ28=) | | |
| [sdks/go/pkg/beam/core/util/reflectx/tags.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3V0aWwvcmVmbGVjdHgvdGFncy5nbw==) | | |
| [sdks/go/pkg/beam/core/runtime/exec/encode.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvZXhlYy9lbmNvZGUuZ28=) | | |
| [sdks/go/pkg/beam/encoding.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9lbmNvZGluZy5nbw==) | | |
| [sdks/go/pkg/beam/core/graph/coder/stringutf8.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL2dyYXBoL2NvZGVyL3N0cmluZ3V0ZjguZ28=) | | |
| ... and [630 more](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [163ac6a...3249a0c](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r719664915
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -386,6 +399,180 @@ public void testInfiniteSkew() {
BoundedWindow.TIMESTAMP_MAX_VALUE));
}
+ @Test
+ @Category({ValidatesRunner.class})
+ public void testRunnerNoSkew() {
+ PCollection<Duration> input =
+ p.apply("create", Create.timestamped(Arrays.asList(new Duration(0L), new Duration(1L)), Arrays.asList(0L, 1L)));
+ input.apply(ParDo.of(new SkewingDoFn(Duration.ZERO)));
+ // The errors differ between runners but at least check that the output timestamp is printed.
+ thrown.expectMessage(String.format("%s", new Instant(0L)));
Review comment:
Note that you can solve this problem by having the DoFn catch the exception from within the processElement method outputting `success` or `exception stack trace/message` messages and performing a PAssert on the output PCollection.
##########
File path: sdks/java/harness/src/test/java/org/apache/beam/fn/harness/FnApiDoFnRunnerTest.java
##########
@@ -1604,6 +1606,11 @@ public ProcessContinuation processElement(
}
}
+ @Override
+ public Duration getAllowedTimestampSkew() {
Review comment:
Why this change?
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,13 +1190,20 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
- checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
- "output timestamp %s should be after input message timestamp or output timestamp of firing timers %s",
- outputTimestamp,
- elementInputTimestamp);
+ // The documentation of getAllowedTimestampSkew explicitly permits Long.MAX_VALUE to be used
+ // for infinite skew. Defend against underflow in that case for timestamps before the epoch.
+ if (fn.getAllowedTimestampSkew().getMillis() != Long.MAX_VALUE
Review comment:
You can handle the proper bounds via:
```
Instant lowerBound;
try {
lowerBound = elementInputTimestamp.minus(fn.getAllowedTimestampSkew());
catch (ArithmeticException e) {
lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
}
if (outputTimestamp.isBefore(lowerBound)) {
...
}
```
Finally it would make sense to check the upper bound as well of `BoundedWindow.TIMESTAMP_MAX_VALUE`
##########
File path: sdks/java/harness/src/test/java/org/apache/beam/fn/harness/FnApiDoFnRunnerTest.java
##########
@@ -164,7 +166,7 @@
@SuppressWarnings({
"rawtypes", // TODO(https://issues.apache.org/jira/browse/BEAM-10556)
})
-public class FnApiDoFnRunnerTest implements Serializable {
Review comment:
Why drop serializable?
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -386,6 +399,180 @@ public void testInfiniteSkew() {
BoundedWindow.TIMESTAMP_MAX_VALUE));
}
+ @Test
+ @Category({ValidatesRunner.class})
Review comment:
Need to tag with `UsesTimersInParDo.class` and/or `UsesTimerMap.class`.
##########
File path: sdks/java/harness/src/test/java/org/apache/beam/fn/harness/FnApiDoFnRunnerTest.java
##########
@@ -3878,6 +3885,186 @@ public void testProcessElementForWindowedTruncateAndSizeRestriction() throws Exc
}
}
+ @RunWith(JUnit4.class)
+ public static class ExceptionThrowingExecutionTest {
+ @Rule public final ExpectedException thrown = ExpectedException.none();
+
+ public static final String TEST_TRANSFORM_ID = "pTransformId";
+
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<String, String> {
+ private final Duration allowedSkew;
+
+ private SkewingDoFn(Duration allowedSkew) {
+ this.allowedSkew = allowedSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ Duration duration = new Duration(Long.valueOf(context.element()));
+ context.outputWithTimestamp(context.element(), context.timestamp().minus(duration));
+ }
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ @Test
+ public void testDoFnSkewNotAllowed() throws Exception {
+ Pipeline p = Pipeline.create();
+ PCollection<String> valuePCollection = p.apply(Create.of("0", "1"));
+ PCollection<String> outputPCollection =
+ valuePCollection.apply(TEST_TRANSFORM_ID, ParDo.of(new SkewingDoFn(Duration.ZERO)));
+
+ SdkComponents sdkComponents = SdkComponents.create(p.getOptions());
+ RunnerApi.Pipeline pProto = PipelineTranslation.toProto(p, sdkComponents);
+ String inputPCollectionId = sdkComponents.registerPCollection(valuePCollection);
+ String outputPCollectionId = sdkComponents.registerPCollection(outputPCollection);
+ RunnerApi.PTransform pTransform =
+ pProto
+ .getComponents()
+ .getTransformsOrThrow(
+ pProto
+ .getComponents()
+ .getTransformsOrThrow(TEST_TRANSFORM_ID)
+ .getSubtransforms(0));
+
+ List<WindowedValue<String>> mainOutputValues = new ArrayList<>();
+ MetricsContainerStepMap metricsContainerRegistry = new MetricsContainerStepMap();
+ PCollectionConsumerRegistry consumers =
+ new PCollectionConsumerRegistry(
+ metricsContainerRegistry, mock(ExecutionStateTracker.class));
+
+ consumers.register(
+ outputPCollectionId,
+ TEST_TRANSFORM_ID,
+ (FnDataReceiver) (FnDataReceiver<WindowedValue<String>>) mainOutputValues::add,
+ StringUtf8Coder.of());
+ PTransformFunctionRegistry startFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "start");
+ PTransformFunctionRegistry finishFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "finish");
+ List<ThrowingRunnable> teardownFunctions = new ArrayList<>();
+
+ new FnApiDoFnRunner.Factory<>()
+ .createRunnerForPTransform(
+ PipelineOptionsFactory.create(),
+ null /* beamFnDataClient */,
+ null /**/,
+ null /* beamFnTimerClient */,
+ TEST_TRANSFORM_ID,
+ pTransform,
+ Suppliers.ofInstance("57L")::get,
+ pProto.getComponents().getPcollectionsMap(),
+ pProto.getComponents().getCodersMap(),
+ pProto.getComponents().getWindowingStrategiesMap(),
+ consumers,
+ startFunctionRegistry,
+ finishFunctionRegistry,
+ null /* addResetFunction */,
+ teardownFunctions::add,
+ null /* addProgressRequestCallback */,
+ null /* splitListener */,
+ null /* bundleFinalizer */);
+
+ thrown.expect(UserCodeException.class);
+ thrown.expectMessage(
+ String.format("timestamp %s", new Instant(0).minus(new Duration(1L))));
+ thrown.expectMessage(
+ String.format(
+ "allowed skew (%s)",
+ PeriodFormat.getDefault().print(Duration.ZERO.toPeriod())));
+
+ Iterables.getOnlyElement(startFunctionRegistry.getFunctions()).run();
+ mainOutputValues.clear();
+
+ FnDataReceiver<WindowedValue<?>> mainInput =
+ consumers.getMultiplexingConsumer(inputPCollectionId);
+ mainInput.accept(valueInGlobalWindow("0"));
+ mainInput.accept(
+ timestampedValueInGlobalWindow("1", new Instant(0L)));
+ }
+
+ @Test
+ public void testDoFnSkewAllowed() throws Exception {
+ Pipeline p = Pipeline.create();
+ PCollection<String> valuePCollection = p.apply(Create.of("0", "3"));
+ PCollection<String> outputPCollection =
+ valuePCollection.apply(TEST_TRANSFORM_ID, ParDo.of(new SkewingDoFn(new Duration(5L))));
+
+ SdkComponents sdkComponents = SdkComponents.create(p.getOptions());
+ RunnerApi.Pipeline pProto = PipelineTranslation.toProto(p, sdkComponents);
+ String inputPCollectionId = sdkComponents.registerPCollection(valuePCollection);
+ String outputPCollectionId = sdkComponents.registerPCollection(outputPCollection);
+ RunnerApi.PTransform pTransform =
+ pProto
+ .getComponents()
+ .getTransformsOrThrow(
+ pProto
+ .getComponents()
+ .getTransformsOrThrow(TEST_TRANSFORM_ID)
+ .getSubtransforms(0));
+
+ List<WindowedValue<String>> mainOutputValues = new ArrayList<>();
+ MetricsContainerStepMap metricsContainerRegistry = new MetricsContainerStepMap();
+ PCollectionConsumerRegistry consumers =
+ new PCollectionConsumerRegistry(
+ metricsContainerRegistry, mock(ExecutionStateTracker.class));
+
+ consumers.register(
+ outputPCollectionId,
+ TEST_TRANSFORM_ID,
+ (FnDataReceiver) (FnDataReceiver<WindowedValue<String>>) mainOutputValues::add,
+ StringUtf8Coder.of());
+ PTransformFunctionRegistry startFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "start");
+ PTransformFunctionRegistry finishFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "finish");
+ List<ThrowingRunnable> teardownFunctions = new ArrayList<>();
+
+ new FnApiDoFnRunner.Factory<>()
+ .createRunnerForPTransform(
+ PipelineOptionsFactory.create(),
+ null /* beamFnDataClient */,
+ null /**/,
Review comment:
```suggestion
null /* beamFnStateClient */,
```
##########
File path: sdks/java/harness/src/test/java/org/apache/beam/fn/harness/FnApiDoFnRunnerTest.java
##########
@@ -3878,6 +3885,186 @@ public void testProcessElementForWindowedTruncateAndSizeRestriction() throws Exc
}
}
+ @RunWith(JUnit4.class)
+ public static class ExceptionThrowingExecutionTest {
+ @Rule public final ExpectedException thrown = ExpectedException.none();
+
+ public static final String TEST_TRANSFORM_ID = "pTransformId";
+
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<String, String> {
+ private final Duration allowedSkew;
+
+ private SkewingDoFn(Duration allowedSkew) {
+ this.allowedSkew = allowedSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ Duration duration = new Duration(Long.valueOf(context.element()));
+ context.outputWithTimestamp(context.element(), context.timestamp().minus(duration));
+ }
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ @Test
+ public void testDoFnSkewNotAllowed() throws Exception {
+ Pipeline p = Pipeline.create();
+ PCollection<String> valuePCollection = p.apply(Create.of("0", "1"));
+ PCollection<String> outputPCollection =
+ valuePCollection.apply(TEST_TRANSFORM_ID, ParDo.of(new SkewingDoFn(Duration.ZERO)));
+
+ SdkComponents sdkComponents = SdkComponents.create(p.getOptions());
+ RunnerApi.Pipeline pProto = PipelineTranslation.toProto(p, sdkComponents);
+ String inputPCollectionId = sdkComponents.registerPCollection(valuePCollection);
+ String outputPCollectionId = sdkComponents.registerPCollection(outputPCollection);
+ RunnerApi.PTransform pTransform =
+ pProto
+ .getComponents()
+ .getTransformsOrThrow(
+ pProto
+ .getComponents()
+ .getTransformsOrThrow(TEST_TRANSFORM_ID)
+ .getSubtransforms(0));
+
+ List<WindowedValue<String>> mainOutputValues = new ArrayList<>();
+ MetricsContainerStepMap metricsContainerRegistry = new MetricsContainerStepMap();
+ PCollectionConsumerRegistry consumers =
+ new PCollectionConsumerRegistry(
+ metricsContainerRegistry, mock(ExecutionStateTracker.class));
+
+ consumers.register(
+ outputPCollectionId,
+ TEST_TRANSFORM_ID,
+ (FnDataReceiver) (FnDataReceiver<WindowedValue<String>>) mainOutputValues::add,
+ StringUtf8Coder.of());
+ PTransformFunctionRegistry startFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "start");
+ PTransformFunctionRegistry finishFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "finish");
+ List<ThrowingRunnable> teardownFunctions = new ArrayList<>();
+
+ new FnApiDoFnRunner.Factory<>()
+ .createRunnerForPTransform(
+ PipelineOptionsFactory.create(),
+ null /* beamFnDataClient */,
+ null /**/,
Review comment:
```suggestion
null /* beamFnStateClient */,
```
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -472,6 +659,40 @@ 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 = "testTimerFamily";
+ private final Duration allowedSkew;
+
+ @TimerFamily(TIMER_ID)
+ private final TimerSpec timers = TimerSpecs.timerMap(TimeDomain.PROCESSING_TIME);
Review comment:
Can you not use `TimerMap` as it limits the number of runners this can run on and use individual timers?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] je-ik commented on a change in pull request #15540: Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
je-ik commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r712816378
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,12 +1190,12 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
- "output timestamp %s should be after input message timestamp or output timestamp of firing timers %s",
+ !outputTimestamp.plus(fn.getAllowedTimestampSkew()).isBefore(elementInputTimestamp),
Review comment:
We should probably explicitly handle Long.MAX_VALUE here to make sure it does not overflow. Or is that handled by the Instant itself?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r714358223
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,12 +1190,12 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
Review comment:
So I ran the new validates runner tests against FnApiDoFnRunner. Turns out it was not properly accounting for DoFN skew when setting timers. But it wasn't doing any checking on timestamps at all for normal DoFn emit.
I added checks for the normal DoFn case and allowed for skew in the timer case. It was a bit more extensive change then I expected so PTAL. If these checks make sense I can add some unit testing for the FnApiRunner, though most of the tests seem focused on splitting and not other functionality.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r719830252
##########
File path: sdks/java/harness/src/test/java/org/apache/beam/fn/harness/FnApiDoFnRunnerTest.java
##########
@@ -1604,6 +1606,11 @@ public ProcessContinuation processElement(
}
}
+ @Override
+ public Duration getAllowedTimestampSkew() {
Review comment:
This test was triggering the check. I believe because it outputs timestamps at a given position past the min timestamp which seemed sufficiently weird / hacky that I'm not surprised it was triggering it. Hard to tell for sure but I don't believe the tests using this DoFn set timestamp on the input records at all so not surprising that the new output timestamp is less than the actual timestamp.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r746172053
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1958,6 +2071,87 @@ public Duration getAllowedTimestampSkew() {
pipeline.run();
}
+
+ @Test
+ @Category({ValidatesRunner.class, UsesTimersInParDo.class})
+ public void testRunnerNoSkew() {
Review comment:
This is an odd thing to have to optimize for but it's cleaner nevertheless, done. :)
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r746171509
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -865,6 +871,28 @@ public BundleFinalizer bundleFinalizer() {
throw new UnsupportedOperationException(
"Bundle finalization is not supported in non-portable pipelines.");
}
Review comment:
Done.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] edited a comment on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
codecov[bot] edited a comment on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-953406612
# [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#15540](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7b0de72) into [master](https://codecov.io/gh/apache/beam/commit/cd4b7f3b1af4f51bdab1a0b1a98f94b5288c09ec?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (cd4b7f3) will **increase** coverage by `0.02%`.
> The diff coverage is `n/a`.
> :exclamation: Current head 7b0de72 differs from pull request most recent head bafa6ce. Consider uploading reports for the commit bafa6ce to get more accurate results
[](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #15540 +/- ##
==========================================
+ Coverage 83.50% 83.53% +0.02%
==========================================
Files 445 445
Lines 61299 61210 -89
==========================================
- Hits 51187 51131 -56
+ Misses 10112 10079 -33
```
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [...hon/apache\_beam/runners/direct/test\_stream\_impl.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9kaXJlY3QvdGVzdF9zdHJlYW1faW1wbC5weQ==) | `94.02% <0.00%> (-2.24%)` | :arrow_down: |
| [...python/apache\_beam/runners/worker/worker\_status.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvd29ya2VyX3N0YXR1cy5weQ==) | `78.26% <0.00%> (-1.45%)` | :arrow_down: |
| [sdks/python/apache\_beam/internal/metrics/metric.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW50ZXJuYWwvbWV0cmljcy9tZXRyaWMucHk=) | `90.00% <0.00%> (-1.00%)` | :arrow_down: |
| [...ks/python/apache\_beam/runners/worker/sdk\_worker.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvc2RrX3dvcmtlci5weQ==) | `89.25% <0.00%> (-0.17%)` | :arrow_down: |
| [...dks/python/apache\_beam/metrics/monitoring\_infos.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vbWV0cmljcy9tb25pdG9yaW5nX2luZm9zLnB5) | `96.93% <0.00%> (-0.07%)` | :arrow_down: |
| [.../python/apache\_beam/io/gcp/resource\_identifiers.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vaW8vZ2NwL3Jlc291cmNlX2lkZW50aWZpZXJzLnB5) | `100.00% <0.00%> (ø)` | |
| [.../python/apache\_beam/portability/api/metrics\_pb2.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcG9ydGFiaWxpdHkvYXBpL21ldHJpY3NfcGIyLnB5) | `100.00% <0.00%> (ø)` | |
| [...on/apache\_beam/portability/api/metrics\_pb2\_urns.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcG9ydGFiaWxpdHkvYXBpL21ldHJpY3NfcGIyX3VybnMucHk=) | `100.00% <0.00%> (ø)` | |
| [...hon/apache\_beam/runners/worker/bundle\_processor.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy93b3JrZXIvYnVuZGxlX3Byb2Nlc3Nvci5weQ==) | `93.64% <0.00%> (+0.12%)` | :arrow_up: |
| [...runners/interactive/display/pcoll\_visualization.py](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9weXRob24vYXBhY2hlX2JlYW0vcnVubmVycy9pbnRlcmFjdGl2ZS9kaXNwbGF5L3Bjb2xsX3Zpc3VhbGl6YXRpb24ucHk=) | `86.36% <0.00%> (+0.50%)` | :arrow_up: |
| ... and [3 more](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [cd4b7f3...bafa6ce](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r713451202
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,12 +1190,12 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
- "output timestamp %s should be after input message timestamp or output timestamp of firing timers %s",
+ !outputTimestamp.plus(fn.getAllowedTimestampSkew()).isBefore(elementInputTimestamp),
Review comment:
Seems like it's not handled by the instant. I copied what is done above which is to just special case the MAX_VALUE. Technically I think this could still overflow if it's almost the MAX_VALUE? But seems like it's pretty unlikely that it would be almost max value but not max value. I also added a test for the max value case.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r719810671
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,13 +1190,20 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
- checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
- "output timestamp %s should be after input message timestamp or output timestamp of firing timers %s",
- outputTimestamp,
- elementInputTimestamp);
+ // The documentation of getAllowedTimestampSkew explicitly permits Long.MAX_VALUE to be used
+ // for infinite skew. Defend against underflow in that case for timestamps before the epoch.
+ if (fn.getAllowedTimestampSkew().getMillis() != Long.MAX_VALUE
Review comment:
Sounds good.
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,13 +1190,20 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
- checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
- "output timestamp %s should be after input message timestamp or output timestamp of firing timers %s",
- outputTimestamp,
- elementInputTimestamp);
+ // The documentation of getAllowedTimestampSkew explicitly permits Long.MAX_VALUE to be used
+ // for infinite skew. Defend against underflow in that case for timestamps before the epoch.
+ if (fn.getAllowedTimestampSkew().getMillis() != Long.MAX_VALUE
Review comment:
Sounds good. Done.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #15540: Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-924151564
Is there an associated JIRA?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-936715458
> > Won't this allow for infinite skew since if have a timer at `X` and skew of `-1` then the first time the timer is processed you can output at time `X-1` and when it gets scheduled again you can now output at `X-2` since the the new timers timestamp is `X-1`?
>
> So my understanding of the reason for these checks is to stop people from doing the wrong thing without realizing it. We don't even take any different action based on this variable. It seems okay to apply this to each specific output timestamp and let you skew more if you chain timers in this fashion.
>
> On a more practical note, there's reasons why you might want a timer to output an earlier element if you've properly set up watermark holds. There's currently no way to do that so we need some allowance. It would probably be better if we could constrain skew from the first output timestamp but I don't think that's available in the later timers, right?
>
> If you disagree with the approach, I can bring this up on the email thread for others to chime in in case they are not checking here.
I think users will be surprised that their data will be dropped as late once they pass the watermark skew bound if they output past it. The existing logic had guards for this explicitly since it would be surprising for users so I do believe it is important enough to discuss whether there is another approach to solve this or we are ok with this happening.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r737743322
##########
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:
talked offline and fixed it so it properly handles this case.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
lukecwik commented on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-966495531
Run Java PreCommit
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r714358223
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,12 +1190,12 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
Review comment:
So I ran the new validates runner tests against FnApiDoFnRunner. Turns out it was not properly accounting for DoFN skew when setting timers. But it wasn't doing any checking on timestamps at all for normal DoFn emit.
I added checks for the normal DoFn case and allowed for skew in the timer case. It was a bit more extensive change then I expected so PTAL. If these checks make sense I can add some unit testing for the FnApiRunner.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r719830252
##########
File path: sdks/java/harness/src/test/java/org/apache/beam/fn/harness/FnApiDoFnRunnerTest.java
##########
@@ -1604,6 +1606,11 @@ public ProcessContinuation processElement(
}
}
+ @Override
+ public Duration getAllowedTimestampSkew() {
Review comment:
This test was triggering the check. I believe because it outputs timestamps at a given position past the min timestamp which seemed sufficiently weird / hacky that I'm not surprised it was triggering it.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r731267071
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -386,6 +399,180 @@ public void testInfiniteSkew() {
BoundedWindow.TIMESTAMP_MAX_VALUE));
}
+ @Test
+ @Category({ValidatesRunner.class})
+ public void testRunnerNoSkew() {
+ PCollection<Duration> input =
+ p.apply("create", Create.timestamped(Arrays.asList(new Duration(0L), new Duration(1L)), Arrays.asList(0L, 1L)));
+ input.apply(ParDo.of(new SkewingDoFn(Duration.ZERO)));
+ // The errors differ between runners but at least check that the output timestamp is printed.
+ thrown.expectMessage(String.format("%s", new Instant(0L)));
Review comment:
done.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r732207960
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -386,6 +397,186 @@ public void testInfiniteSkew() {
BoundedWindow.TIMESTAMP_MAX_VALUE));
}
+ @Test
+ @Category({ValidatesRunner.class, UsesTimersInParDo.class})
+ public void testRunnerNoSkew() {
Review comment:
Done.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r714358223
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,12 +1190,12 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
Review comment:
So I ran the new validates runner tests against FnApiDoFnRunner. Turns out it was not properly accounting for DoFN skew when setting timers. But it wasn't doing any checking on timestamps at all for normal DoFn emit.
I added checks for the normal DoFn case and allowed for skew in the timer case.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-964692427
@lukecwik PTAL, still need to check that everything runs internally but otherwise should be good.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r737900079
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1791,6 +1791,119 @@ public void finishBundle(FinishBundleContext c) {
/** Tests to validate output timestamps. */
@RunWith(JUnit4.class)
public static class TimestampTests extends SharedTestBase implements Serializable {
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<Duration, String> {
+ private final Duration allowedSkew;
+ static final String CORRECT_ELEMENT = "element";
+
+ private SkewingDoFn(Duration allowedSkew) {
+ this.allowedSkew = allowedSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ Instant outputTimestamp = context.timestamp().minus(context.element());
+ try {
+ context.outputWithTimestamp(CORRECT_ELEMENT, outputTimestamp);
+ } catch (IllegalArgumentException e) {
+ // Errors differ between runners but at least check that the output timestamp is printed.
+ if (e.getMessage().contains("timestamp " + outputTimestamp)) {
+ context.output(outputTimestamp.toString());
+ }
+ }
+ }
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ /**
+ * 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>, String> {
+ static final String TIMER_ID = "testTimerId";
+ private final Duration allowedSkew;
+ static final String CORRECT_ELEMENT = "element";
+
+ @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) {
+
Review comment:
```suggestion
```
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -865,6 +871,28 @@ public BundleFinalizer bundleFinalizer() {
throw new UnsupportedOperationException(
"Bundle finalization is not supported in non-portable pipelines.");
}
Review comment:
We should be checking timestamp `OnWindowExpiration#outputWithTimestamp` as well:
https://github.com/apache/beam/blob/7b0de72a84298f19365d3a2166822105d534c92b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java#L1053
https://github.com/apache/beam/blob/7b0de72a84298f19365d3a2166822105d534c92b/runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java#L1063
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1791,6 +1791,119 @@ public void finishBundle(FinishBundleContext c) {
/** Tests to validate output timestamps. */
@RunWith(JUnit4.class)
public static class TimestampTests extends SharedTestBase implements Serializable {
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<Duration, String> {
+ private final Duration allowedSkew;
+ static final String CORRECT_ELEMENT = "element";
+
+ private SkewingDoFn(Duration allowedSkew) {
+ this.allowedSkew = allowedSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ Instant outputTimestamp = context.timestamp().minus(context.element());
+ try {
+ context.outputWithTimestamp(CORRECT_ELEMENT, outputTimestamp);
+ } catch (IllegalArgumentException e) {
+ // Errors differ between runners but at least check that the output timestamp is printed.
+ if (e.getMessage().contains("timestamp " + outputTimestamp)) {
+ context.output(outputTimestamp.toString());
+ }
+ }
+ }
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ /**
+ * 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>, String> {
Review comment:
```suggestion
private static class ProcessElementTimerSkewingDoFn extends DoFn<KV<String, Duration>, String> {
```
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1791,6 +1791,119 @@ public void finishBundle(FinishBundleContext c) {
/** Tests to validate output timestamps. */
@RunWith(JUnit4.class)
public static class TimestampTests extends SharedTestBase implements Serializable {
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<Duration, String> {
Review comment:
nit: I would consider combining the DoFns into three variants:
* ProcessElementTimestampSkewingDoFn
* OnTimerTimestampSkewingDoFn
* OnWindowExpirationTimestampSkewingDoFn
and ensure that you test the normal `outputWithTimestamp` and timer output timestamp in one go instead of trying to have a different DoFn for each. This will also help with the currently confusing naming that exists (including the names I suggested).
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1791,6 +1791,119 @@ public void finishBundle(FinishBundleContext c) {
/** Tests to validate output timestamps. */
@RunWith(JUnit4.class)
public static class TimestampTests extends SharedTestBase implements Serializable {
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<Duration, String> {
+ private final Duration allowedSkew;
+ static final String CORRECT_ELEMENT = "element";
+
+ private SkewingDoFn(Duration allowedSkew) {
+ this.allowedSkew = allowedSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ Instant outputTimestamp = context.timestamp().minus(context.element());
+ try {
+ context.outputWithTimestamp(CORRECT_ELEMENT, outputTimestamp);
+ } catch (IllegalArgumentException e) {
+ // Errors differ between runners but at least check that the output timestamp is printed.
+ if (e.getMessage().contains("timestamp " + outputTimestamp)) {
+ context.output(outputTimestamp.toString());
+ }
+ }
+ }
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ /**
+ * 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>, String> {
+ static final String TIMER_ID = "testTimerId";
+ private final Duration allowedSkew;
+ static final String CORRECT_ELEMENT = "element";
+
+ @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) {
+
+ Instant outputTimestamp = context.timestamp().minus(context.element().getValue());
+ try {
+ timer.withOutputTimestamp(outputTimestamp).set(new Instant(0));
+ context.output(CORRECT_ELEMENT);
+ } catch (IllegalArgumentException e) {
+ // Errors differ between runners but at least check that the output timestamp is printed.
Review comment:
ditto on messaging consolidation
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1791,6 +1791,119 @@ public void finishBundle(FinishBundleContext c) {
/** Tests to validate output timestamps. */
@RunWith(JUnit4.class)
public static class TimestampTests extends SharedTestBase implements Serializable {
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<Duration, String> {
+ private final Duration allowedSkew;
+ static final String CORRECT_ELEMENT = "element";
+
+ private SkewingDoFn(Duration allowedSkew) {
+ this.allowedSkew = allowedSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ Instant outputTimestamp = context.timestamp().minus(context.element());
+ try {
+ context.outputWithTimestamp(CORRECT_ELEMENT, outputTimestamp);
+ } catch (IllegalArgumentException e) {
+ // Errors differ between runners but at least check that the output timestamp is printed.
+ if (e.getMessage().contains("timestamp " + outputTimestamp)) {
+ context.output(outputTimestamp.toString());
+ }
+ }
+ }
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ /**
+ * 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>, String> {
+ static final String TIMER_ID = "testTimerId";
+ private final Duration allowedSkew;
+ static final String CORRECT_ELEMENT = "element";
+
+ @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) {
+
+ Instant outputTimestamp = context.timestamp().minus(context.element().getValue());
+ try {
+ timer.withOutputTimestamp(outputTimestamp).set(new Instant(0));
+ context.output(CORRECT_ELEMENT);
+ } catch (IllegalArgumentException e) {
+ // Errors differ between runners but at least check that the output timestamp is printed.
+ if (e.getMessage().contains("timestamp " + outputTimestamp)) {
+ context.output(outputTimestamp.toString());
+ }
+ }
+ }
+
+ @OnTimer(TIMER_ID)
+ public void onTimer(OnTimerContext context) {}
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ /**
+ * 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 TimerOutputSkewingDoFn extends DoFn<KV<String, String>, String> {
Review comment:
```suggestion
private static class OnTimerOutputSkewingDoFn extends DoFn<KV<String, String>, String> {
```
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1791,6 +1791,119 @@ public void finishBundle(FinishBundleContext c) {
/** Tests to validate output timestamps. */
@RunWith(JUnit4.class)
public static class TimestampTests extends SharedTestBase implements Serializable {
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<Duration, String> {
+ private final Duration allowedSkew;
+ static final String CORRECT_ELEMENT = "element";
+
+ private SkewingDoFn(Duration allowedSkew) {
+ this.allowedSkew = allowedSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ Instant outputTimestamp = context.timestamp().minus(context.element());
+ try {
+ context.outputWithTimestamp(CORRECT_ELEMENT, outputTimestamp);
+ } catch (IllegalArgumentException e) {
+ // Errors differ between runners but at least check that the output timestamp is printed.
+ if (e.getMessage().contains("timestamp " + outputTimestamp)) {
+ context.output(outputTimestamp.toString());
+ }
+ }
+ }
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ /**
+ * 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>, String> {
+ static final String TIMER_ID = "testTimerId";
+ private final Duration allowedSkew;
+ static final String CORRECT_ELEMENT = "element";
+
+ @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) {
+
+ Instant outputTimestamp = context.timestamp().minus(context.element().getValue());
+ try {
+ timer.withOutputTimestamp(outputTimestamp).set(new Instant(0));
+ context.output(CORRECT_ELEMENT);
+ } catch (IllegalArgumentException e) {
+ // Errors differ between runners but at least check that the output timestamp is printed.
+ if (e.getMessage().contains("timestamp " + outputTimestamp)) {
+ context.output(outputTimestamp.toString());
+ }
+ }
+ }
+
+ @OnTimer(TIMER_ID)
+ public void onTimer(OnTimerContext context) {}
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ /**
+ * 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 TimerOutputSkewingDoFn extends DoFn<KV<String, String>, String> {
+ static final String TIMER_ID = "testTimerId";
+ static final String CORRECT_ELEMENT = "element";
+ private final Duration allowedSkew;
+ private final Duration outputTimestampSkew;
+
+ @TimerId(TIMER_ID)
+ private static final TimerSpec timer = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private TimerOutputSkewingDoFn(Duration allowedSkew, Duration outputTimestampSkew) {
+ this.allowedSkew = allowedSkew;
+ this.outputTimestampSkew = outputTimestampSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context, @TimerId(TIMER_ID) Timer timer) {
+ timer.set(context.timestamp());
+ }
+
+ @OnTimer(TIMER_ID)
+ public void onTimer(OnTimerContext context) {
+ Instant outputTimestamp = context.timestamp().minus(outputTimestampSkew);
Review comment:
We should also have a variant that covers timer skew within the `onTimer` method and also one that checks skew for `OnWindowExpiration`
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1791,6 +1791,119 @@ public void finishBundle(FinishBundleContext c) {
/** Tests to validate output timestamps. */
@RunWith(JUnit4.class)
public static class TimestampTests extends SharedTestBase implements Serializable {
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<Duration, String> {
+ private final Duration allowedSkew;
+ static final String CORRECT_ELEMENT = "element";
+
+ private SkewingDoFn(Duration allowedSkew) {
+ this.allowedSkew = allowedSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ Instant outputTimestamp = context.timestamp().minus(context.element());
+ try {
+ context.outputWithTimestamp(CORRECT_ELEMENT, outputTimestamp);
+ } catch (IllegalArgumentException e) {
+ // Errors differ between runners but at least check that the output timestamp is printed.
Review comment:
I was under the impression that we consolidated all the error messages to be the same, is this still not the case?
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1958,6 +2071,87 @@ public Duration getAllowedTimestampSkew() {
pipeline.run();
}
+
+ @Test
+ @Category({ValidatesRunner.class, UsesTimersInParDo.class})
+ public void testRunnerNoSkew() {
Review comment:
If we combine the DoFns into the three suggested can we rename these tests to:
* testProcessElementTimestampYYYSkew()
* testOnTimerTimestampYYYSkew()
* testOnWindowExpirationTimestampYYYSkew()
where `YYY` is `Allowed` or `Denied`
otherwise you can have names like `testProcessElementOutputSkewingYYYSkew()`
Finally, it seems like it would be pretty easy to combine the allowed and denied case into a singular test case. Will help reduce the explosion of cases since each validates runner test will add minutes of runtime to the Dataflow validates runner suite.
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1791,6 +1791,119 @@ public void finishBundle(FinishBundleContext c) {
/** Tests to validate output timestamps. */
@RunWith(JUnit4.class)
public static class TimestampTests extends SharedTestBase implements Serializable {
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<Duration, String> {
Review comment:
```suggestion
private static class ProcessElementOutputSkewingDoFn extends DoFn<Duration, String> {
```
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] edited a comment on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
codecov[bot] edited a comment on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-953406612
# [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#15540](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7b0de72) into [master](https://codecov.io/gh/apache/beam/commit/bc82f6f84e72c520be397977edf4bf645782ac8f?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (bc82f6f) will **increase** coverage by `37.42%`.
> The diff coverage is `n/a`.
> :exclamation: Current head 7b0de72 differs from pull request most recent head c43415e. Consider uploading reports for the commit c43415e to get more accurate results
[](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #15540 +/- ##
===========================================
+ Coverage 46.10% 83.53% +37.42%
===========================================
Files 196 445 +249
Lines 19498 61210 +41712
===========================================
+ Hits 8990 51131 +42141
- Misses 9538 10079 +541
+ Partials 970 0 -970
```
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [sdks/go/pkg/beam/io/textio/sdf.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9pby90ZXh0aW8vc2RmLmdv) | | |
| [sdks/go/pkg/beam/testing/teststream/teststream.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS90ZXN0aW5nL3Rlc3RzdHJlYW0vdGVzdHN0cmVhbS5nbw==) | | |
| [sdks/go/pkg/beam/core/util/reflectx/util.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3V0aWwvcmVmbGVjdHgvdXRpbC5nbw==) | | |
| [...ks/go/pkg/beam/core/runtime/coderx/coderx.shims.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvY29kZXJ4L2NvZGVyeC5zaGltcy5nbw==) | | |
| [sdks/go/pkg/beam/core/runtime/exec/emit.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvZXhlYy9lbWl0Lmdv) | | |
| [sdks/go/pkg/beam/core/runtime/genx/genx.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvZ2VueC9nZW54Lmdv) | | |
| [sdks/go/pkg/beam/validate.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS92YWxpZGF0ZS5nbw==) | | |
| [sdks/go/pkg/beam/core/graph/coder/bytes.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL2dyYXBoL2NvZGVyL2J5dGVzLmdv) | | |
| [sdks/go/pkg/beam/transforms/stats/sum\_switch.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS90cmFuc2Zvcm1zL3N0YXRzL3N1bV9zd2l0Y2guZ28=) | | |
| [sdks/go/pkg/beam/testing/passert/count.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS90ZXN0aW5nL3Bhc3NlcnQvY291bnQuZ28=) | | |
| ... and [631 more](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [bc82f6f...c43415e](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r719827260
##########
File path: sdks/java/harness/src/test/java/org/apache/beam/fn/harness/FnApiDoFnRunnerTest.java
##########
@@ -3878,6 +3885,186 @@ public void testProcessElementForWindowedTruncateAndSizeRestriction() throws Exc
}
}
+ @RunWith(JUnit4.class)
+ public static class ExceptionThrowingExecutionTest {
+ @Rule public final ExpectedException thrown = ExpectedException.none();
+
+ public static final String TEST_TRANSFORM_ID = "pTransformId";
+
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<String, String> {
+ private final Duration allowedSkew;
+
+ private SkewingDoFn(Duration allowedSkew) {
+ this.allowedSkew = allowedSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ Duration duration = new Duration(Long.valueOf(context.element()));
+ context.outputWithTimestamp(context.element(), context.timestamp().minus(duration));
+ }
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ @Test
+ public void testDoFnSkewNotAllowed() throws Exception {
+ Pipeline p = Pipeline.create();
+ PCollection<String> valuePCollection = p.apply(Create.of("0", "1"));
+ PCollection<String> outputPCollection =
+ valuePCollection.apply(TEST_TRANSFORM_ID, ParDo.of(new SkewingDoFn(Duration.ZERO)));
+
+ SdkComponents sdkComponents = SdkComponents.create(p.getOptions());
+ RunnerApi.Pipeline pProto = PipelineTranslation.toProto(p, sdkComponents);
+ String inputPCollectionId = sdkComponents.registerPCollection(valuePCollection);
+ String outputPCollectionId = sdkComponents.registerPCollection(outputPCollection);
+ RunnerApi.PTransform pTransform =
+ pProto
+ .getComponents()
+ .getTransformsOrThrow(
+ pProto
+ .getComponents()
+ .getTransformsOrThrow(TEST_TRANSFORM_ID)
+ .getSubtransforms(0));
+
+ List<WindowedValue<String>> mainOutputValues = new ArrayList<>();
+ MetricsContainerStepMap metricsContainerRegistry = new MetricsContainerStepMap();
+ PCollectionConsumerRegistry consumers =
+ new PCollectionConsumerRegistry(
+ metricsContainerRegistry, mock(ExecutionStateTracker.class));
+
+ consumers.register(
+ outputPCollectionId,
+ TEST_TRANSFORM_ID,
+ (FnDataReceiver) (FnDataReceiver<WindowedValue<String>>) mainOutputValues::add,
+ StringUtf8Coder.of());
+ PTransformFunctionRegistry startFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "start");
+ PTransformFunctionRegistry finishFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "finish");
+ List<ThrowingRunnable> teardownFunctions = new ArrayList<>();
+
+ new FnApiDoFnRunner.Factory<>()
+ .createRunnerForPTransform(
+ PipelineOptionsFactory.create(),
+ null /* beamFnDataClient */,
+ null /**/,
+ null /* beamFnTimerClient */,
+ TEST_TRANSFORM_ID,
+ pTransform,
+ Suppliers.ofInstance("57L")::get,
+ pProto.getComponents().getPcollectionsMap(),
+ pProto.getComponents().getCodersMap(),
+ pProto.getComponents().getWindowingStrategiesMap(),
+ consumers,
+ startFunctionRegistry,
+ finishFunctionRegistry,
+ null /* addResetFunction */,
+ teardownFunctions::add,
+ null /* addProgressRequestCallback */,
+ null /* splitListener */,
+ null /* bundleFinalizer */);
+
+ thrown.expect(UserCodeException.class);
+ thrown.expectMessage(
+ String.format("timestamp %s", new Instant(0).minus(new Duration(1L))));
+ thrown.expectMessage(
+ String.format(
+ "allowed skew (%s)",
+ PeriodFormat.getDefault().print(Duration.ZERO.toPeriod())));
+
+ Iterables.getOnlyElement(startFunctionRegistry.getFunctions()).run();
+ mainOutputValues.clear();
+
+ FnDataReceiver<WindowedValue<?>> mainInput =
+ consumers.getMultiplexingConsumer(inputPCollectionId);
+ mainInput.accept(valueInGlobalWindow("0"));
+ mainInput.accept(
+ timestampedValueInGlobalWindow("1", new Instant(0L)));
+ }
+
+ @Test
+ public void testDoFnSkewAllowed() throws Exception {
+ Pipeline p = Pipeline.create();
+ PCollection<String> valuePCollection = p.apply(Create.of("0", "3"));
+ PCollection<String> outputPCollection =
+ valuePCollection.apply(TEST_TRANSFORM_ID, ParDo.of(new SkewingDoFn(new Duration(5L))));
+
+ SdkComponents sdkComponents = SdkComponents.create(p.getOptions());
+ RunnerApi.Pipeline pProto = PipelineTranslation.toProto(p, sdkComponents);
+ String inputPCollectionId = sdkComponents.registerPCollection(valuePCollection);
+ String outputPCollectionId = sdkComponents.registerPCollection(outputPCollection);
+ RunnerApi.PTransform pTransform =
+ pProto
+ .getComponents()
+ .getTransformsOrThrow(
+ pProto
+ .getComponents()
+ .getTransformsOrThrow(TEST_TRANSFORM_ID)
+ .getSubtransforms(0));
+
+ List<WindowedValue<String>> mainOutputValues = new ArrayList<>();
+ MetricsContainerStepMap metricsContainerRegistry = new MetricsContainerStepMap();
+ PCollectionConsumerRegistry consumers =
+ new PCollectionConsumerRegistry(
+ metricsContainerRegistry, mock(ExecutionStateTracker.class));
+
+ consumers.register(
+ outputPCollectionId,
+ TEST_TRANSFORM_ID,
+ (FnDataReceiver) (FnDataReceiver<WindowedValue<String>>) mainOutputValues::add,
+ StringUtf8Coder.of());
+ PTransformFunctionRegistry startFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "start");
+ PTransformFunctionRegistry finishFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "finish");
+ List<ThrowingRunnable> teardownFunctions = new ArrayList<>();
+
+ new FnApiDoFnRunner.Factory<>()
+ .createRunnerForPTransform(
+ PipelineOptionsFactory.create(),
+ null /* beamFnDataClient */,
+ null /**/,
Review comment:
done.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r746960039
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1957,6 +2125,84 @@ public Duration getAllowedTimestampSkew() {
pipeline.run();
}
+
+ @Test
+ @Category({ValidatesRunner.class, UsesTimersInParDo.class})
+ public void testProcessElementSkew() {
+ List<KV<String, Duration>> durations =
+ Arrays.asList(KV.of("0", Duration.millis(0L)), KV.of("2", Duration.millis(1L)));
+ PCollection<KV<String, Duration>> input =
+ pipeline.apply("create", Create.timestamped(durations, Arrays.asList(0L, 1L)));
+
+ PCollection<String> noSkew =
+ input.apply("noSkew", ParDo.of(new ProcessElementTimestampSkewingDoFn(Duration.ZERO)));
+ PAssert.that(noSkew)
+ .containsInAnyOrder(
+ TIMER_ELEMENT,
+ OUTPUT_ELEMENT,
+ TIMER_ELEMENT + new Instant(0L),
+ OUTPUT_ELEMENT + new Instant(0L));
+
+ PCollection<String> skew =
+ input.apply(
+ "skew", ParDo.of(new ProcessElementTimestampSkewingDoFn(Duration.millis(2L))));
+ PAssert.that(skew)
+ .containsInAnyOrder(TIMER_ELEMENT, OUTPUT_ELEMENT, TIMER_ELEMENT, OUTPUT_ELEMENT);
+ pipeline.run();
+ }
+
+ @Test
+ @Category({UsesTimersInParDo.class, ValidatesRunner.class})
+ public void testOnTimerTimestampSkew() {
+ List<KV<String, String>> durations = Arrays.asList(KV.of("0", "0"));
+ PCollection<KV<String, String>> input =
+ pipeline.apply("create", Create.timestamped(durations, Arrays.asList(0L)));
+ PCollection<String> noSkew =
+ input.apply(
+ "noskew",
+ ParDo.of(new OnTimerTimestampSkewingDoFn(Duration.millis(0L), Duration.millis(3L))));
+ PAssert.that(noSkew)
+ .containsInAnyOrder(OUTPUT_ELEMENT + new Instant(-3L), TIMER_ELEMENT + new Instant(-3L));
+ PCollection<String> skew =
+ input.apply(
+ "skew",
+ ParDo.of(new OnTimerTimestampSkewingDoFn(Duration.millis(3L), Duration.millis(2L))));
+ PAssert.that(skew).containsInAnyOrder(OUTPUT_ELEMENT, TIMER_ELEMENT);
+
+ pipeline.run();
+ }
+
+ @Test
+ @Category({UsesOnWindowExpiration.class, UsesTimersInParDo.class, ValidatesRunner.class})
+ public void testOnWindowTimestampSkew() {
+ Duration windowDuration = Duration.millis(10L);
+ List<KV<String, String>> durations = Arrays.asList(KV.of("key", "0"));
+ PCollection<KV<String, String>> input =
+ pipeline.apply("create", Create.timestamped(durations, Arrays.asList(0L)));
+
+ PCollection<String> noSkew =
+ input
+ .apply("noSkewWindow", Window.into(FixedWindows.of(windowDuration)))
+ .apply(
+ "noskew",
+ ParDo.of(
+ new OnWindowExpirationTimestampSkewingDoFn(
+ Duration.millis(0L), Duration.millis(3L))));
+ PAssert.that(noSkew)
+ .containsInAnyOrder(
+ OUTPUT_ELEMENT + new Instant(windowDuration.minus(3L).minus(1L).getMillis()));
Review comment:
> Task :sdks:java:core:compileTestJava
```
/home/runner/work/beam/beam/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java:2196: warning: [JodaPlusMinusLong] Use of JodaTime's type.plus(long) or type.minus(long) is not allowed (where <type> = {Duration,Instant,DateTime,DateMidnight}). Please use type.plus(Duration.millis(long)) or type.minus(Duration.millis(long)) instead.
OUTPUT_ELEMENT + new Instant(windowDuration.minus(3L).minus(1L).getMillis()));
^
(see https://errorprone.info/bugpattern/JodaPlusMinusLong)
Did you mean 'OUTPUT_ELEMENT + new Instant(windowDuration.minus(Duration.millis(3L)).minus(1L).getMillis()));'?
```
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1790,6 +1791,173 @@ public void finishBundle(FinishBundleContext c) {
/** Tests to validate output timestamps. */
@RunWith(JUnit4.class)
public static class TimestampTests extends SharedTestBase implements Serializable {
+
+ static final String TIMER_ELEMENT = "timer";
+ static final String OUTPUT_ELEMENT = "output";
+
+ /**
+ * Checks that the given message is correct and includes the element timestamp, allowed skew,
+ * and output timestamp.
+ */
+ static boolean hasExpectedError(
+ IllegalArgumentException e,
+ Duration allowedSkew,
+ Instant elementTimestamp,
+ Instant outputTimestamp) {
+ return e.getMessage().contains("timestamp of the ")
+ && e.getMessage().contains(elementTimestamp.toString())
+ && e.getMessage().contains("timestamp " + outputTimestamp)
+ && e.getMessage()
+ .contains("allowed skew (" + PeriodFormat.getDefault().print(allowedSkew.toPeriod()))
+ && e.getMessage().contains("getAllowedTimestampSkew");
+ }
+
+ /**
+ * A {@link DoFn} that outputs an element at and 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 ProcessElementTimestampSkewingDoFn
+ extends DoFn<KV<String, Duration>, String> {
+
+ static final String TIMER_ID = "testTimerId";
+ private final Duration allowedSkew;
+
+ @TimerId(TIMER_ID)
+ private static final TimerSpec timer = TimerSpecs.timer(TimeDomain.PROCESSING_TIME);
+
+ private ProcessElementTimestampSkewingDoFn(Duration allowedSkew) {
+ this.allowedSkew = allowedSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context, @TimerId(TIMER_ID) Timer timer) {
+ Instant outputTimestamp = context.timestamp().minus(context.element().getValue());
+ try {
+ context.outputWithTimestamp(OUTPUT_ELEMENT, outputTimestamp);
+ } catch (IllegalArgumentException e) {
+ if (hasExpectedError(e, allowedSkew, context.timestamp(), outputTimestamp)) {
+ context.output(OUTPUT_ELEMENT + outputTimestamp.toString());
+ }
+ }
+ try {
+ timer.withOutputTimestamp(outputTimestamp).set(new Instant(0));
+ context.output(TIMER_ELEMENT);
+ } catch (IllegalArgumentException e) {
+ if (hasExpectedError(e, allowedSkew, context.timestamp(), outputTimestamp)) {
+ context.output(TIMER_ELEMENT + outputTimestamp.toString());
+ }
+ }
+ }
+
+ @OnTimer(TIMER_ID)
+ public void onTimer(OnTimerContext context) {}
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ /**
+ * A {@link DoFn} that sets a timer that outputs an element and sets a second 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 OnTimerTimestampSkewingDoFn extends DoFn<KV<String, String>, String> {
+
+ static final String FIRST_TIMER_ID = "firstTestTimerId";
+ static final String SECOND_TIMER_ID = "secondTestTimerId";
+ private final Duration allowedSkew;
+ private final Duration outputTimestampSkew;
+
+ @TimerId(FIRST_TIMER_ID)
+ private static final TimerSpec firstTimer = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ @TimerId(SECOND_TIMER_ID)
+ private static final TimerSpec secondTimer = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private OnTimerTimestampSkewingDoFn(Duration allowedSkew, Duration outputTimestampSkew) {
+ this.allowedSkew = allowedSkew;
+ this.outputTimestampSkew = outputTimestampSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context, @TimerId(FIRST_TIMER_ID) Timer timer) {
+ timer.set(context.timestamp());
+ }
+
+ @OnTimer(SECOND_TIMER_ID)
+ public void onSecondTimer(OnTimerContext context) {}
+
+ @OnTimer(FIRST_TIMER_ID)
+ public void onFirstTimer(OnTimerContext context, @TimerId(SECOND_TIMER_ID) Timer timer) {
+ Instant outputTimestamp = context.timestamp().minus(outputTimestampSkew);
+ try {
+ context.outputWithTimestamp(OUTPUT_ELEMENT, outputTimestamp);
+ } catch (IllegalArgumentException e) {
+ if (hasExpectedError(e, allowedSkew, context.timestamp(), outputTimestamp)) {
+ context.output(OUTPUT_ELEMENT + outputTimestamp);
+ }
+ }
+ try {
+ timer.withOutputTimestamp(outputTimestamp).set(context.timestamp());
+ context.output(TIMER_ELEMENT);
+ } catch (IllegalArgumentException e) {
+ if (hasExpectedError(e, allowedSkew, context.timestamp(), outputTimestamp)) {
+ context.output(TIMER_ELEMENT + outputTimestamp);
+ }
+ }
+ }
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ /**
+ * A {@link DoFn} that on window expiration outputs an element with an output timestamp equal
+ * output timestamp equal to the input timestamp minus the input element's value. Keys are
+ * ignored but required for timers.
Review comment:
```suggestion
* A {@link DoFn} that on window expiration outputs an element with an output timestamp equal
* to the input timestamp minus the input element's value. Keys are ignored but required for timers.
```
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1957,6 +2125,84 @@ public Duration getAllowedTimestampSkew() {
pipeline.run();
}
+
+ @Test
+ @Category({ValidatesRunner.class, UsesTimersInParDo.class})
+ public void testProcessElementSkew() {
+ List<KV<String, Duration>> durations =
+ Arrays.asList(KV.of("0", Duration.millis(0L)), KV.of("2", Duration.millis(1L)));
+ PCollection<KV<String, Duration>> input =
+ pipeline.apply("create", Create.timestamped(durations, Arrays.asList(0L, 1L)));
+
+ PCollection<String> noSkew =
+ input.apply("noSkew", ParDo.of(new ProcessElementTimestampSkewingDoFn(Duration.ZERO)));
+ PAssert.that(noSkew)
Review comment:
The logic within `testOnTimerTimestampSkew` is much simpler to understand because the skew and noskew variants are differentiated.
Do you mind splitting this one up similarly?
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r746171825
##########
File path: sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/ParDoTest.java
##########
@@ -1791,6 +1791,119 @@ public void finishBundle(FinishBundleContext c) {
/** Tests to validate output timestamps. */
@RunWith(JUnit4.class)
public static class TimestampTests extends SharedTestBase implements Serializable {
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<Duration, String> {
+ private final Duration allowedSkew;
+ static final String CORRECT_ELEMENT = "element";
+
+ private SkewingDoFn(Duration allowedSkew) {
+ this.allowedSkew = allowedSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ Instant outputTimestamp = context.timestamp().minus(context.element());
+ try {
+ context.outputWithTimestamp(CORRECT_ELEMENT, outputTimestamp);
+ } catch (IllegalArgumentException e) {
+ // Errors differ between runners but at least check that the output timestamp is printed.
+ if (e.getMessage().contains("timestamp " + outputTimestamp)) {
+ context.output(outputTimestamp.toString());
+ }
+ }
+ }
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ /**
+ * 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>, String> {
+ static final String TIMER_ID = "testTimerId";
+ private final Duration allowedSkew;
+ static final String CORRECT_ELEMENT = "element";
+
+ @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) {
+
+ Instant outputTimestamp = context.timestamp().minus(context.element().getValue());
+ try {
+ timer.withOutputTimestamp(outputTimestamp).set(new Instant(0));
+ context.output(CORRECT_ELEMENT);
+ } catch (IllegalArgumentException e) {
+ // Errors differ between runners but at least check that the output timestamp is printed.
+ if (e.getMessage().contains("timestamp " + outputTimestamp)) {
+ context.output(outputTimestamp.toString());
+ }
+ }
+ }
+
+ @OnTimer(TIMER_ID)
+ public void onTimer(OnTimerContext context) {}
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ /**
+ * 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 TimerOutputSkewingDoFn extends DoFn<KV<String, String>, String> {
+ static final String TIMER_ID = "testTimerId";
+ static final String CORRECT_ELEMENT = "element";
+ private final Duration allowedSkew;
+ private final Duration outputTimestampSkew;
+
+ @TimerId(TIMER_ID)
+ private static final TimerSpec timer = TimerSpecs.timer(TimeDomain.EVENT_TIME);
+
+ private TimerOutputSkewingDoFn(Duration allowedSkew, Duration outputTimestampSkew) {
+ this.allowedSkew = allowedSkew;
+ this.outputTimestampSkew = outputTimestampSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context, @TimerId(TIMER_ID) Timer timer) {
+ timer.set(context.timestamp());
+ }
+
+ @OnTimer(TIMER_ID)
+ public void onTimer(OnTimerContext context) {
+ Instant outputTimestamp = context.timestamp().minus(outputTimestampSkew);
Review comment:
Done.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] codecov[bot] edited a comment on pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
codecov[bot] edited a comment on pull request #15540:
URL: https://github.com/apache/beam/pull/15540#issuecomment-953406612
# [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) Report
> Merging [#15540](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (7b0de72) into [master](https://codecov.io/gh/apache/beam/commit/163ac6a3c10c26898ad89ca8bedde8ef78ee7ee2?el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) (163ac6a) will **increase** coverage by `37.44%`.
> The diff coverage is `n/a`.
> :exclamation: Current head 7b0de72 differs from pull request most recent head 8954610. Consider uploading reports for the commit 8954610 to get more accurate results
[](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
```diff
@@ Coverage Diff @@
## master #15540 +/- ##
===========================================
+ Coverage 46.08% 83.53% +37.44%
===========================================
Files 195 445 +250
Lines 19457 61210 +41753
===========================================
+ Hits 8967 51131 +42164
- Misses 9521 10079 +558
+ Partials 969 0 -969
```
| [Impacted Files](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | Coverage Δ | |
|---|---|---|
| [sdks/go/pkg/beam/core/runtime/exec/pcollection.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvZXhlYy9wY29sbGVjdGlvbi5nbw==) | | |
| [...ks/go/pkg/beam/core/runtime/harness/cache\_hooks.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3J1bnRpbWUvaGFybmVzcy9jYWNoZV9ob29rcy5nbw==) | | |
| [sdks/go/pkg/beam/core/util/hooks/hooks.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3V0aWwvaG9va3MvaG9va3MuZ28=) | | |
| [sdks/go/pkg/beam/transforms/stats/sum.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS90cmFuc2Zvcm1zL3N0YXRzL3N1bS5nbw==) | | |
| [sdks/go/pkg/beam/core/util/protox/base64.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3V0aWwvcHJvdG94L2Jhc2U2NC5nbw==) | | |
| [sdks/go/pkg/beam/forward.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9mb3J3YXJkLmdv) | | |
| [...o/pkg/beam/runners/dataflow/dataflowlib/metrics.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9ydW5uZXJzL2RhdGFmbG93L2RhdGFmbG93bGliL21ldHJpY3MuZ28=) | | |
| [sdks/go/pkg/beam/core/util/reflectx/calls.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3V0aWwvcmVmbGVjdHgvY2FsbHMuZ28=) | | |
| [sdks/go/pkg/beam/core/util/symtab/symtab.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9jb3JlL3V0aWwvc3ltdGFiL3N5bXRhYi5nbw==) | | |
| [sdks/go/pkg/beam/encoding.go](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation#diff-c2Rrcy9nby9wa2cvYmVhbS9lbmNvZGluZy5nbw==) | | |
| ... and [630 more](https://codecov.io/gh/apache/beam/pull/15540/diff?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=continue&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=footer&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Last update [163ac6a...8954610](https://codecov.io/gh/apache/beam/pull/15540?src=pr&el=lastupdated&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=The+Apache+Software+Foundation).
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] laraschmidt commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
laraschmidt commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r728361761
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,13 +1195,24 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
- checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
- "output timestamp %s should be after input message timestamp or output timestamp of firing timers %s",
- outputTimestamp,
- elementInputTimestamp);
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elementInputTimestamp.minus(fn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (outputTimestamp.isBefore(lowerBound) || outputTimestamp.isAfter(upperBound)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "output timestamp %s (allowed skew %s) should be after input message timestamp or"
+ + " output timestamp of firing timers %s and before %s",
+ outputTimestamp,
+ PeriodFormat.getDefault().print(fn.getAllowedTimestampSkew().toPeriod()),
+ elementInputTimestamp,
+ upperBound));
Review comment:
Done.
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1897,14 +1898,26 @@ private Instant minTargetAndGcTime(Instant target) {
return Timer.cleared(userKey, dynamicTimerTag, Collections.singletonList(boundedWindow));
}
+ @SuppressWarnings("deprecation") // Allowed Skew is deprecated for users, but must be respected
private Timer<K> getTimerForTime(Instant scheduledTime) {
if (outputTimestamp != null) {
- checkArgument(
- !outputTimestamp.isBefore(elementTimestampOrTimerHoldTimestamp),
- "output timestamp %s should be after input message timestamp or output timestamp of"
- + " firing timers %s",
- outputTimestamp,
- elementTimestampOrTimerHoldTimestamp);
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elementTimestampOrTimerHoldTimestamp.minus(doFn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (outputTimestamp.isBefore(lowerBound) || outputTimestamp.isAfter(upperBound)) {
Review comment:
done
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1999,6 +2012,28 @@ public void set(String dynamicTimerTag, Instant absoluteTime) {
}
}
+ @SuppressWarnings("deprecation") // Allowed Skew is deprecated for users, but must be respected
+ private void checkTimestamp(Instant timestamp) {
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = currentElement.getTimestamp().minus(doFn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (timestamp.isBefore(lowerBound) || timestamp.isAfter(upperBound)) {
Review comment:
done
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1999,6 +2012,28 @@ public void set(String dynamicTimerTag, Instant absoluteTime) {
}
}
+ @SuppressWarnings("deprecation") // Allowed Skew is deprecated for users, but must be respected
+ private void checkTimestamp(Instant timestamp) {
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = currentElement.getTimestamp().minus(doFn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (timestamp.isBefore(lowerBound) || timestamp.isAfter(upperBound)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Cannot output with timestamp %s. Output timestamps must be no earlier than the "
+ + "timestamp of the current input (%s) minus the allowed skew (%s). See the "
Review comment:
fixed
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] lukecwik commented on a change in pull request #15540: [BEAM-12931] Allow for DoFn#getAllowedTimestampSkew() when checking the output timestamp
Posted by GitBox <gi...@apache.org>.
lukecwik commented on a change in pull request #15540:
URL: https://github.com/apache/beam/pull/15540#discussion_r727598137
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1897,14 +1898,26 @@ private Instant minTargetAndGcTime(Instant target) {
return Timer.cleared(userKey, dynamicTimerTag, Collections.singletonList(boundedWindow));
}
+ @SuppressWarnings("deprecation") // Allowed Skew is deprecated for users, but must be respected
private Timer<K> getTimerForTime(Instant scheduledTime) {
if (outputTimestamp != null) {
- checkArgument(
- !outputTimestamp.isBefore(elementTimestampOrTimerHoldTimestamp),
- "output timestamp %s should be after input message timestamp or output timestamp of"
- + " firing timers %s",
- outputTimestamp,
- elementTimestampOrTimerHoldTimestamp);
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elementTimestampOrTimerHoldTimestamp.minus(doFn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (outputTimestamp.isBefore(lowerBound) || outputTimestamp.isAfter(upperBound)) {
Review comment:
```suggestion
try {
lowerBound = elementTimestampOrTimerHoldTimestamp.minus(doFn.getAllowedTimestampSkew());
} catch (ArithmeticException e) {
lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
}
if (outputTimestamp.isBefore(lowerBound) || outputTimestamp.isAfter(BoundedWindow.TIMESTAMP_MAX_VALUE)) {
```
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1999,6 +2012,28 @@ public void set(String dynamicTimerTag, Instant absoluteTime) {
}
}
+ @SuppressWarnings("deprecation") // Allowed Skew is deprecated for users, but must be respected
+ private void checkTimestamp(Instant timestamp) {
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = currentElement.getTimestamp().minus(doFn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (timestamp.isBefore(lowerBound) || timestamp.isAfter(upperBound)) {
Review comment:
```suggestion
try {
lowerBound = currentElement.getTimestamp().minus(doFn.getAllowedTimestampSkew());
} catch (ArithmeticException e) {
lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
}
if (timestamp.isBefore(lowerBound) || timestamp.isAfter(BoundedWindow.TIMESTAMP_MAX_VALUE)) {
```
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,13 +1195,24 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
- checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
- "output timestamp %s should be after input message timestamp or output timestamp of firing timers %s",
- outputTimestamp,
- elementInputTimestamp);
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elementInputTimestamp.minus(fn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (outputTimestamp.isBefore(lowerBound) || outputTimestamp.isAfter(upperBound)) {
Review comment:
```suggestion
try {
lowerBound = elementInputTimestamp.minus(fn.getAllowedTimestampSkew());
} catch (ArithmeticException e) {
lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
}
if (outputTimestamp.isBefore(lowerBound) || outputTimestamp.isAfter(BoundedWindow.TIMESTAMP_MAX_VALUE)) {
```
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -386,6 +397,186 @@ public void testInfiniteSkew() {
BoundedWindow.TIMESTAMP_MAX_VALUE));
}
+ @Test
+ @Category({ValidatesRunner.class, UsesTimersInParDo.class})
+ public void testRunnerNoSkew() {
+ PCollection<Duration> input =
+ p.apply(
+ "create",
+ Create.timestamped(
+ Arrays.asList(new Duration(0L), new Duration(1L)), Arrays.asList(0L, 1L)));
+ input.apply(ParDo.of(new SkewingDoFn(Duration.ZERO)));
+ // The errors differ between runners but at least check that the output timestamp is printed.
+ thrown.expectMessage(String.format("%s", new Instant(0L)));
+ thrown.expectMessage("output");
+ p.run();
+ }
+
+ @Test
+ @Category({UsesTimersInParDo.class, ValidatesRunner.class})
+ public void testRunnerAllowedSkew() {
+ PCollection<Duration> input =
+ p.apply(
+ "create",
+ Create.timestamped(
+ Arrays.asList(new Duration(0L), new Duration(1L)), Arrays.asList(0L, 1L)));
+ input.apply(ParDo.of(new SkewingDoFn(new Duration(2L))));
+ p.run();
+ }
+
+ /**
+ * Demonstrates that attempting to set a timer with an output timestamp before the timestamp of
+ * the current element with zero {@link DoFn#getAllowedTimestampSkew() allowed timestamp skew}
+ * throws.
+ */
+ @Test
+ public void testTimerBackwardsInTimeNoSkew() {
+ TimerSkewingDoFn fn = new TimerSkewingDoFn(Duration.ZERO);
+ DoFnRunner<KV<String, Duration>, Duration> runner =
+ new SimpleDoFnRunner<>(
+ null,
+ fn,
+ NullSideInputReader.empty(),
+ new ListOutputManager(),
+ new TupleTag<>(),
+ Collections.emptyList(),
+ mockStepContext,
+ null,
+ Collections.emptyMap(),
+ WindowingStrategy.of(new GlobalWindows()),
+ DoFnSchemaInformation.create(),
+ Collections.emptyMap());
+
+ runner.startBundle();
+ // A timer with output timestamp at the current timestamp is fine.
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(KV.of("1", Duration.ZERO), new Instant(0)));
+
+ thrown.expect(UserCodeException.class);
+ thrown.expectCause(isA(IllegalArgumentException.class));
+ thrown.expectMessage("output timestamp of firing timers");
+ thrown.expectMessage(
+ String.format("output timestamp %s", new Instant(0).minus(Duration.millis(1L))));
+ thrown.expectMessage(
+ String.format(
+ "allowed skew %s", PeriodFormat.getDefault().print(Duration.ZERO.toPeriod())));
+
+ // A timer with output timestamp before (current time - skew) is forbidden
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of("2", Duration.millis(1L)), new Instant(0)));
+ }
+
+ /**
+ * Demonstrates that attempting to have a timer with output timestamp before the timestamp of the
+ * current element plus the value of {@link DoFn#getAllowedTimestampSkew()} throws, but between
+ * that value and the current timestamp succeeds.
+ */
+ @Test
+ public void testTimerSkew() {
+ TimerSkewingDoFn fn = new TimerSkewingDoFn(Duration.standardMinutes(10L));
+ DoFnRunner<KV<String, Duration>, Duration> runner =
+ new SimpleDoFnRunner<>(
+ null,
+ fn,
+ NullSideInputReader.empty(),
+ new ListOutputManager(),
+ new TupleTag<>(),
+ Collections.emptyList(),
+ mockStepContext,
+ null,
+ Collections.emptyMap(),
+ WindowingStrategy.of(new GlobalWindows()),
+ DoFnSchemaInformation.create(),
+ Collections.emptyMap());
+
+ runner.startBundle();
+ // Timer with output timestamp between "now" and "now - allowed skew" succeeds.
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of("1", Duration.standardMinutes(5L)), new Instant(0)));
+ thrown.expect(UserCodeException.class);
+ thrown.expectCause(isA(IllegalArgumentException.class));
+ thrown.expectMessage("output timestamp of firing timers");
+ thrown.expectMessage(
+ String.format("output timestamp %s", new Instant(0).minus(Duration.standardHours(1L))));
+ thrown.expectMessage(
+ String.format(
+ "allowed skew %s",
+ PeriodFormat.getDefault().print(Duration.standardMinutes(10L).toPeriod())));
+ // Timer with output timestamp before "now - allowed skew" fails.
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of("2", Duration.standardHours(1L)), new Instant(0)));
+ }
+
+ /**
+ * Demonstrates that attempting to output an element with a timestamp before the current one
+ * always succeeds when {@link DoFn#getAllowedTimestampSkew()} is equal to {@link Long#MAX_VALUE}
+ * milliseconds.
+ */
+ @Test
+ public void testTimerInfiniteSkew() {
+ TimerSkewingDoFn fn = new TimerSkewingDoFn(Duration.millis(Long.MAX_VALUE));
+ DoFnRunner<KV<String, Duration>, Duration> runner =
+ new SimpleDoFnRunner<>(
+ null,
+ fn,
+ NullSideInputReader.empty(),
+ new ListOutputManager(),
+ new TupleTag<>(),
+ Collections.emptyList(),
+ mockStepContext,
+ null,
+ Collections.emptyMap(),
+ WindowingStrategy.of(new GlobalWindows()),
+ DoFnSchemaInformation.create(),
+ Collections.emptyMap());
+
+ runner.startBundle();
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of("1", Duration.millis(1L)), new Instant(0)));
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of("2", Duration.millis(1L)),
+ BoundedWindow.TIMESTAMP_MIN_VALUE.plus(Duration.millis(1))));
+ runner.processElement(
+ WindowedValue.timestampedValueInGlobalWindow(
+ KV.of(
+ "3",
+ // This is the maximum amount a timestamp in beam can move (from the maximum
+ // timestamp
+ // to the minimum timestamp).
+ Duration.millis(BoundedWindow.TIMESTAMP_MAX_VALUE.getMillis())
+ .minus(Duration.millis(BoundedWindow.TIMESTAMP_MIN_VALUE.getMillis()))),
+ BoundedWindow.TIMESTAMP_MAX_VALUE));
+ }
+
+ @Test
+ @Category({UsesTimersInParDo.class, ValidatesRunner.class})
+ public void testRunnerTimerNoSkew() {
+ List<KV<String, Duration>> durations =
+ Arrays.asList(KV.of("0", new Duration(0L)), KV.of("2", new Duration(1L)));
+ PCollection<KV<String, Duration>> input =
+ p.apply("create", Create.timestamped(durations, Arrays.asList(0L, 2L)));
+ input.apply(ParDo.of(new TimerSkewingDoFn(Duration.ZERO)));
+ // The errors differ between runners but at least check that the output timestamp is printed.
+ thrown.expectMessage(String.format("%s", new Instant(1L)));
Review comment:
I don't think this will work for Dataflow streaming.
This passes on Jenkins because all of these tests don't run in both streaming and batch modes but will not work when we run tests inside Google.
##########
File path: sdks/java/harness/src/test/java/org/apache/beam/fn/harness/FnApiDoFnRunnerTest.java
##########
@@ -3878,6 +3885,182 @@ public void testProcessElementForWindowedTruncateAndSizeRestriction() throws Exc
}
}
+ @RunWith(JUnit4.class)
+ public static class ExceptionThrowingExecutionTest {
+ @Rule public final ExpectedException thrown = ExpectedException.none();
+
+ public static final String TEST_TRANSFORM_ID = "pTransformId";
+
+ /**
+ * A {@link DoFn} that outputs elements with timestamp equal to the input timestamp minus the
+ * input element.
+ */
+ private static class SkewingDoFn extends DoFn<String, String> {
+ private final Duration allowedSkew;
+
+ private SkewingDoFn(Duration allowedSkew) {
+ this.allowedSkew = allowedSkew;
+ }
+
+ @ProcessElement
+ public void processElement(ProcessContext context) {
+ Duration duration = new Duration(Long.valueOf(context.element()));
+ context.outputWithTimestamp(context.element(), context.timestamp().minus(duration));
+ }
+
+ @Override
+ public Duration getAllowedTimestampSkew() {
+ return allowedSkew;
+ }
+ }
+
+ @Test
+ public void testDoFnSkewNotAllowed() throws Exception {
+ Pipeline p = Pipeline.create();
+ PCollection<String> valuePCollection = p.apply(Create.of("0", "1"));
+ PCollection<String> outputPCollection =
+ valuePCollection.apply(TEST_TRANSFORM_ID, ParDo.of(new SkewingDoFn(Duration.ZERO)));
+
+ SdkComponents sdkComponents = SdkComponents.create(p.getOptions());
+ RunnerApi.Pipeline pProto = PipelineTranslation.toProto(p, sdkComponents);
+ String inputPCollectionId = sdkComponents.registerPCollection(valuePCollection);
+ String outputPCollectionId = sdkComponents.registerPCollection(outputPCollection);
+ RunnerApi.PTransform pTransform =
+ pProto
+ .getComponents()
+ .getTransformsOrThrow(
+ pProto
+ .getComponents()
+ .getTransformsOrThrow(TEST_TRANSFORM_ID)
+ .getSubtransforms(0));
+
+ List<WindowedValue<String>> mainOutputValues = new ArrayList<>();
+ MetricsContainerStepMap metricsContainerRegistry = new MetricsContainerStepMap();
+ PCollectionConsumerRegistry consumers =
+ new PCollectionConsumerRegistry(
+ metricsContainerRegistry, mock(ExecutionStateTracker.class));
+
+ consumers.register(
+ outputPCollectionId,
+ TEST_TRANSFORM_ID,
+ (FnDataReceiver) (FnDataReceiver<WindowedValue<String>>) mainOutputValues::add,
+ StringUtf8Coder.of());
+ PTransformFunctionRegistry startFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "start");
+ PTransformFunctionRegistry finishFunctionRegistry =
+ new PTransformFunctionRegistry(
+ mock(MetricsContainerStepMap.class), mock(ExecutionStateTracker.class), "finish");
+ List<ThrowingRunnable> teardownFunctions = new ArrayList<>();
+
+ new FnApiDoFnRunner.Factory<>()
+ .createRunnerForPTransform(
+ PipelineOptionsFactory.create(),
+ null /* beamFnDataClient */,
+ null /* beamFnStateClient */,
+ null /* beamFnTimerClient */,
+ TEST_TRANSFORM_ID,
+ pTransform,
+ Suppliers.ofInstance("57L")::get,
+ pProto.getComponents().getPcollectionsMap(),
+ pProto.getComponents().getCodersMap(),
+ pProto.getComponents().getWindowingStrategiesMap(),
+ consumers,
+ startFunctionRegistry,
+ finishFunctionRegistry,
+ null /* addResetFunction */,
+ teardownFunctions::add,
+ null /* addProgressRequestCallback */,
+ null /* splitListener */,
+ null /* bundleFinalizer */);
+
+ thrown.expect(UserCodeException.class);
+ thrown.expectMessage(String.format("timestamp %s", new Instant(0).minus(new Duration(1L))));
+ thrown.expectMessage(
+ String.format(
+ "allowed skew (%s)", PeriodFormat.getDefault().print(Duration.ZERO.toPeriod())));
+
+ Iterables.getOnlyElement(startFunctionRegistry.getFunctions()).run();
+ mainOutputValues.clear();
+
+ FnDataReceiver<WindowedValue<?>> mainInput =
+ consumers.getMultiplexingConsumer(inputPCollectionId);
+ mainInput.accept(valueInGlobalWindow("0"));
+ mainInput.accept(timestampedValueInGlobalWindow("1", new Instant(0L)));
Review comment:
Please swap to use `assertThrows` since `thrown.expect` is deprecated because it doesn't do a good job at showing which line of code is causing the exception.
##########
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:
It would be nice to be able to test the onTimer variant as well and not just when timers are set within processElement.
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -386,6 +399,180 @@ public void testInfiniteSkew() {
BoundedWindow.TIMESTAMP_MAX_VALUE));
}
+ @Test
+ @Category({ValidatesRunner.class})
+ public void testRunnerNoSkew() {
+ PCollection<Duration> input =
+ p.apply("create", Create.timestamped(Arrays.asList(new Duration(0L), new Duration(1L)), Arrays.asList(0L, 1L)));
+ input.apply(ParDo.of(new SkewingDoFn(Duration.ZERO)));
+ // The errors differ between runners but at least check that the output timestamp is printed.
+ thrown.expectMessage(String.format("%s", new Instant(0L)));
Review comment:
The issue is that `thrown.expect` is going to rely on Pipeline.run throwing the underlying cause for the job failure. I don't think all runners properly propagate the error (e.g. Dataflow streaming) that caused the pipeline to fail. The other failure tests in `ParDoTest.java` use `NeedsRunner` which run with the DirectRunner and not all runners. So if you want to only test DirectRunner mark as `NeedsRunner` instead of `ValidatesRunner`
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -437,19 +437,24 @@ public Instant timestamp() {
@SuppressWarnings("deprecation") // Allowed Skew is deprecated for users, but must be respected
private void checkTimestamp(Instant timestamp) {
- // The documentation of getAllowedTimestampSkew explicitly permits Long.MAX_VALUE to be used
- // for infinite skew. Defend against underflow in that case for timestamps before the epoch
- if (fn.getAllowedTimestampSkew().getMillis() != Long.MAX_VALUE
- && timestamp.isBefore(elem.getTimestamp().minus(fn.getAllowedTimestampSkew()))) {
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elem.getTimestamp().minus(fn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (timestamp.isBefore(lowerBound) || timestamp.isAfter(upperBound)) {
Review comment:
```suggestion
try {
lowerBound = elem.getTimestamp().minus(fn.getAllowedTimestampSkew());
} catch (ArithmeticException e) {
lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
}
if (timestamp.isBefore(lowerBound) || timestamp.isAfter(BoundedWindow.TIMESTAMP_MAX_VALUE)) {
```
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -437,19 +437,24 @@ public Instant timestamp() {
@SuppressWarnings("deprecation") // Allowed Skew is deprecated for users, but must be respected
private void checkTimestamp(Instant timestamp) {
- // The documentation of getAllowedTimestampSkew explicitly permits Long.MAX_VALUE to be used
- // for infinite skew. Defend against underflow in that case for timestamps before the epoch
- if (fn.getAllowedTimestampSkew().getMillis() != Long.MAX_VALUE
- && timestamp.isBefore(elem.getTimestamp().minus(fn.getAllowedTimestampSkew()))) {
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elem.getTimestamp().minus(fn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (timestamp.isBefore(lowerBound) || timestamp.isAfter(upperBound)) {
throw new IllegalArgumentException(
String.format(
"Cannot output with timestamp %s. Output timestamps must be no earlier than the "
- + "timestamp of the current input (%s) minus the allowed skew (%s). See the "
- + "DoFn#getAllowedTimestampSkew() Javadoc for details on changing the allowed "
- + "skew.",
+ + "timestamp of the current input (%s) minus the allowed skew (%s) and no "
+ + "later than %s. See the DoFn#getAllowedTimestampSkew() Javadoc for details "
+ + "on changing the allowed skew.",
timestamp,
elem.getTimestamp(),
- PeriodFormat.getDefault().print(fn.getAllowedTimestampSkew().toPeriod())));
+ PeriodFormat.getDefault().print(fn.getAllowedTimestampSkew().toPeriod()),
+ upperBound));
Review comment:
```suggestion
BoundedWindow.TIMESTAMP_MAX_VALUE));
```
##########
File path: runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java
##########
@@ -1190,13 +1195,24 @@ public Timer withOutputTimestamp(Instant outputTimestamp) {
* </ul>
*/
private void setAndVerifyOutputTimestamp() {
-
if (outputTimestamp != null) {
- checkArgument(
- !outputTimestamp.isBefore(elementInputTimestamp),
- "output timestamp %s should be after input message timestamp or output timestamp of firing timers %s",
- outputTimestamp,
- elementInputTimestamp);
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = elementInputTimestamp.minus(fn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (outputTimestamp.isBefore(lowerBound) || outputTimestamp.isAfter(upperBound)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "output timestamp %s (allowed skew %s) should be after input message timestamp or"
+ + " output timestamp of firing timers %s and before %s",
+ outputTimestamp,
+ PeriodFormat.getDefault().print(fn.getAllowedTimestampSkew().toPeriod()),
+ elementInputTimestamp,
+ upperBound));
Review comment:
The previous message isn't great. Can we improve this to be the similar to the other message like:
```
Cannot output timer with timestamp %s. Output timestamps must be no earlier than the timestamp of the current input (%s) minus the allowed skew (%s) and no later than %s. See the DoFn#getAllowedTimestampSkew() Javadoc for details on changing the allowed skew.
```
Ditto here and in `FnApiDoFnRunner.java`
##########
File path: runners/core-java/src/test/java/org/apache/beam/runners/core/SimpleDoFnRunnerTest.java
##########
@@ -386,6 +397,186 @@ public void testInfiniteSkew() {
BoundedWindow.TIMESTAMP_MAX_VALUE));
}
+ @Test
+ @Category({ValidatesRunner.class, UsesTimersInParDo.class})
+ public void testRunnerNoSkew() {
Review comment:
Please move the `ValidatesRunner` tests to `ParDoTest.java` since that is where the bulk of the other ParDo specific `ValidatesRunner` and `NeedsRunner` test exist.
##########
File path: sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java
##########
@@ -1999,6 +2012,28 @@ public void set(String dynamicTimerTag, Instant absoluteTime) {
}
}
+ @SuppressWarnings("deprecation") // Allowed Skew is deprecated for users, but must be respected
+ private void checkTimestamp(Instant timestamp) {
+ Instant lowerBound;
+ Instant upperBound = BoundedWindow.TIMESTAMP_MAX_VALUE;
+ try {
+ lowerBound = currentElement.getTimestamp().minus(doFn.getAllowedTimestampSkew());
+ } catch (ArithmeticException e) {
+ lowerBound = BoundedWindow.TIMESTAMP_MIN_VALUE;
+ }
+ if (timestamp.isBefore(lowerBound) || timestamp.isAfter(upperBound)) {
+ throw new IllegalArgumentException(
+ String.format(
+ "Cannot output with timestamp %s. Output timestamps must be no earlier than the "
+ + "timestamp of the current input (%s) minus the allowed skew (%s). See the "
Review comment:
This message differs from the one in SimpleDoFnRunner as it is missing the upper bound.
--
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: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org