diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/logging/DataflowWorkerLoggingHandler.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/logging/DataflowWorkerLoggingHandler.java
index e8d674af8c92..6c383acc74f7 100644
--- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/logging/DataflowWorkerLoggingHandler.java
+++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/logging/DataflowWorkerLoggingHandler.java
@@ -43,6 +43,8 @@
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Iterators;
 import com.google.protobuf.Struct;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
 import java.io.BufferedOutputStream;
 import java.io.File;
 import java.io.FileOutputStream;
@@ -147,6 +149,8 @@ static ResourceBundle resourceBundleForNonDirectLogLevelHint(Level nonDirectLogL
   /** If true, add SLF4J MDC to custom_data of the log message. */
   private final AtomicBoolean logCustomMdc = new AtomicBoolean(false);
 
+  private final AtomicBoolean logOpenTelemetryTraceSpanIdAndSampled = new AtomicBoolean(false);
+
   // Only instantiated and set if enableDirectLogging is called.
   private static class DirectLoggingState {
     DirectLoggingState(
@@ -250,6 +254,10 @@ public void setLogMdc(boolean enabled) {
     logCustomMdc.set(enabled);
   }
 
+  public void setLogOpenTelemetryTraceAndSpanId(boolean enabled) {
+    logOpenTelemetryTraceSpanIdAndSampled.set(enabled);
+  }
+
   private static Pair<ImmutableMap<String, String>, ImmutableMap<String, String>>
       labelsFromOptionsAndMetadata(PipelineOptions options) {
     DataflowPipelineOptions dataflowOptions = options.as(DataflowPipelineOptions.class);
@@ -385,7 +393,16 @@ LogEntry constructDirectLogEntry(
         LogEntry.newBuilder(Payload.JsonPayload.of(payloadBuilder.build()))
             .setTimestamp(Instant.ofEpochMilli(record.getMillis()))
             .setSeverity(severityFor(record.getLevel()));
-
+    if (logOpenTelemetryTraceSpanIdAndSampled.get()) {
+      SpanContext spanContext = Span.current().getSpanContext();
+      if (spanContext.isValid() && spanContext.isSampled()) {
+        builder =
+            builder
+                .setTrace(spanContext.getTraceId())
+                .setSpanId(spanContext.getSpanId())
+                .setTraceSampled(spanContext.isSampled());
+      }
+    }
     if (stepId != null) {
       builder.setResource(
           MonitoredResource.newBuilder(RESOURCE_TYPE)
@@ -606,6 +623,16 @@ public synchronized void publishToDisk(
       writeIfNotEmpty(generator, "work", DataflowWorkerLoggingMDC.getWorkId());
       writeIfNotEmpty(generator, "logger", record.getLoggerName());
       writeIfNotEmpty(generator, "exception", formatException(record.getThrown()));
+
+      if (logOpenTelemetryTraceSpanIdAndSampled.get()) {
+        SpanContext spanContext = Span.current().getSpanContext();
+        if (spanContext.isValid() && spanContext.isSampled()) {
+          generator.writeStringField("trace", spanContext.getTraceId());
+          generator.writeStringField("spanId", spanContext.getSpanId());
+          generator.writeBooleanField("trace_sampled", spanContext.isSampled());
+        }
+      }
+
       if (logCustomMdc.get()) {
         @Nullable Map<String, String> mdcMap = MDC.getCopyOfContextMap();
         if (mdcMap != null && !mdcMap.isEmpty()) {
diff --git a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/logging/DataflowWorkerLoggingInitializer.java b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/logging/DataflowWorkerLoggingInitializer.java
index d854ae74ebaf..1627c96a3427 100644
--- a/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/logging/DataflowWorkerLoggingInitializer.java
+++ b/runners/google-cloud-dataflow-java/worker/src/main/java/org/apache/beam/runners/dataflow/worker/logging/DataflowWorkerLoggingInitializer.java
@@ -383,6 +383,10 @@ public static synchronized void configure(DataflowWorkerLoggingOptions options)
       loggingHandler.setLogMdc(true);
     }
 
+    if (harnessOptions.getLogOpenTelemetryTraceAndSpanId()) {
+      loggingHandler.setLogOpenTelemetryTraceAndSpanId(true);
+    }
+
     if (usedDeprecated) {
       LOG.warn(
           "Deprecated DataflowWorkerLoggingOptions are used for log level settings."
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/SdkHarnessOptions.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/SdkHarnessOptions.java
index 7267dda9ed0b..531c86ef0438 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/SdkHarnessOptions.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/SdkHarnessOptions.java
@@ -114,6 +114,13 @@ enum LogLevel {
 
   void setLogMdc(boolean value);
 
+  @Description(
+      "This option controls if OpenTelemetry trace, spanId and sampled will be appended to log entries. This will allow to stitch traces to logs.")
+  @Default.Boolean(false)
+  boolean getLogOpenTelemetryTraceAndSpanId();
+
+  void setLogOpenTelemetryTraceAndSpanId(boolean value);
+
   /** This option controls whether logging will be redirected through the FnApi. */
   @Description(
       "Controls whether logging will be redirected through the FnApi. In normal usage, setting "