Introduce OutputBuilder in Java SDK by kennknowles · Pull Request #34902 · apache/beam

kennknowles · 2025-05-09T15:41:29Z

With the addition of extended metadata for elements, we need to make code - especially user code - more robust to added metadata fields. This pull request adds OutputBuilder which should be used to builds new values, and builders should always be obtained by pulling metadata from some context (for example a currently-in-process element) when possible.

Concretely, this pull request

Adds OutputBuilder as a public interface extending WindowedValue.
Adds WindowedValues.builder(WindowedValueReceiver) to produce an OutputBuilder that will output to the provider receiver.

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.

sdks/java/core/src/main/java/org/apache/beam/sdk/util/WindowedValue.java

...va/core/src/main/java/org/apache/beam/sdk/util/construction/SplittableParDoNaiveBounded.java

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

derrickaw · 2025-08-12T14:20:40Z

waiting on author

kennknowles · 2025-09-12T17:16:32Z

R: @stankiewicz

github-actions · 2025-09-12T17:17:36Z

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers

kennknowles · 2025-09-12T17:18:44Z

I think Radek you have the most context, and can critique the details where I have missed something. Especially in FnApiDoFnRunner I see a couple spots where I think I must have gotten impatient - wherever I decompose a WindowedValue and then it gets put back together, we will lose metadata (once it is introduced).

I realize I also need some end-user tests of the OutputBuilder. It is very very thoroughly tested by how it is used internal to the SDK but there are no examples of use in a DoFn in an actual pipeline.

kennknowles · 2025-09-12T17:19:14Z

I also need to check that this does not create noticeable allocation/GC overhead by the new style.

