[OpenTelemetry] FNHarness, Dataflow Runner v1 - set open telemetry settings. (#38785)

Author: stankiewicz

runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/StreamingDataflowWorker.java

@@ -115,6 +115,7 @@
 import org.apache.beam.sdk.io.gcp.bigquery.BigQuerySinkMetrics;
 import org.apache.beam.sdk.metrics.MetricsEnvironment;
 import org.apache.beam.sdk.options.ExperimentalOptions;
+import org.apache.beam.sdk.options.SdkHarnessOptions;
 import org.apache.beam.sdk.util.construction.CoderTranslation;
 import org.apache.beam.sdk.values.WindowedValues;
 import org.apache.beam.vendor.grpc.v1p69p0.io.grpc.auth.MoreCallCredentials;
@@ -1042,6 +1043,18 @@ public static void main(String[] args) throws Exception {
       WindowedValues.FullWindowedValueCoder.setMetadataSupported();
     }
 
+    SdkHarnessOptions sdkHarnessOptions = options.as(SdkHarnessOptions.class);
+    Map<String, String> openTelemetryProperties = sdkHarnessOptions.getOpenTelemetryProperties();
+    if (openTelemetryProperties != null && !openTelemetryProperties.isEmpty()) {
+      openTelemetryProperties.forEach(
+          (k, v) -> {
+            if (k != null && v != null) {
+              System.setProperty(k, v);
+            }
+          });
+      LOG.info("Enabled Open Telemetry with properties: {}", openTelemetryProperties);
+    }
+
     LOG.debug("Creating StreamingDataflowWorker from options: {}", options);
     StreamingDataflowWorker worker = StreamingDataflowWorker.fromOptions(options);
 

sdks/java/harness/src/main/java/org/apache/beam/fn/harness/FnHarness.java

@@ -296,6 +296,17 @@ public static void main(
       // Register standard file systems.
       FileSystems.setDefaultPipelineOptions(options);
       CoderTranslation.verifyModelCodersRegistered();
+      SdkHarnessOptions sdkHarnessOptions = options.as(SdkHarnessOptions.class);
+      Map<String, String> openTelemetryProperties = sdkHarnessOptions.getOpenTelemetryProperties();
+      if (openTelemetryProperties != null && !openTelemetryProperties.isEmpty()) {
+        openTelemetryProperties.forEach(
+            (k, v) -> {
+              if (k != null && v != null) {
+                System.setProperty(k, v);
+              }
+            });
+        LOG.info("Enabled Open Telemetry with properties: {}", openTelemetryProperties);
+      }
       EnumMap<
               BeamFnApi.InstructionRequest.RequestCase,
               ThrowingFunction<InstructionRequest, BeamFnApi.InstructionResponse.Builder>>