[stankiewicz]

Introduces standard coder definitions in beam_runner_api.proto for WindowedValue and Timer V2:

• WINDOWED_VALUE_V2 (beam:coder:windowed_value:v2 = 19)
• TIMER_V2 (beam:coder:timer:v2 = 20)

These coders enable systemic element metadata propagation (e.g. pipeline drain signals like DrainMode::DRAINING to terminate looping timers, as well as tracing context) across Fn API boundaries.

Encoding utilizes the MSB (0x80) of the first PaneInfo byte as a metadata presence bit:

• When bit 0x80 is 0, metadata is absent, maintaining an identical wire footprint to V1 with zero overhead for standard pipelines.
• When bit 0x80 is 1, a VarInt64 length prefix and serialized ElementMetadata proto bytes immediately follow PaneInfo.

[github-actions]

📋 Review Summary

This PR introduces the protocol buffer specifications for WINDOWED_VALUE_V2 (beam:coder:windowed_value:v2), PARAM_WINDOWED_VALUE_V2 (beam:coder:param_windowed_value:v2), and TIMER_V2 (beam:coder:timer:v2) in beam_runner_api.proto to support encoding optional ElementMetadata when the 0x80 bit of the PaneInfo byte is set.

🔍 General Feedback

• The new coder URN definitions, enum field numbers, and wire format documentation are well-structured, backward-compatible, and align with existing Java SDK implementations (WindowedValues.FullWindowedValueCoder and Timer.Coder) and beam_fn_api.proto (ElementMetadata).
• One minor inaccuracy was spotted in the PARAM_WINDOWED_VALUE_V2 payload documentation regarding how the constant value is serialized.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

💬 Inline Findings

• model/pipeline/src/main/proto/org/apache/beam/model/pipeline/v1/beam_runner_api.proto (line 943): 🟢 Low: FullWindowedValueCoder is parameterized by two coders (the value coder and the window coder), not just GlobalWindow.Coder. As implemented in Java's ParamWindowedValueCoder.toProto (and Python's ParamWindowedValueCoderImpl._from_proto), the parameterized payload is encoded using FullWindowedValueCoder.of(ByteArrayCoder.of(), getWindowCoder()) with an empty byte array new byte[0] as the dummy value.

    // Payload: The parameterized WindowedValue encoded via FullWindowedValueCoder(ByteArrayCoder, windowCoder)

[stankiewicz]

R: @acrites
R: @kennknowles

[github-actions]

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

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 79.92%. Comparing base (a8d6afd) to head (58f02df).
⚠️ Report is 58 commits behind head on master.

Additional details and impacted files

@@            Coverage Diff             @@
##           master   #40744      +/-   ##
==========================================
- Coverage   80.04%   79.92%   -0.12%     
==========================================
  Files         490      493       +3     
  Lines       82916    83402     +486     
==========================================
+ Hits        66373    66662     +289     
- Misses      16543    16740     +197     

Flag	Coverage Δ
python	79.92% <ø> (-0.12%)	⬇️

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.