stankiewicz · 2025-09-18T17:44:37Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java

        PaneInfo paneInfo) {
-      checkTimestamp(timestamp);
-      outputTo(mainOutputConsumer, WindowedValues.of(output, timestamp, windows, paneInfo));
+      builder(output).setTimestamp(timestamp).setWindows(windows).setPaneInfo(paneInfo).output();


I just like it so much!

stankiewicz · 2025-09-18T17:59:58Z

sdks/java/core/src/main/java/org/apache/beam/sdk/values/WindowedValues.java

+
+  /** Create a Builder that takes element metadata from the provideed delegate. */
+  public static <T> Builder<T> builder(WindowedValue<T> template) {
+    return new Builder<T>()


thi is great, so if we start to expand with new fields, this is the place to propagate them.

stankiewicz

read all of it, this will massively help extend WindowedValue with new fields while avoiding the journey of extending each output method. thanks!

kennknowles · 2025-09-18T18:57:25Z

Thank you!

OK, now it looks like somehow my attempts to make the distroless integration tests use the HEAD container caused actually maybe other tests to also not use the HEAD container. I know that internal tests are all green but I will see which of these tests are actually broken versus not configured to use the right container.

kennknowles · 2025-09-23T14:28:47Z

OK! All green and I am going to merge.

kellen · 2025-11-14T14:48:23Z

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java

+          .withValue(value)
+          .setReceiver(
+              windowedValue -> {
+                checkTimestamp(windowedValue.getTimestamp());


Hey @kennknowles Can you explain the reasoning about checking the timestamp here? In Scio we have some DoFns that propagate the element timestamp and this additional check breaks their current behavior. Not sure if we've been doing the wrong thing or not

My intention was only to maintain functionality. I am sure that I debugged a failing test around "invalid timestamps are correctly rejected" but I cannot recall exactly which one it is. I'm re-running some suites to find it and I'll report back here. I definitely don't want any currently-correct code to start failing!

Thanks! For context, one of the failing tests is for our AsyncDoFn, where we process elements, spinning off a set of futures while waiting for their respective responses, awaiting on those responses, and outputting those elements back to their original windows with their original timestamps.

The corresponding failing test is here, giving the following exception:

[info] - should propagate element metadata *** FAILED *** (45 milliseconds) [info] org.apache.beam.sdk.Pipeline$PipelineExecutionException: java.lang.IllegalArgumentException: Cannot output with timestamp 1970-01-01T00:00:00.001Z. Output timestamps must be no earlier than the timestamp of the current input or timer (1970-01-01T00:00:00.008Z) minus the allowed skew (0 milliseconds) and no later than 294247-01-10T04:00:54.775Z. See the DoFn#getAllowedTimestampSkew() Javadoc for details on changing the allowed skew. [info] at org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFinish(DirectRunner.java:377) [info] at org.apache.beam.runners.direct.DirectRunner$DirectPipelineResult.waitUntilFinish(DirectRunner.java:345) [info] at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:218) [info] at org.apache.beam.runners.direct.DirectRunner.run(DirectRunner.java:67) [info] at org.apache.beam.sdk.Pipeline.run(Pipeline.java:325) [info] at org.apache.beam.sdk.testing.TestPipeline.run(TestPipeline.java:442) [info] at org.apache.beam.sdk.testing.TestPipeline.run(TestPipeline.java:381) [info] at com.spotify.scio.ScioContext.execute(ScioContext.scala:671) [info] at com.spotify.scio.ScioContext$$anonfun$run$1.apply(ScioContext.scala:658) [info] at com.spotify.scio.ScioContext$$anonfun$run$1.apply(ScioContext.scala:646) [info] ... [info] Cause: java.lang.IllegalArgumentException: Cannot output with timestamp 1970-01-01T00:00:00.001Z. Output timestamps must be no earlier than the timestamp of the current input or timer (1970-01-01T00:00:00.008Z) minus the allowed skew (0 milliseconds) and no later than 294247-01-10T04:00:54.775Z. See the DoFn#getAllowedTimestampSkew() Javadoc for details on changing the allowed skew. [info] at org.apache.beam.runners.core.SimpleDoFnRunner.checkTimestamp(SimpleDoFnRunner.java:263) [info] at org.apache.beam.runners.core.SimpleDoFnRunner.access$1300(SimpleDoFnRunner.java:89) [info] at org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.lambda$outputWindowedValue$0(SimpleDoFnRunner.java:462) [info] at org.apache.beam.sdk.values.WindowedValues$Builder.output(WindowedValues.java:210) [info] at org.apache.beam.runners.core.SimpleDoFnRunner$DoFnProcessContext.outputWindowedValue(SimpleDoFnRunner.java:465) [info] at org.apache.beam.sdk.transforms.DoFnOutputReceivers$WindowedContextOutputReceiver.output(DoFnOutputReceivers.java:123) [info] at org.apache.beam.sdk.values.WindowedValues$Builder.output(WindowedValues.java:210) [info] at org.apache.beam.sdk.transforms.DoFn$OutputReceiver.outputWindowedValue(DoFn.java:416) [info] at com.spotify.scio.transforms.BaseAsyncLookupDoFn.lambda$processElement$0(BaseAsyncLookupDoFn.java:190) [info] at com.spotify.scio.transforms.BaseAsyncLookupDoFn.flush(BaseAsyncLookupDoFn.java:310) [info] ...

Interesting. This is actually a red herring. The error is not coming from this checkTimestamp. This is the portable / Dataflow v2 path.

The SimpleDoFnRunner is the v1 codepath and I see it now:

The failure is thrown from https://github.com/apache/beam/pull/34902/files/2cff4cc48abb7be6e7f91d3ab76dd6235c87938e#diff-560313cd06556c758d162abea624cab2646038cdabf483cdad69d392bf720a0bL252

The added check is in this block https://github.com/apache/beam/pull/34902/files/2cff4cc48abb7be6e7f91d3ab76dd6235c87938e#diff-560313cd06556c758d162abea624cab2646038cdabf483cdad69d392bf720a0bR502

I've updated #36822 to remove that check. On the other hand, it looks to be correct. I wonder if the problem is that elem is not captured by the lambda and is being accessed as a field later when the local element has changed when the AsyncDoFn is outputting and it is validating the timestamp against a later element. SimpleDoFnRunner is not designed for concurrent or async use, so that could be the problem. A simple fix that would preserve safety would be to capture it in a local variable that ends up in the closure of the lambda.

Confirmed the test that fails: https://develocity.apache.org/s/unfvg6wymazji/tests/overview?outcome=FAILED

https://github.com/apache/beam/pull/36838/files might fix it - if you have a test environment where it is easy to pull this in and try it (or test against a snapshot once the nightlies run)

github-actions bot added java build labels May 9, 2025

kennknowles commented May 9, 2025

View reviewed changes

sdks/java/core/src/main/java/org/apache/beam/sdk/util/WindowedValue.java Outdated Show resolved Hide resolved

kennknowles force-pushed the OutputBuilder branch from a805235 to 0ae2e78 Compare May 9, 2025 16:04

kennknowles changed the title ~~DO NOT MERGE: demonstrating impact of OutputBuilder~~ DO NOT MERGE: demonstrating OutputBuilder in Java SDK May 9, 2025

kennknowles force-pushed the OutputBuilder branch from 0ae2e78 to efe4648 Compare May 30, 2025 15:15

github-actions bot added io extensions runners spark sql dataflow direct gcp kafka core flink fn-execution jet samza twister2 spanner labels May 30, 2025

kennknowles force-pushed the OutputBuilder branch from efe4648 to a888fb9 Compare June 2, 2025 20:44

kennknowles changed the title ~~DO NOT MERGE: demonstrating OutputBuilder in Java SDK~~ Feedback requested: introduce OutputBuilder in Java SDK Jun 2, 2025

kennknowles commented Jun 3, 2025

View reviewed changes

...va/core/src/main/java/org/apache/beam/sdk/util/construction/SplittableParDoNaiveBounded.java Show resolved Hide resolved

kennknowles commented Jun 3, 2025

View reviewed changes

sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/DoFn.java Show resolved Hide resolved

kennknowles force-pushed the OutputBuilder branch 3 times, most recently from b4f82b1 to 52afeda Compare June 20, 2025 20:12

kennknowles force-pushed the OutputBuilder branch from 52afeda to 38d34d3 Compare July 1, 2025 19:13

kennknowles force-pushed the OutputBuilder branch 2 times, most recently from 4289ea0 to fe0763a Compare July 21, 2025 22:43

github-actions bot added the spanner label Jul 21, 2025

stankiewicz mentioned this pull request Jul 30, 2025

Element metadata in windowed value for offset deduplication #35739

Merged

3 tasks

github-actions bot added the Next Action: Author label Aug 12, 2025

kennknowles force-pushed the OutputBuilder branch from fe0763a to 0a72b50 Compare August 13, 2025 14:44

stankiewicz reviewed Sep 18, 2025

View reviewed changes

stankiewicz approved these changes Sep 18, 2025

View reviewed changes

kennknowles mentioned this pull request Sep 18, 2025

Cleanly separate v1 worker and v2 sdk harness container image handling in DataflowRunner #36138

Merged

3 tasks

kennknowles added 3 commits September 22, 2025 11:54

Touch trigger files for integration tests for OutputBuilder

8ca2a0a

Move WindowMatchers to main component of runners-core

aa27e9e

Add OutputBuilder to the Java SDK and use in runners

2cff4cc

Amar3tto mentioned this pull request Oct 2, 2025

The PostRelease Nightly Snapshot job is flaky #30505

Closed

Abacn mentioned this pull request Oct 3, 2025

[Feature Request]: OpenTelemetry Support #33176

Open

17 tasks

gemini-code-assist bot mentioned this pull request Oct 14, 2025

Call out OutputBuilder change in CHANGES.md #36511

Merged

3 tasks

claudevdm mentioned this pull request Nov 3, 2025

Updated forked spannerio. GoogleCloudPlatform/DataflowTemplates#2955

Closed

kellen reviewed Nov 14, 2025

View reviewed changes

clairemcginty mentioned this pull request Jan 7, 2026

Rework AsyncDoFn/FileDownloadDoFn to avoid outputting element outside of allowed timestamp bound spotify/scio#5866

Open

Conversation

kennknowles commented May 9, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

GitHub Actions Tests Status (on master branch)

Uh oh!

Uh oh!

Uh oh!

Uh oh!

derrickaw commented Aug 12, 2025

Uh oh!

kennknowles commented Sep 12, 2025

Uh oh!

github-actions bot commented Sep 12, 2025

Uh oh!

kennknowles commented Sep 12, 2025

Uh oh!

kennknowles commented Sep 12, 2025

Uh oh!

stankiewicz Sep 18, 2025

Choose a reason for hiding this comment

Uh oh!

stankiewicz Sep 18, 2025

Choose a reason for hiding this comment

Uh oh!

stankiewicz left a comment

Choose a reason for hiding this comment

Uh oh!

kennknowles commented Sep 18, 2025

Uh oh!

kennknowles commented Sep 23, 2025

Uh oh!

kellen Nov 14, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kennknowles Nov 14, 2025

Choose a reason for hiding this comment

Uh oh!

kellen Nov 15, 2025

Choose a reason for hiding this comment

Uh oh!

kennknowles Nov 17, 2025 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Choose a reason for hiding this comment

Uh oh!

kennknowles Nov 17, 2025

Choose a reason for hiding this comment

Uh oh!

kennknowles Nov 17, 2025

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

kennknowles commented May 9, 2025 •

edited

Loading

kellen Nov 14, 2025 •

edited

Loading

kennknowles Nov 17, 2025 •

edited

Loading