[Java] Propagate record ID and offset from Windmill context to WindowedValue #37232

Suvrat1629 · 2026-01-06T13:53:02Z

Please add a meaningful description for your change here

Propagate element-level metadata (record ID and offset) from Windmill to WindowedValue in UngroupedWindmillReader
This change extracts the record ID and record offset that are already available in StreamingModeExecutionContext (populated from Windmill messages) and passes them into the created WindowedValue objects.

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.

gemini-code-assist · 2026-01-06T13:53:19Z

Summary of Changes

Hello @Suvrat1629, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request addresses issue #37030 by enhancing the Apache Beam Dataflow runner's ability to carry essential element-level metadata. It ensures that the record ID and offset, originating from Windmill messages, are correctly extracted and attached to WindowedValue objects. This fundamental change provides richer context for data elements, which is vital for advanced processing capabilities such as offset-based deduplication in streaming pipelines.

Highlights

Metadata Propagation: This pull request enables the propagation of element-level metadata, specifically record ID and offset, from the Windmill context to WindowedValue objects within the UngroupedWindmillReader.
Contextual Data Extraction: The changes extract the record ID and record offset from the StreamingModeExecutionContext (where they are populated from Windmill messages) and pass them into the WindowedValue objects being created.
Support for Offset-Based Deduplication: The propagated metadata is crucial for supporting features like offset-based deduplication, ensuring that unique data elements can be identified and processed correctly.

🧠 New Feature in Public Preview: You can now enable Memory to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	`/gemini review`	Performs a code review for the current pull request in its current state.
Pull Request Summary	`/gemini summary`	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	`/gemini help`	Displays a list of available commands.

Customization

To customize Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counter productive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for Github and other Google products, sign up here.

You can also get AI-powered code generation, chat, as well as code reviews directly in the IDE at no cost with the Gemini Code Assist IDE Extension.

Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

github-actions · 2026-01-06T15:08:23Z

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

Suvrat1629 · 2026-01-07T04:10:45Z

assign set of reviewers

github-actions · 2026-01-07T04:11:50Z

Assigning reviewers:

R: @damccorm added as fallback since no labels match configuration

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

stop reviewer notifications - opt out of the automated review tooling
remind me after tests pass - tag the comment author after tests pass
waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).

Suvrat1629 · 2026-01-07T04:16:58Z

/gemini review

gemini-code-assist

Code Review

This pull request propagates the record ID and offset from the Windmill context to WindowedValue objects. The changes in UngroupedWindmillReader correctly extract this information from StreamingModeExecutionContext when offset-based deduplication is supported and pass it to WindowedValues.of(). I've added one comment to prevent a potential RuntimeException if an empty record ID is received from the context.

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/UngroupedWindmillReader.java

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

Suvrat1629 · 2026-01-07T04:24:43Z

/gemini review

gemini-code-assist

Code Review

This pull request aims to propagate the record ID and offset from the Windmill context to WindowedValue objects in UngroupedWindmillReader. This is a valuable addition for enabling features like exactly-once processing. The implementation correctly passes the extracted record ID and offset to WindowedValues.of(). However, I have a significant concern about the source of this metadata. The current approach of reading from StreamingModeExecutionContext seems likely to assign the same ID and offset to all elements within a WorkItem, which would be incorrect. My review comment details this concern and suggests an alternative.

gemini-code-assist · 2026-01-07T04:27:44Z

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/UngroupedWindmillReader.java

+      if (context.offsetBasedDeduplicationSupported()) {
+        byte[] rawId = context.getCurrentRecordId();
+        if (rawId != null && rawId.length > 0) {
+          recordId = new String(rawId, StandardCharsets.UTF_8);
+        }
+        byte[] rawOffset = context.getCurrentRecordOffset();
+        if (rawOffset != null && rawOffset.length == Longs.BYTES) {
+          recordOffset = Longs.fromByteArray(rawOffset);
+        }
+      }


This logic for retrieving record ID and offset from StreamingModeExecutionContext appears to have a potential flaw. The UngroupedWindmillReaderIterator iterates over messages within a WorkItem, but the underlying activeReader in the context (from which getCurrentRecordId and getCurrentRecordOffset get their values) does not appear to be advanced per message. This will likely result in all elements from the same WorkItem receiving the same record ID and offset, which would be incorrect for element-level metadata and could break features like deduplication.

A more robust approach would be to extract the record ID and offset directly from the Windmill.InputMessageBundle for each message, if this information is available there. This would ensure that each element gets its unique metadata.

stankiewicz · 2026-01-08T10:00:12Z

hi Suvrak,

purpose of the issue was to track changes around https://github.com/apache/beam/blob/v2.70.0/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/UngroupedWindmillReader.java#L135
where we have several context items in ElementMetadata that has to land in WindowedValue.
ElementMetadata is used as transport structure between stages.
offset and record_id id you've added are not part of it, they are not read from windmill so this change is wrong.
In 2.71/master todo was removed from the code (https://github.com/apache/beam/blob/master/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/UngroupedWindmillReader.java) and you can see drain being propagated from ElementMetadata.
In the future we will be adding new context pieces like tracing context and cdc info.

Suvrat1629 · 2026-01-08T11:56:18Z

@stankiewicz
Thanks for the clarification — I see now that recordId/offset are Dataflow-specific and not in ElementMetadata. Should I close this PR, or is there a separate need for propagating these?

stankiewicz · 2026-01-08T12:41:47Z

I think you can close this PR. Thank you for looking into this!

github-actions bot added runners dataflow labels Jan 6, 2026

Suvrat1629 added 2 commits January 6, 2026 21:09

additional metadata

e9d048f

minor import fix

eba0eac

Suvrat1629 force-pushed the context-switch branch from 3a189b6 to eba0eac Compare January 6, 2026 15:59

Suvrat1629 added 2 commits January 6, 2026 21:32

fix

f718f9c

minor changes

300f50c

github-actions bot added the Next Action: Reviewers label Jan 7, 2026

gemini-code-assist bot reviewed Jan 7, 2026

View reviewed changes

...va/worker/src/main/java/org/apache/beam/runners/dataflow/worker/UngroupedWindmillReader.java Outdated Show resolved Hide resolved

addressing gemini comments

3e97f03

Co-authored-by: gemini-code-assist[bot] <176961590+gemini-code-assist[bot]@users.noreply.github.com>

gemini-code-assist bot reviewed Jan 7, 2026

View reviewed changes

Suvrat1629 closed this Jan 8, 2026

[Java] Propagate record ID and offset from Windmill context to WindowedValue #37232

[Java] Propagate record ID and offset from Windmill context to WindowedValue #37232

Conversation

Suvrat1629 commented Jan 6, 2026

GitHub Actions Tests Status (on master branch)

Uh oh!

gemini-code-assist bot commented Jan 6, 2026

Summary of Changes

Highlights

Footnotes

Uh oh!

github-actions bot commented Jan 6, 2026

Uh oh!

Suvrat1629 commented Jan 7, 2026

Uh oh!

github-actions bot commented Jan 7, 2026

Uh oh!

Suvrat1629 commented Jan 7, 2026

Uh oh!

gemini-code-assist bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

Uh oh!

Suvrat1629 commented Jan 7, 2026

Uh oh!

gemini-code-assist bot left a comment

Choose a reason for hiding this comment

Code Review

Uh oh!

gemini-code-assist bot Jan 7, 2026

Choose a reason for hiding this comment

Uh oh!

stankiewicz commented Jan 8, 2026

Uh oh!

Suvrat1629 commented Jan 8, 2026

Uh oh!

stankiewicz commented Jan 8, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

2 participants