diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
index 88c998ed9105..c80f5e2701e6 100644
--- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
+++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy
@@ -1418,9 +1418,10 @@ class BeamModulePlugin implements Plugin<Project> {
         getSourceDirectories().setFrom(
             project.files(project.sourceSets.main.allSource.srcDirs)
             )
-        getExecutionData().setFrom(project.file(
-            project.getLayout().getBuildDirectory().file("jacoco/test.exec")
-            ))
+       getExecutionData().setFrom(project.fileTree(
+          dir: project.getLayout().getBuildDirectory().dir("jacoco"),
+          include: ['*.exec']
+         ))
         reports {
           html.required = true
           xml.required = true
diff --git a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java
index 57f65ae62651..74ba8fa5a747 100644
--- a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java
+++ b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/SolaceIO.java
@@ -423,6 +423,7 @@ public class SolaceIO {
       Duration.standardSeconds(30);
   private static final Duration DEFAULT_ACK_DEADLINE = Duration.standardSeconds(30);
   public static final boolean DEFAULT_NACK_ON_TIMEOUT = false;
+  private static final boolean DEFAULT_ENABLE_OPEN_TELEMETRY_TRACING = false;
   public static final int DEFAULT_WRITER_NUM_SHARDS = 20;
   public static final int DEFAULT_WRITER_CLIENTS_PER_WORKER = 4;
   public static final Boolean DEFAULT_WRITER_PUBLISH_LATENCY_METRICS = false;
@@ -472,7 +473,8 @@ public static Read<Solace.Record> read() {
             .setDeduplicateRecords(DEFAULT_DEDUPLICATE_RECORDS)
             .setWatermarkIdleDurationThreshold(DEFAULT_WATERMARK_IDLE_DURATION_THRESHOLD)
             .setAckDeadline(DEFAULT_ACK_DEADLINE)
-            .setNackOnTimeout(DEFAULT_NACK_ON_TIMEOUT));
+            .setNackOnTimeout(DEFAULT_NACK_ON_TIMEOUT)
+            .setEnableOpenTelemetryTracing(DEFAULT_ENABLE_OPEN_TELEMETRY_TRACING));
   }
 
   /**
@@ -503,7 +505,8 @@ public static <T> Read<T> read(
             .setDeduplicateRecords(DEFAULT_DEDUPLICATE_RECORDS)
             .setWatermarkIdleDurationThreshold(DEFAULT_WATERMARK_IDLE_DURATION_THRESHOLD)
             .setAckDeadline(DEFAULT_ACK_DEADLINE)
-            .setNackOnTimeout(DEFAULT_NACK_ON_TIMEOUT));
+            .setNackOnTimeout(DEFAULT_NACK_ON_TIMEOUT)
+            .setEnableOpenTelemetryTracing(DEFAULT_ENABLE_OPEN_TELEMETRY_TRACING));
   }
 
   /**
@@ -627,6 +630,11 @@ public Read<T> withDeduplicateRecords(boolean deduplicateRecords) {
       return this;
     }
 
+    public Read<T> withEnableOpenTelemetryTracing() {
+      configurationBuilder.setEnableOpenTelemetryTracing(true);
+      return this;
+    }
+
     /**
      * Set a factory that creates a {@link org.apache.beam.sdk.io.solace.broker.SempClientFactory}.
      *
@@ -733,6 +741,8 @@ abstract static class Configuration<T> {
 
       abstract boolean getNackOnTimeout();
 
+      abstract boolean getEnableOpenTelemetryTracing();
+
       public static <T> Builder<T> builder() {
         Builder<T> builder =
             new org.apache.beam.sdk.io.solace.AutoValue_SolaceIO_Read_Configuration.Builder<T>();
@@ -752,6 +762,8 @@ public abstract static class Builder<T> {
 
         abstract Builder<T> setDeduplicateRecords(boolean deduplicateRecords);
 
+        abstract Builder<T> setEnableOpenTelemetryTracing(boolean enableOpenTelemetryTracing);
+
         abstract Builder<T> setParseFn(
             SerializableFunction<@Nullable BytesXMLMessage, @Nullable T> parseFn);
 
@@ -777,6 +789,12 @@ public void validate(@Nullable PipelineOptions options) {
       checkState(
           (configuration.getQueue() == null ^ configuration.getTopic() == null),
           "SolaceIO.Read: One of the Solace {Queue, Topic} must be set.");
+
+      if (configuration.getEnableOpenTelemetryTracing()) {
+        checkState(
+            TypeDescriptor.of(Solace.Record.class).isSupertypeOf(configuration.getTypeDescriptor()),
+            "OpenTelemetry tracing is only supported when the output type is Solace.Record.");
+      }
     }
 
     @Override
@@ -793,20 +811,24 @@ public PCollection<T> expand(PBegin input) {
 
       Coder<T> coder = inferCoder(input.getPipeline(), configuration.getTypeDescriptor());
 
-      return input.apply(
-          org.apache.beam.sdk.io.Read.from(
-              new UnboundedSolaceSource<>(
-                  initializedQueue,
-                  sempClientFactory,
-                  sessionServiceFactory,
-                  configuration.getMaxNumConnections(),
-                  configuration.getDeduplicateRecords(),
-                  coder,
-                  configuration.getTimestampFn(),
-                  configuration.getWatermarkIdleDurationThreshold(),
-                  configuration.getParseFn(),
-                  configuration.getAckDeadline(),
-                  configuration.getNackOnTimeout())));
+      PCollection<T> output =
+          input.apply(
+              org.apache.beam.sdk.io.Read.from(
+                  new UnboundedSolaceSource<>(
+                      initializedQueue,
+                      sempClientFactory,
+                      sessionServiceFactory,
+                      configuration.getMaxNumConnections(),
+                      configuration.getDeduplicateRecords(),
+                      coder,
+                      configuration.getTimestampFn(),
+                      configuration.getWatermarkIdleDurationThreshold(),
+                      configuration.getParseFn(),
+                      configuration.getAckDeadline(),
+                      configuration.getNackOnTimeout(),
+                      configuration.getEnableOpenTelemetryTracing())));
+
+      return output;
     }
 
     @VisibleForTesting
diff --git a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReader.java b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReader.java
index 0bfd9f1c45d9..d4604de1f1eb 100644
--- a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReader.java
+++ b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReader.java
@@ -80,6 +80,7 @@ class UnboundedSolaceReader<T> extends UnboundedReader<T> {
 
   private final Duration ackDeadline;
   private final boolean nackOnTimeout;
+  private final boolean enableOpenTelemetryTracing;
 
   /**
    * Map to track pending checkpoints and their messages. Accessed by both reader
@@ -135,6 +136,7 @@ public UnboundedSolaceReader(UnboundedSolaceSource<T> currentSource) {
     this.ackExecutor = Executors.newFixedThreadPool(4);
     this.ackDeadline = java.time.Duration.ofMillis(currentSource.getAckDeadline().getMillis());
     this.nackOnTimeout = currentSource.getNackOnTimeout();
+    this.enableOpenTelemetryTracing = currentSource.getEnableOpenTelemetryTracing();
   }
 
   private SessionService getSessionService() {
@@ -175,6 +177,9 @@ public boolean advance() {
       return false;
     }
     solaceOriginalRecord = receivedXmlMessage;
+    if (enableOpenTelemetryTracing) {
+      LOG.debug("SolaceIO.Read: OpenTelemetry tracing is enabled for the received message.");
+    }
     solaceMappedRecord = getCurrentSource().getParseFn().apply(receivedXmlMessage);
     receivedMessages.add(receivedXmlMessage);
     messagesReceived.inc();
diff --git a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceSource.java b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceSource.java
index 0ca9bff9ebf9..15fe46f9f078 100644
--- a/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceSource.java
+++ b/sdks/java/io/solace/src/main/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceSource.java
@@ -51,6 +51,7 @@ public class UnboundedSolaceSource<T> extends UnboundedSource<T, SolaceCheckpoin
   private final SerializableFunction<@Nullable BytesXMLMessage, @Nullable T> parseFn;
   private final Duration ackDeadline;
   private final boolean nackOnTimeout;
+  private final boolean enableOpenTelemetryTracing;
 
   public Queue getQueue() {
     return queue;
@@ -84,6 +85,10 @@ public boolean getNackOnTimeout() {
     return nackOnTimeout;
   }
 
+  public boolean getEnableOpenTelemetryTracing() {
+    return enableOpenTelemetryTracing;
+  }
+
   public UnboundedSolaceSource(
       Queue queue,
       SempClientFactory sempClientFactory,
@@ -95,7 +100,8 @@ public UnboundedSolaceSource(
       Duration watermarkIdleDurationThreshold,
       SerializableFunction<@Nullable BytesXMLMessage, @Nullable T> parseFn,
       Duration ackDeadline,
-      boolean nackOnTimeout) {
+      boolean nackOnTimeout,
+      boolean enableOpenTelemetryTracing) {
     this.queue = queue;
     this.sempClientFactory = sempClientFactory;
     this.sessionServiceFactory = sessionServiceFactory;
@@ -107,6 +113,7 @@ public UnboundedSolaceSource(
     this.parseFn = parseFn;
     this.ackDeadline = ackDeadline;
     this.nackOnTimeout = nackOnTimeout;
+    this.enableOpenTelemetryTracing = enableOpenTelemetryTracing;
   }
 
   @Override
@@ -152,7 +159,8 @@ private List<UnboundedSolaceSource<T>> getSolaceSources(
               watermarkIdleDurationThreshold,
               parseFn,
               ackDeadline,
-              nackOnTimeout);
+              nackOnTimeout,
+              enableOpenTelemetryTracing);
       sourceList.add(source);
     }
     return sourceList;
diff --git a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/SolaceIOReadTest.java b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/SolaceIOReadTest.java
index 29e82142f94f..9ffe98bacfb4 100644
--- a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/SolaceIOReadTest.java
+++ b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/SolaceIOReadTest.java
@@ -99,7 +99,8 @@ private static UnboundedSolaceSource<Record> getSource(Read<Record> spec, TestPi
         configuration.getWatermarkIdleDurationThreshold(),
         configuration.getParseFn(),
         configuration.getAckDeadline(),
-        configuration.getNackOnTimeout());
+        configuration.getNackOnTimeout(),
+        configuration.getEnableOpenTelemetryTracing());
   }
 
   @Test
@@ -542,7 +543,8 @@ public void testCheckpointMarkSafety() throws Exception {
   @Test
   public void testDefaultCoder() {
     Coder<SolaceCheckpointMark> coder =
-        new UnboundedSolaceSource<>(null, null, null, 0, false, null, null, null, null, null, false)
+        new UnboundedSolaceSource<>(
+                null, null, null, 0, false, null, null, null, null, null, false, false)
             .getCheckpointMarkCoder();
     CoderProperties.coderSerializable(coder);
   }
diff --git a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReaderTest.java b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReaderTest.java
index 33d463201a5e..8a62b23cc7f3 100644
--- a/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReaderTest.java
+++ b/sdks/java/io/solace/src/test/java/org/apache/beam/sdk/io/solace/read/UnboundedSolaceReaderTest.java
@@ -67,7 +67,8 @@ public void testCheckpointTimeoutDefault_EvictsWithoutNack() throws Exception {
             org.joda.time.Duration.standardSeconds(1), // watermarkIdleDurationThreshold
             input -> SolaceDataUtils.getSolaceRecord("payload_test0", "450"), // parseFn
             org.joda.time.Duration.standardSeconds(30), // ackDeadline
-            false // nackOnTimeout (default)
+            false, // nackOnTimeout (default)
+            false // enableOpenTelemetryTracing (default)
             );
 
     UnboundedSolaceReader<Solace.Record> reader =
@@ -122,7 +123,8 @@ public void testCheckpointTimeout_WithNackEnabled() throws Exception {
             org.joda.time.Duration.standardSeconds(1), // watermarkIdleDurationThreshold
             input -> SolaceDataUtils.getSolaceRecord("payload_test0", "450"), // parseFn
             org.joda.time.Duration.standardSeconds(30), // ackDeadline
-            true // nackOnTimeout
+            true, // nackOnTimeout
+            false // enableOpenTelemetryTracing (default)
             );
 
     UnboundedSolaceReader<Solace.Record> reader =
@@ -188,7 +190,8 @@ public void testCheckpointTimeout_NackExceptionHandledGracefully() throws Except
             org.joda.time.Duration.standardSeconds(1), // watermarkIdleDurationThreshold
             input -> SolaceDataUtils.getSolaceRecord("payload_test0", "450"), // parseFn
             org.joda.time.Duration.standardSeconds(30), // ackDeadline
-            true // nackOnTimeout
+            true, // nackOnTimeout
+            false // enableOpenTelemetryTracing (default)
             );
 
     UnboundedSolaceReader<Solace.Record> reader =