[OpenTelemetry] Spanner change streams otel configuration cleanup

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java

@@ -549,6 +549,7 @@ public static ReadChangeStream readChangeStream() {
         .setRealTimeCheckpointInterval(DEFAULT_REAL_TIME_CHECKPOINT_INTERVAL)
         .setHeartbeatMillis(DEFAULT_HEARTBEAT_MILLIS)
         .setCancelQueryOnHeartbeat(false)
+        .setPerChangeRecordOpenTelemetryTracing(false)
         .build();
   }
 
@@ -1952,6 +1953,8 @@ public abstract static class ReadChangeStream
 
     abstract boolean getCancelQueryOnHeartbeat();
 
+    abstract boolean getPerChangeRecordOpenTelemetryTracing();
+
     abstract Builder toBuilder();
 
     @AutoValue.Builder
@@ -1987,6 +1990,8 @@ abstract static class Builder {
 
       abstract Builder setClientCertKeyPath(ValueProvider<String> clientCertKeyPath);
 
+      abstract Builder setPerChangeRecordOpenTelemetryTracing(boolean enableOpenTelemetryTracing);
+
       /**
        * When caught up to real-time, checkpoint processing of change stream this often. This sets a
        * bound on latency of processing if a steady trickle of elements prevents the heartbeat
@@ -2040,17 +2045,13 @@ public ReadChangeStream withDatabaseId(ValueProvider<String> databaseId) {
       return withSpannerConfig(config.withDatabaseId(databaseId));
     }
 
-    /** Specifies whether OpenTelemetry tracing is enabled. */
-    public ReadChangeStream withEnableOpenTelemetryTracing(boolean enableOpenTelemetryTracing) {
-      return withEnableOpenTelemetryTracing(
-          ValueProvider.StaticValueProvider.of(enableOpenTelemetryTracing));
-    }
-
-    /** Specifies whether OpenTelemetry tracing is enabled. */
-    public ReadChangeStream withEnableOpenTelemetryTracing(
-        ValueProvider<Boolean> enableOpenTelemetryTracing) {
-      SpannerConfig config = getSpannerConfig();
-      return withSpannerConfig(config.withEnableOpenTelemetryTracing(enableOpenTelemetryTracing));
+    /**
+     * Specifies whether OpenTelemetry tracing should start trace and span on each change record.
+     * Spanner Client library tracing can be enabled via the SpannerConfig passed to
+     * withSpannerConfig.
+     */
+    public ReadChangeStream withPerChangeRecordOpenTelemetryTracing(boolean openTelemetryTracing) {
+      return toBuilder().setPerChangeRecordOpenTelemetryTracing(openTelemetryTracing).build();
     }
 
     /** Specifies the change stream name. */
@@ -2355,7 +2356,8 @@ && getInclusiveStartAt().toSqlTimestamp().after(getInclusiveEndAt().toSqlTimesta
               actionFactory,
               metrics,
               getRealTimeCheckpointInterval(),
-              getCancelQueryOnHeartbeat());
+              getCancelQueryOnHeartbeat(),
+              getPerChangeRecordOpenTelemetryTracing());
       final PostProcessingMetricsDoFn postProcessingMetricsDoFn =
           new PostProcessingMetricsDoFn(metrics);
 

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/ActionFactory.java

@@ -177,6 +177,7 @@ public synchronized PartitionEventRecordAction partitionEventRecordAction(
    *     org.apache.beam.sdk.io.gcp.spanner.changestreams.model.PartitionEventRecord}s
    * @param metrics metrics gathering class
    * @param realTimeCheckpointInterval the duration added to current time for the end timestamp
+   * @param openTelemetry instance for trace reporting
    * @return single instance of the {@link QueryChangeStreamAction}
    */
   public synchronized QueryChangeStreamAction queryChangeStreamAction(

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java

@@ -24,6 +24,7 @@
 import com.google.cloud.spanner.SpannerException;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanKind;
 import io.opentelemetry.api.trace.Tracer;
 import io.opentelemetry.context.Scope;
 import java.util.List;
@@ -250,7 +251,11 @@ public ProcessContinuation run(
         Optional<ProcessContinuation> maybeContinuation;
         for (final ChangeStreamRecord record : records) {
           if (record instanceof DataChangeRecord) {
-            Span span = getTracer().spanBuilder("DataChangeRecord.run").startSpan();
+            Span span =
+                getTracer()
+                    .spanBuilder("SpannerIO.DataChangeRecord")
+                    .setSpanKind(SpanKind.CONSUMER)
+                    .startSpan();
             try (Scope ignored = span.makeCurrent()) {
               maybeContinuation =
                   dataChangeRecordAction.run(

sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFn.java

@@ -17,6 +17,7 @@
  */
 package org.apache.beam.sdk.io.gcp.spanner.changestreams.dofn;
 
+import io.opentelemetry.api.OpenTelemetry;
 import java.io.Serializable;
 import java.math.BigDecimal;
 import org.apache.beam.sdk.io.gcp.spanner.changestreams.ChangeStreamMetrics;
@@ -77,6 +78,7 @@ public class ReadChangeStreamPartitionDoFn extends DoFn<PartitionMetadata, DataC
   private final ChangeStreamMetrics metrics;
   private final boolean isMutableChangeStream;
   private final boolean cancelQueryOnHeartbeat;
+  private final boolean enablePerChangeRecordOpenTelemetryTracing;
 
   /**
    * Needs to be set through the {@link
@@ -103,14 +105,16 @@ public class ReadChangeStreamPartitionDoFn extends DoFn<PartitionMetadata, DataC
    * @param metrics the {@link ChangeStreamMetrics} to emit partition related metrics
    * @param realTimeCheckpointInterval duration to be used for the next end timestamp
    * @param cancelQueryOnHeartbeat flag to improve low latency checkpointing
+   * @param enablePerChangeRecordOpenTelemetryTracing flag to add change records traces
    */
   public ReadChangeStreamPartitionDoFn(
       DaoFactory daoFactory,
       MapperFactory mapperFactory,
       ActionFactory actionFactory,
       ChangeStreamMetrics metrics,
       Duration realTimeCheckpointInterval,
-      boolean cancelQueryOnHeartbeat) {
+      boolean cancelQueryOnHeartbeat,
+      boolean enablePerChangeRecordOpenTelemetryTracing) {
     this.daoFactory = daoFactory;
     this.actionFactory = actionFactory;
     this.mapperFactory = mapperFactory;
@@ -119,6 +123,7 @@ public ReadChangeStreamPartitionDoFn(
     this.realTimeCheckpointInterval = realTimeCheckpointInterval;
     this.cancelQueryOnHeartbeat = cancelQueryOnHeartbeat;
     this.throughputEstimator = new NullThroughputEstimator<>();
+    this.enablePerChangeRecordOpenTelemetryTracing = enablePerChangeRecordOpenTelemetryTracing;
   }
 
   @GetInitialWatermarkEstimatorState
@@ -236,7 +241,9 @@ public void setup(PipelineOptions options) {
             metrics,
             isMutableChangeStream,
             realTimeCheckpointInterval,
-            options.as(SdkHarnessOptions.class).getOpenTelemetry());
+            enablePerChangeRecordOpenTelemetryTracing
+                ? options.as(SdkHarnessOptions.class).getOpenTelemetry()
+                : OpenTelemetry.noop());
   }
 
   /**

sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFnTest.java

@@ -113,7 +113,8 @@ public void setUp() {
             actionFactory,
             metrics,
             Duration.standardMinutes(2),
-            CANCEL_QUERY_ON_HEARTBEAT);
+            CANCEL_QUERY_ON_HEARTBEAT,
+            false);
     doFn.setThroughputEstimator(throughputEstimator);
 
     partition =