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 2020/07/01 17:19:27 UTC
[GitHub] [beam] je-ik opened a new pull request #12152: [WIP] Fix direct runner race during watermark update
je-ik opened a new pull request #12152:
URL: https://github.com/apache/beam/pull/12152
Fix race condition when updating downstream watermarks in DirectRunner
------------------------
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).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Dataflow | Flink | Samza | Spark | Twister2
--- | --- | --- | --- | --- | --- | ---
Go | [](https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) | ---
Java | [](https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/) | [](https://builds.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | [](https://ci-beam.apache.org/job/beam_PostCommit_Python2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python35/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python2_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python35_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/) | ---
XLang | --- | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/) | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website
--- | --- | --- | --- | ---
Non-portable | [](https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/)
Portable | --- | [](https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | ---
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.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] je-ik commented on a change in pull request #12152: [WIP] Fix direct runner race during watermark update
Posted by GitBox <gi...@apache.org>.
je-ik commented on a change in pull request #12152:
URL: https://github.com/apache/beam/pull/12152#discussion_r448511608
##########
File path: runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
##########
@@ -1222,6 +1209,7 @@ public synchronized void refreshAll() {
while (!toRefresh.isEmpty()) {
toRefresh = refreshAllOf(toRefresh);
}
+ pendingRefreshes.clear();
Review comment:
This was not cleared ever before.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] je-ik commented on pull request #12152: [WIP] Fix direct runner race during watermark update
Posted by GitBox <gi...@apache.org>.
je-ik commented on pull request #12152:
URL: https://github.com/apache/beam/pull/12152#issuecomment-652605959
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] je-ik commented on a change in pull request #12152: [WIP] Fix direct runner race during watermark update
Posted by GitBox <gi...@apache.org>.
je-ik commented on a change in pull request #12152:
URL: https://github.com/apache/beam/pull/12152#discussion_r448509205
##########
File path: runners/direct-java/src/main/java/org/apache/beam/runners/direct/QuiescenceDriver.java
##########
@@ -70,6 +71,8 @@ public static ExecutionDriver create(
private final Map<AppliedPTransform<?, ?, ?>, ConcurrentLinkedQueue<CommittedBundle<?>>>
pendingRootBundles;
private final Queue<WorkUpdate> pendingWork = new ConcurrentLinkedQueue<>();
+ private final Map<AppliedPTransform<?, ?, ?>, Collection<CommittedBundle<?>>> inflightBundles =
+ new ConcurrentHashMap<>();
Review comment:
We collect here bundles and AppliedPTransforms that have started to process bundle, but have not completed it yet. The reason for that is that the bundle processing *might* change output watermark of a PTransform *before* enqueuing the resulting bundle to pendingUpdates of downstream PTransform, which can lead to watermark being updated past the emitted elements.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] je-ik commented on a change in pull request #12152: [WIP] Fix direct runner race during watermark update
Posted by GitBox <gi...@apache.org>.
je-ik commented on a change in pull request #12152:
URL: https://github.com/apache/beam/pull/12152#discussion_r448509888
##########
File path: runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java
##########
@@ -128,19 +108,6 @@ public void cleanup() throws Exception {
final DoFn<KV<K, InputT>, OutputT> doFn = application.getTransform().getDoFn();
final DoFnSignature signature = DoFnSignatures.getSignature(doFn.getClass());
- // If the DoFn is stateful, schedule state clearing.
- // It is semantically correct to schedule any number of redundant clear tasks; the
- // cache is used to limit the number of tasks to avoid performance degradation.
- if (signature.stateDeclarations().size() > 0) {
- for (final WindowedValue<?> element : inputBundle.getElements()) {
- for (final BoundedWindow window : element.getWindows()) {
- cleanupRegistry.get(
- AppliedPTransformOutputKeyAndWindow.create(
- application, (StructuralKey<K>) inputBundle.getKey(), window));
- }
- }
- }
-
Review comment:
This is removed because it is somewhat duplicate to what StatefulDoFnRunner already does. We can return it back, it is not directly related to purpose of this PR, it is just a cleanup that I did during bug hunting.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] je-ik commented on a change in pull request #12152: [WIP] Fix direct runner race during watermark update
Posted by GitBox <gi...@apache.org>.
je-ik commented on a change in pull request #12152:
URL: https://github.com/apache/beam/pull/12152#discussion_r448511263
##########
File path: runners/direct-java/src/main/java/org/apache/beam/runners/direct/WatermarkManager.java
##########
@@ -1110,17 +1109,7 @@ public void updateWatermarks(
pendingUpdates.offer(
PendingWatermarkUpdate.create(
executable, completed, timerUpdate, unprocessedInputs, outputs, earliestHold));
- tryApplyPendingUpdates();
- }
-
- private void tryApplyPendingUpdates() {
- if (refreshLock.tryLock()) {
- try {
- applyNUpdates(MAX_INCREMENTAL_UPDATES);
- } finally {
- refreshLock.unlock();
- }
- }
Review comment:
Not sure what was the purpose for limiting the number of applied updates, but that seems to be wrong - we need to apply all updates at all times, because otherwise we again might not hold the downstream watermark back enough.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] je-ik closed pull request #12152: Fix direct runner race during watermark update
Posted by GitBox <gi...@apache.org>.
je-ik closed pull request #12152:
URL: https://github.com/apache/beam/pull/12152
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] je-ik commented on a change in pull request #12152: [WIP] Fix direct runner race during watermark update
Posted by GitBox <gi...@apache.org>.
je-ik commented on a change in pull request #12152:
URL: https://github.com/apache/beam/pull/12152#discussion_r448510564
##########
File path: runners/direct-java/src/main/java/org/apache/beam/runners/direct/StatefulParDoEvaluatorFactory.java
##########
@@ -270,11 +177,13 @@ public void processElement(WindowedValue<KeyedWorkItem<K, KV<K, InputT>>> gbkRes
delegateEvaluator.processElement(windowedValue);
}
- Instant currentInputWatermark = timerInternals.currentInputWatermarkTime();
+ final Instant inputWatermarkTime = timerInternals.currentInputWatermarkTime();
PriorityQueue<TimerData> toBeFiredTimers =
new PriorityQueue<>(Comparator.comparing(TimerData::getTimestamp));
gbkResult.getValue().timersIterable().forEach(toBeFiredTimers::add);
- while (!toBeFiredTimers.isEmpty()) {
+
+ while (!timerInternals.containsUpdateForTimeBefore(inputWatermarkTime)
+ && !toBeFiredTimers.isEmpty()) {
Review comment:
It can happen that elements processed inside this bundle set timer, so it is technically better to check this before firing any timer. In practice it probably doesn't matter, because DirectRunner currently apparently doesn't fix bundles with timers and bundles with elements.
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org