|
| 1 | +--- |
| 2 | +# Licensed to the Apache Software Foundation (ASF) under one |
| 3 | +# or more contributor license agreements. See the NOTICE file |
| 4 | +# distributed with this work for additional information |
| 5 | +# regarding copyright ownership. The ASF licenses this file |
| 6 | +# to you under the Apache License, Version 2.0 (the |
| 7 | +# "License"); you may not use this file except in compliance |
| 8 | +# with the License. You may obtain a copy of the License at |
| 9 | +# |
| 10 | +# http://www.apache.org/licenses/LICENSE-2.0 |
| 11 | +# |
| 12 | +# Unless required by applicable law or agreed to in writing, |
| 13 | +# software distributed under the License is distributed on an |
| 14 | +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY |
| 15 | +# KIND, either express or implied. See the License for the |
| 16 | +# specific language governing permissions and limitations |
| 17 | +# under the License. |
| 18 | + |
| 19 | +name: adding-new-metadata |
| 20 | +description: Guide on how to add and propagate new metadata fields in Apache Beam's WindowedValue, extending protos, windmill persistence, and runner interfaces to avoid metadata loss. |
| 21 | +--- |
| 22 | + |
| 23 | +# Adding New Metadata to WindowedValue |
| 24 | + |
| 25 | +This skill provides a comprehensive guide on adding new metadata (e.g., CDC metadata, drain mode flags, OpenTelemetry trace context) to Apache Beam's `WindowedValue` and ensuring it propagates correctly through the execution engine. Failing to propagate metadata in all necessary places will result in metadata loss during pipeline execution. |
| 26 | + |
| 27 | +## 1. Extending the Proto Model |
| 28 | + |
| 29 | +When adding new metadata that must cross worker boundaries or be serialized by the Fn API, the proto definitions must be updated. |
| 30 | + |
| 31 | +* **Key Files:** `model/fn-execution/src/main/proto/org/apache/beam/model/fn_execution/v1/beam_fn_api.proto` |
| 32 | +* **Action:** Add the new metadata field to the appropriate message (`ElementMetadata`). |
| 33 | +* **Note:** Add proper documentation in proto. Type of the field can be different from the type in WindowedValue, see OpenTelemetry trace context for example. |
| 34 | + |
| 35 | +## 2. WindowedValue Interface and Implementations |
| 36 | + |
| 37 | +The `WindowedValue` is the core container for elements flowing through a Beam pipeline. It holds the value, timestamp, windows, pane info, and any additional metadata. |
| 38 | + |
| 39 | +### Core Interface Updates |
| 40 | +* **Key File:** `sdks/java/core/src/main/java/org/apache/beam/sdk/values/WindowedValue.java` |
| 41 | +* **Action:** Add getter methods for your new metadata. |
| 42 | + |
| 43 | +### Concrete Implementations |
| 44 | +You must update **all** concrete implementations of `WindowedValue` to store and return the new metadata. If you miss one, metadata will be silently dropped. |
| 45 | +* `ValueInGlobalWindow` |
| 46 | +* `ValueInSingleWindow` |
| 47 | +* `ValueInEmptyWindows` (often used inside runners, like Dataflow's worker package) |
| 48 | +* **Action:** Update constructors, factory methods (`of()`), fields in these classes and coders. |
| 49 | + |
| 50 | +### OutputBuilder vs. Context Output |
| 51 | +* **IMPORTANT:** Do **not** add new arguments to legacy methods like `context.outputWindowedValue(...)` or `WindowedValue.of(value, timestamp, windows, pane)`. This causes brittleness and breaks the API for every new metadata field. |
| 52 | +* **Action:** Modify `OutputBuilder` (`sdks/java/core/src/main/java/org/apache/beam/sdk/values/OutputBuilder.java`) to accept the new metadata (e.g., `.withDrainMode(...)`, `.withTraceContext(...)`). Use the builder pattern when constructing outputs to propagate offset and record IDs smoothly. |
| 53 | + |
| 54 | +## 3. Windmill Persistence (Dataflow Streaming Engine) Runner v1 |
| 55 | + |
| 56 | +For the Dataflow streaming runner, metadata must survive serialization to and from the Windmill backend. |
| 57 | + |
| 58 | +* **Serialization (Sink):** |
| 59 | + * **File:** `runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/WindmillSink.java` |
| 60 | + * **Action:** Extract the metadata from the `WindowedValue`, and add it to already created ElementMetadata proto builder. |
| 61 | +* **Deserialization (Reader):** |
| 62 | + * **Files:** `runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/UngroupedWindmillReader.java` and `WindowingWindmillReader.java` |
| 63 | + * **Action:** Extract the metadata from ElementMetadata proto and reconstruct the `WindowedValue` using the updated factory methods/builders that include the metadata. This is incremental work, as plenty of metadata is already extracted from the proto. |
| 64 | + |
| 65 | +## 4. Propagation Across Core Classes |
| 66 | + |
| 67 | +Metadata must be explicitly copied or forwarded whenever a `WindowedValue` is transformed, buffered, or processed. |
| 68 | + |
| 69 | +### DoFn Runners (Java Core) |
| 70 | +You must ensure that when a DoFn processes an element and outputs a new element, the appropriate metadata from the *input* is propagated to the *output* (unless explicitly changed by the logic). |
| 71 | +* `runners/core-java/src/main/java/org/apache/beam/runners/core/SimpleDoFnRunner.java` |
| 72 | +* `runners/core-java/src/main/java/org/apache/beam/runners/core/StatefulDoFnRunner.java` |
| 73 | +* `runners/core-java/src/main/java/org/apache/beam/runners/core/LateDataDroppingDoFnRunner.java` |
| 74 | +* `runners/core-java/src/main/java/org/apache/beam/runners/core/ProcessFnRunner.java` |
| 75 | +* `sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnApiDoFnRunner.java` |
| 76 | + |
| 77 | +**Action:** When these runners call `outputWindowedValue()`, they should extract the metadata from the input or current context and attach it using the `OutputBuilder` or the new `WindowedValue` interfaces. |
| 78 | + |
| 79 | +### Grouping and Reducing |
| 80 | +* `runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnRunner.java` |
| 81 | +* `runners/core-java/src/main/java/org/apache/beam/runners/core/ReduceFnContextFactory.java` |
| 82 | +* **Action:** Ensure that during GroupByKey/Combine operations, if metadata needs to be preserved (e.g., `CausedByDrain`), it is correctly passed into the `ReduceFnContextFactory` and propagated when outputting the grouped results. |
| 83 | + |
| 84 | +### Splittable DoFns (SDF) |
| 85 | +* `runners/core-java/src/main/java/org/apache/beam/runners/core/OutputAndTimeBoundedSplittableProcessElementInvoker.java` |
| 86 | +* `sdks/java/core/src/main/java/org/apache/beam/sdk/util/construction/SplittableParDoNaiveBounded.java` |
| 87 | + |
| 88 | +### Timers |
| 89 | +If metadata needs to survive timer firings (e.g., knowing an `@OnTimer` fired because of a system drain), it must be added to Timer data structures. This is a bit of uncharted area which was only implemented for CausedByDrain metadata that comes from backend, not from persisted metadata. In order to persist all WindowedValue metadata across timer, more work has to be done, below are some pointers: |
| 90 | +* `runners/core-java/src/main/java/org/apache/beam/runners/core/TimerInternals.java` and implementations (e.g., `WindmillTimerInternals.java` in Dataflow). |
| 91 | +* `runners/samza/src/test/java/org/apache/beam/runners/samza/runtime/KeyedTimerData.java` (or generic `TimerData`). |
| 92 | +* **Action:** Add the field to `TimerData`, next to `CausedByDrain`. Propagate it when setting the timer and expose it when the timer fires so it bubbles up. |
| 93 | +* Eventually, metadata from Timer lands in WindowedValue, so it can be exposed to users. Keep field names, types, and getters similar to WindowedValue as much as possible, as common interface may be introduced eventually. |
| 94 | + |
| 95 | +## 5. Exposing Metadata to the User (DoFn Signatures) |
| 96 | + |
| 97 | +User needs to access the metadata in their `DoFn` (e.g., `@ProcessElement public void process(ProcessContext c, CausedByDrain drain) { ... }`), you must update the reflection and bytecode generation logic. |
| 98 | + |
| 99 | +* **Files:** |
| 100 | + * `sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignatures.java` |
| 101 | + * `sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnSignature.java` |
| 102 | + * `sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/DoFnInvoker.java` |
| 103 | + * `sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/reflect/ByteBuddyDoFnInvokerFactory.java` |
| 104 | +* **Action:** Add logic to detect the new parameter type in the DoFn method signature. Generate bytecode using ByteBuddy to extract the property from the `WindowedValue` or context and pass it as an argument during method invocation. |
| 105 | + |
| 106 | +## Checklist for Adding New Metadata |
| 107 | + |
| 108 | +1. [ ] Define the metadata in `beam_fn_api.proto` (if applicable). |
| 109 | +2. [ ] Add getters to the `WindowedValue` interface. |
| 110 | +3. [ ] Update `ValueInGlobalWindow`, `ValueInSingleWindow`, `ValueInEmptyWindows` to store the metadata. |
| 111 | +4. [ ] Update `OutputBuilder` to accept the metadata. |
| 112 | +5. [ ] Update `WindmillSink` to serialize the metadata to the backend. |
| 113 | +6. [ ] Update `UngroupedWindmillReader` and `WindowingWindmillReader` to deserialize the metadata. |
| 114 | +7. [ ] Update `WindmillKeyedWorkItem`. |
| 115 | +8. [ ] Update `SimpleDoFnRunner`, `StatefulDoFnRunner`, and `FnApiDoFnRunner` to propagate the metadata from input to output. |
| 116 | +9. [ ] Update `ReduceFnRunner` and `OutputAndTimeBoundedSplittableProcessElementInvoker` for complex transform propagation. |
| 117 | +10. [ ] If required by timers, update `TimerData` and `TimerInternals`. |
| 118 | +11. [ ] If exposed to the user, update `DoFnSignatures` and `ByteBuddyDoFnInvokerFactory`. |
| 119 | +12. [ ] Update other runners (Flink, Spark, Samza) to ensure they propagate the new `WindowedValue` fields correctly in their specific operators/runners. |
0 commit comments