Fix Reshuffle implementation in Java SDK #28853

kennknowles · 2023-10-05T18:54:21Z

This fixes the Reshuffle default implementation to preserve all metadata and be a true no-op, whereas before it did not preserve the pane info.

Regarding update compatibility, my understanding of runners that have long-lived streaming jobs and perform updates, as of this writing:

Flink: overrides Reshuffle
Samza: overrides Reshuffle
Dataflow v1: overrides Reshuffle
Dataflow v2: executes default implementation

I went ahead and used the update compatibility flag anyhow just in case, so it can be reverted to exactly the old implementation on demand.

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

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

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

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

kennknowles · 2023-10-19T01:16:03Z

R: @robertwb

github-actions · 2023-10-19T01:17:32Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control

kennknowles · 2023-10-19T11:20:26Z

Failure was :runners:spark:3:test https://ci-beam.apache.org/job/beam_PreCommit_Java_Commit/28710/testReport/junit/org.apache.beam.runners.spark.translation.streaming/ResumeFromCheckpointStreamingTest/testWithResume/

kennknowles · 2023-10-19T11:20:31Z

run java precommit

kennknowles · 2023-10-19T11:22:19Z

The failure in the AWS2 tests was the same Spark resume from checkpoint test. It shouldn't even have been run in that suite. It looks like our invocations are a bit messed up (@damccorm FYI but I'm assuming this is just the same as it was in Jenkins).

kennknowles · 2023-10-19T15:44:07Z

run java precommit

kennknowles · 2023-10-19T15:44:52Z

I get the impression this somehow actually slows things down. Everything looks like a timeout. Or perhaps it just invalidates too much of the Gradle cache?

kennknowles · 2023-10-19T18:28:50Z

https://ge.apache.org/s/s6jldnngfhe32/tests/overview?outcome=FAILED#:runners:java-fn-execution:test-5-org.apache.beam.runners.fnexecution.control.remoteexecutiontest-11-testbundleprocessorthrowsexecutionexceptionwhenusercodethrows-44

I haven't figured out if this is flaky or environmental. I can't repro it locally.

kennknowles · 2023-10-19T18:28:57Z

run java precommit

kennknowles · 2023-10-20T14:05:13Z

@robertwb finally green - I don't think this accomplishes everything from the dev thread, just the SDK implementation. It does not impact runners that override or specially translate reshuffle.

kennknowles · 2023-10-20T14:05:20Z

run dataflow validatesrunner

kennknowles · 2023-10-20T14:05:25Z

run flink validatesrunner

kennknowles · 2023-10-20T14:05:37Z

run spark validatesrunner

kennknowles · 2023-10-20T14:05:52Z

run samza validatesrunner

robertwb · 2023-10-20T22:33:16Z

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Reshuffle.java

