[OpenTelemetry] Context propagation for Runner v1 and SimpleDoFnRunner#39152
[OpenTelemetry] Context propagation for Runner v1 and SimpleDoFnRunner#39152stankiewicz wants to merge 1 commit into
Conversation
Summary of ChangesHello, I'm Gemini Code Assist1! 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 introduces per-element OpenTelemetry context propagation for Runner v1, specifically targeting Dataflow worker components. By enabling the serialization and deserialization of trace contexts through BeamFnApi metadata, the changes ensure that distributed traces are correctly maintained across different pipeline execution steps and DoFn invocations. Highlights
New Features🧠 You can now enable Memory (public preview) 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 AssistThe 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
Customization To customize the 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 Limitations & Feedback Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. 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. Footnotes
|
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request implements OpenTelemetry context propagation across Beam runners and sinks, specifically integrating it into element processing and Windmill metadata encoding/decoding. Feedback focuses on critical performance and correctness issues: first, avoiding object allocation and try-with-resources overhead on the hot path in SimpleDoFnRunner when no context is present; second, avoiding dynamic evaluation of Context.current() inside WindowedValues's getter, which breaks context propagation across thread boundaries, and instead capturing the context at construction time.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| // builder may have different context set at the beginning of parDo | ||
| // when building WindowedValue we should take current context from storage. | ||
| return openTelemetryContext != Context.current() ? Context.current() : openTelemetryContext; |
There was a problem hiding this comment.
Dynamically evaluating Context.current() inside the getter getOpenTelemetryContext() is highly problematic and breaks context propagation across thread boundaries.
WindowedValue is an immutable value representation of an element. If getOpenTelemetryContext() is called on a different thread (e.g., during asynchronous processing, buffering, or when writing to a sink on a background thread), Context.current() will return the context of that background thread (or Context.root()), which differs from openTelemetryContext. As a result, the getter will incorrectly return the background thread's context instead of the element's original context.
To fix this correctness issue, the current OpenTelemetry context should be captured at construction time (e.g., in the constructor or builder of WindowedValue), and getOpenTelemetryContext() should simply return the captured field without any dynamic checks.
| // builder may have different context set at the beginning of parDo | |
| // when building WindowedValue we should take current context from storage. | |
| return openTelemetryContext != Context.current() ? Context.current() : openTelemetryContext; | |
| return openTelemetryContext; |
There was a problem hiding this comment.
good point but usage is from within the thread that is having the proper context.
stankiewicz
force-pushed
the
otel_propagate_context_through_runner_v1
branch
2 times, most recently
from
9aca3b9 to
a41cbe8
Compare
…nRunner changes for per element propagation
stankiewicz
force-pushed
the
otel_propagate_context_through_runner_v1
branch
from
a41cbe8 to
258ec39
Compare
|
Assigning reviewers: R: @chamikaramj for label java. Note: If you would like to opt out of this review, comment Available commands:
The PR bot will only process comments in the main thread (not review comments). |
1 participant
This PR enables per-element OpenTelemetry context propagation for Runner v1, specifically targeting the Dataflow worker (Windmill sink and source) and DoFn execution. This ensures that distributed traces correctly link parent and child spans across different pipeline steps.
DoFn Execution (SimpleDoFnRunner): Updated processElement to wrap the invocation in an OpenTelemetry Scope tied to the element's OpenTelemetryContext. This ensures spans created inside the DoFn are properly parented to the element's trace.
Context Serialization (WindmillSink): Extracts the OTel context from outbound WindowedValue records and injects it into the BeamFnApi.Elements.ElementMetadata for transmission.
Context Deserialization (UngroupedWindmillReader & WindmillKeyedWorkItem): Extracts the OTel context from incoming ElementMetadata via OpenTelemetryContextPropagator and attaches it to the constructed WindowedValue.
Context Refresh (WindowedValues): Updated getOpenTelemetryContext() to dynamically pull Context.current() if the context changes during ParDo execution, ensuring the recent trace state is stored.
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
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, commentfixes #<ISSUE NUMBER>instead.CHANGES.mdwith noteworthy changes.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.