rename method, fix constructor invocation.

Author: stankiewicz

runners/core-java/src/main/java/org/apache/beam/runners/core/SplittableParDoViaKeyedWorkItems.java

@@ -472,7 +472,8 @@ public String getErrorContext() {
                   read.getPaneInfo(),
                   read.getRecordId(),
                   read.getRecordOffset(),
-                  CausedByDrain.CAUSED_BY_DRAIN);
+                  CausedByDrain.CAUSED_BY_DRAIN,
+                  read.getOpenTelemetryContext());
         }
         elementAndRestriction = KV.of(read, restrictionState.read());
         watermarkEstimatorStateT = watermarkEstimatorState.read();

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

@@ -61,7 +61,7 @@ public Iterable<String> keys(BeamFnApi.Elements.ElementMetadata carrier) {
         }
       };
 
-  static void write(Context from, BeamFnApi.Elements.ElementMetadata.Builder builder) {
+  static void set(Context from, BeamFnApi.Elements.ElementMetadata.Builder builder) {
     W3CTraceContextPropagator.getInstance().inject(from, builder, SETTER);
   }
 

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

@@ -139,7 +139,7 @@ public void encode(
             windowedElem.getOpenTelemetryContext();
         if (openTelemetryContext != null) {
 
-          OpenTelemetryContextPropagator.write(openTelemetryContext, builder);
+          OpenTelemetryContextPropagator.set(openTelemetryContext, builder);
         }
         BeamFnApi.Elements.ElementMetadata metadata = builder.build();
         ByteArrayCoder.of().encode(metadata.toByteArray(), outStream);

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

@@ -356,7 +356,8 @@ public static <T> WindowedValue<T> of(
 
     boolean isGlobal = GlobalWindow.INSTANCE.equals(window);
     if (isGlobal && BoundedWindow.TIMESTAMP_MIN_VALUE.equals(timestamp)) {
-      return new ValueInGlobalWindow<>(value, paneInfo, null, null, causedByDrain);
+      return new ValueInGlobalWindow<>(
+          value, paneInfo, null, null, causedByDrain, openTelemetryContext);
     } else if (isGlobal) {
       return new TimestampedValueInGlobalWindow<>(
           value,
@@ -1021,7 +1022,7 @@ public void encode(WindowedValue<T> windowedElem, OutputStream outStream, Coder.
         io.opentelemetry.context.Context openTelemetryContext =
             windowedElem.getOpenTelemetryContext();
         if (openTelemetryContext != null) {
-          OpenTelemetryContextPropagator.write(openTelemetryContext, builder);
+          OpenTelemetryContextPropagator.set(openTelemetryContext, builder);
         }
         BeamFnApi.Elements.ElementMetadata em =
             builder