+              new DoFn<KV<K, ValueInSingleWindow<V>>, KV<K, V>>() {
+                @Override
+                public Duration getAllowedTimestampSkew() {
+                  return Duration.millis(Long.MAX_VALUE);


The runner doesn't read/act on this value anywhere, does it?

Noting that this is not a change but a move. But no the runner does not act on it. In fact it is unsafe for this reason - the runner may well drop data that you have allowed to be output.

robertwb · 2023-10-20T22:39:19Z

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Reshuffle.java

+        .apply(
+            "RestoreOriginalWindows",
+            Window.into(new RestoreWindowsFn<>(originalStrategy.getWindowFn())))
+        .apply("RestoreOriginalTimestamps", new RestoreTimestamps<>())


What about the PaneInfo information?

Ha of course. The whole point.

I got caught up in the fact that I could not restore windows and timestamps in a single ParDo and had to do a rewindowing. But in fact I don't think there is a way to restore the paneinfo without changing the model after all to at least allow direct manipulation of it.

robertwb · 2023-10-20T22:39:53Z

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/Reshuffle.java

+    PCollection<KV<K, ValueInSingleWindow<V>>> reified =
+        input
+            .apply("SetIdentityWindow", rewindow)
+            .apply("ReifyOriginalMetadata", Reify.windowsInValue());


Won't this break backwards compatibility for everything using a shuffle (though I concede it's fixing a bug)?

I had that same concern and was coming down on the side of "we should change this anyhow". BUT looking into it, every single runner directly implements reshuffle. So this is essentially just a fix for the reference implementation and the default for runners that don't implement it yet (like Dataflow v2).

So I actually need a ValidatesRunner test which I presume most runner will fail. But we don't have to worry about update incompatibility unless/until those runners come into compliance. I still think this change should happen.

Could we test that this does not break compatibility (even if the test is a one-off manual test)?

kennknowles · 2023-11-13T20:33:29Z

OK, I added the capability to outputWindowedValue to the SDK (somewhat naively IDE-guided refactor) and used it in Create in order to write a basic test of what I think we want.

kennknowles · 2023-11-14T13:38:30Z

Gotta go through and poke a few more files

robertwb · 2023-12-05T00:26:44Z

What's the status on this?

kennknowles · 2023-12-05T01:29:28Z

java precommit timed out - perhaps due to cache invalidation due to modifying core SDK

johnjcasey · 2023-12-05T16:03:18Z

...nstruction-java/src/main/java/org/apache/beam/runners/core/construction/SplittableParDo.java

@@ -669,6 +670,15 @@ public void output(RestrictionT part) {
                    public void outputWithTimestamp(RestrictionT part, Instant timestamp) {
                      throw new UnsupportedOperationException();
                    }
+
+                    @Override
+                    public void outputWindowedValue(


It might be easier to make this outputWindowedValue a default implementation for OutputReceiver to simplify this change for all the test implementations of OutputReceiver.

kennknowles · 2023-12-14T17:13:18Z

The outputWindowedValue change has been merged independently, shrinking this one. I'll check it over to make sure the residual change makes sense and is adequately complete.

kennknowles · 2023-12-18T18:52:03Z

Failure is the error on master fixed by #29798 (using \n instead of %n in a format string)

kennknowles · 2023-12-19T17:44:10Z

Noting that this was green but I went ahead and added the old expansion as an option using the new update compat flag. I am going to use the flag to test compatibility on Dataflow v1. We expect the override makes this change irrelevant, so we don't want to actually bug users. And on v2 we want the incompatibility.

kennknowles · 2023-12-22T17:51:43Z

Added internal Dataflow test that reloads the pipeline with changing the compatibility flag.

Abacn · 2024-01-10T16:12:42Z

It seems the newly added test testReshufflePreservesMetadata not working for all runners:

beam_PostCommit_Java_PVR_Samza: https://github.com/apache/beam/actions/runs/7399747764/job/20131866953

beam_PostCommit_Java_PVR_Spark3_Streaming: https://github.com/apache/beam/actions/runs/7399737943/job/20131836389

beam_PostCommit_Java_PVR_Spark3_Batch: https://github.com/apache/beam/runs/20340841429

beam_PostCommit_Java_ValidatesRunner_ULR also

github-actions bot added the java label Oct 5, 2023

kennknowles force-pushed the reshuffle branch from c19a102 to 5d3b54c Compare October 6, 2023 22:21

kennknowles force-pushed the reshuffle branch 3 times, most recently from 24360e1 to b7c954b Compare October 18, 2023 22:48

github-actions bot added runners samza labels Oct 18, 2023

kennknowles force-pushed the reshuffle branch from b7c954b to c474e53 Compare October 19, 2023 00:39

kennknowles marked this pull request as ready for review October 19, 2023 01:15

robertwb reviewed Oct 20, 2023

View reviewed changes

kennknowles force-pushed the reshuffle branch from c474e53 to 31a4714 Compare November 1, 2023 17:02

github-actions bot added the core label Nov 1, 2023

kennknowles force-pushed the reshuffle branch from 31a4714 to 90ac268 Compare November 13, 2023 20:26

kennknowles force-pushed the reshuffle branch from 90ac268 to c23cdaa Compare November 29, 2023 16:30

github-actions bot added the flink label Nov 30, 2023

kennknowles force-pushed the reshuffle branch 2 times, most recently from 2dfa6a0 to 657b7e4 Compare December 4, 2023 20:25

kennknowles mentioned this pull request Dec 5, 2023

Add outputWindowedValue capability to Java SDK #29616

Merged

3 tasks

johnjcasey reviewed Dec 5, 2023

View reviewed changes

kennknowles force-pushed the reshuffle branch from 657b7e4 to adead2f Compare December 14, 2023 17:12

github-actions bot removed io extensions sql gcp kafka core spanner labels Dec 14, 2023

kennknowles force-pushed the reshuffle branch 2 times, most recently from 7817665 to 7a214a0 Compare December 18, 2023 16:34

kennknowles force-pushed the reshuffle branch from 7a214a0 to ff6d4ba Compare December 18, 2023 19:23

Fix Reshuffle implementation in Java SDK

7827d1d

kennknowles force-pushed the reshuffle branch from ff6d4ba to 7827d1d Compare December 19, 2023 17:35

kennknowles merged commit 8bb6610 into apache:master Jan 3, 2024
33 checks passed

kennknowles deleted the reshuffle branch January 3, 2024 15:41

Abacn mentioned this pull request Jan 10, 2024

[Failing Test]: testReshufflePreservesMetadata not working for some runners #29973

Closed

16 tasks

kennknowles mentioned this pull request Feb 28, 2024

[Bug]: BigQuery IO Batch load using File_load causing the same job id ignoring inserts as the job_id is already completed #28219

Closed

15 tasks

Abacn mentioned this pull request Apr 25, 2024

[Bug]: Reshuffle.viaRandomKey timeout since version 2.54.0 #31095

Open

16 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Fix Reshuffle implementation in Java SDK #28853

Fix Reshuffle implementation in Java SDK #28853

kennknowles commented Oct 5, 2023 •

edited

kennknowles commented Oct 19, 2023

github-actions bot commented Oct 19, 2023

kennknowles commented Oct 19, 2023

kennknowles commented Oct 19, 2023

kennknowles commented Oct 19, 2023

kennknowles commented Oct 19, 2023

kennknowles commented Oct 19, 2023

kennknowles commented Oct 19, 2023

kennknowles commented Oct 19, 2023

kennknowles commented Oct 20, 2023

kennknowles commented Oct 20, 2023

kennknowles commented Oct 20, 2023

kennknowles commented Oct 20, 2023

kennknowles commented Oct 20, 2023

robertwb Oct 20, 2023

kennknowles Oct 23, 2023

robertwb Oct 20, 2023

kennknowles Oct 23, 2023

robertwb Oct 20, 2023

kennknowles Oct 23, 2023

reuvenlax Dec 5, 2023

kennknowles commented Nov 13, 2023

kennknowles commented Nov 14, 2023

robertwb commented Dec 5, 2023

kennknowles commented Dec 5, 2023

johnjcasey Dec 5, 2023

kennknowles commented Dec 14, 2023

kennknowles commented Dec 18, 2023

kennknowles commented Dec 19, 2023

kennknowles commented Dec 22, 2023

Abacn commented Jan 10, 2024

Fix Reshuffle implementation in Java SDK #28853

Fix Reshuffle implementation in Java SDK #28853

Conversation

kennknowles commented Oct 5, 2023 • edited

GitHub Actions Tests Status (on master branch)

kennknowles commented Oct 19, 2023

github-actions bot commented Oct 19, 2023

kennknowles commented Oct 19, 2023

kennknowles commented Oct 19, 2023

kennknowles commented Oct 19, 2023

kennknowles commented Oct 19, 2023

kennknowles commented Oct 19, 2023

kennknowles commented Oct 19, 2023

kennknowles commented Oct 19, 2023

kennknowles commented Oct 20, 2023

kennknowles commented Oct 20, 2023

kennknowles commented Oct 20, 2023

kennknowles commented Oct 20, 2023

kennknowles commented Oct 20, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kennknowles commented Nov 13, 2023

kennknowles commented Nov 14, 2023

robertwb commented Dec 5, 2023

kennknowles commented Dec 5, 2023

Choose a reason for hiding this comment

kennknowles commented Dec 14, 2023

kennknowles commented Dec 18, 2023

kennknowles commented Dec 19, 2023

kennknowles commented Dec 22, 2023

Abacn commented Jan 10, 2024

kennknowles commented Oct 5, 2023 •

edited