From f333f5bcc827e9708ad23cf1779194a8a2b9e0b7 Mon Sep 17 00:00:00 2001 From: Radek Stankiewicz Date: Tue, 13 Oct 2026 17:01:43 +0200 Subject: [PATCH 1/2] spanner_cdc_traces_cleanup --- .../beam/sdk/io/gcp/spanner/SpannerIO.java | 23 ++++++++++--------- .../action/QueryChangeStreamAction.java | 7 +++++- .../dofn/ReadChangeStreamPartitionDoFn.java | 11 +++++++-- .../ReadChangeStreamPartitionDoFnTest.java | 3 ++- 4 files changed, 29 insertions(+), 15 deletions(-) diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java index 1d892fb11e4a..95b165886966 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java +++ b/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) + .setEnableOpenTelemetryTracing(false) .build(); } @@ -1952,6 +1953,8 @@ public abstract static class ReadChangeStream abstract boolean getCancelQueryOnHeartbeat(); + abstract boolean getEnableOpenTelemetryTracing(); + abstract Builder toBuilder(); @AutoValue.Builder @@ -1987,6 +1990,8 @@ abstract static class Builder { abstract Builder setClientCertKeyPath(ValueProvider clientCertKeyPath); + abstract Builder setEnableOpenTelemetryTracing(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,12 @@ public ReadChangeStream withDatabaseId(ValueProvider databaseId) { return withSpannerConfig(config.withDatabaseId(databaseId)); } - /** Specifies whether OpenTelemetry tracing is enabled. */ + /** + * Specifies whether OpenTelemetry tracing should start trace and span on each change record. + * For Spanner Client library tracing config, use SpannerConfig (withSpannerConfig). + */ public ReadChangeStream withEnableOpenTelemetryTracing(boolean enableOpenTelemetryTracing) { - return withEnableOpenTelemetryTracing( - ValueProvider.StaticValueProvider.of(enableOpenTelemetryTracing)); - } - - /** Specifies whether OpenTelemetry tracing is enabled. */ - public ReadChangeStream withEnableOpenTelemetryTracing( - ValueProvider enableOpenTelemetryTracing) { - SpannerConfig config = getSpannerConfig(); - return withSpannerConfig(config.withEnableOpenTelemetryTracing(enableOpenTelemetryTracing)); + return toBuilder().setEnableOpenTelemetryTracing(enableOpenTelemetryTracing).build(); } /** Specifies the change stream name. */ @@ -2355,7 +2355,8 @@ && getInclusiveStartAt().toSqlTimestamp().after(getInclusiveEndAt().toSqlTimesta actionFactory, metrics, getRealTimeCheckpointInterval(), - getCancelQueryOnHeartbeat()); + getCancelQueryOnHeartbeat(), + getEnableOpenTelemetryTracing()); final PostProcessingMetricsDoFn postProcessingMetricsDoFn = new PostProcessingMetricsDoFn(metrics); diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java index 7a5780144eab..dc02e3d02275 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/QueryChangeStreamAction.java +++ b/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 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( diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFn.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFn.java index 5901f60c9d18..d935261be2d3 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFn.java +++ b/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(); + this.enableOpenTelemetryTracing = enableOpenTelemetryTracing; } @GetInitialWatermarkEstimatorState @@ -236,7 +241,9 @@ public void setup(PipelineOptions options) { metrics, isMutableChangeStream, realTimeCheckpointInterval, - options.as(SdkHarnessOptions.class).getOpenTelemetry()); + enableOpenTelemetryTracing + ? options.as(SdkHarnessOptions.class).getOpenTelemetry() + : OpenTelemetry.noop()); } /** diff --git a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFnTest.java b/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFnTest.java index 73da42ba48df..2af71539bf5e 100644 --- a/sdks/java/io/google-cloud-platform/src/test/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFnTest.java +++ b/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 = From 22730853546e56251cd10f8e641f8623db5b7984 Mon Sep 17 00:00:00 2001 From: Radek Stankiewicz Date: Wed, 14 Oct 2026 10:58:20 +0200 Subject: [PATCH 2/2] address comments --- .../apache/beam/sdk/io/gcp/spanner/SpannerIO.java | 15 ++++++++------- .../changestreams/action/ActionFactory.java | 1 + .../dofn/ReadChangeStreamPartitionDoFn.java | 10 +++++----- 3 files changed, 14 insertions(+), 12 deletions(-) diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java index 95b165886966..604979f5d085 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/SpannerIO.java @@ -549,7 +549,7 @@ public static ReadChangeStream readChangeStream() { .setRealTimeCheckpointInterval(DEFAULT_REAL_TIME_CHECKPOINT_INTERVAL) .setHeartbeatMillis(DEFAULT_HEARTBEAT_MILLIS) .setCancelQueryOnHeartbeat(false) - .setEnableOpenTelemetryTracing(false) + .setPerChangeRecordOpenTelemetryTracing(false) .build(); } @@ -1953,7 +1953,7 @@ public abstract static class ReadChangeStream abstract boolean getCancelQueryOnHeartbeat(); - abstract boolean getEnableOpenTelemetryTracing(); + abstract boolean getPerChangeRecordOpenTelemetryTracing(); abstract Builder toBuilder(); @@ -1990,7 +1990,7 @@ abstract static class Builder { abstract Builder setClientCertKeyPath(ValueProvider clientCertKeyPath); - abstract Builder setEnableOpenTelemetryTracing(boolean enableOpenTelemetryTracing); + abstract Builder setPerChangeRecordOpenTelemetryTracing(boolean enableOpenTelemetryTracing); /** * When caught up to real-time, checkpoint processing of change stream this often. This sets a @@ -2047,10 +2047,11 @@ public ReadChangeStream withDatabaseId(ValueProvider databaseId) { /** * Specifies whether OpenTelemetry tracing should start trace and span on each change record. - * For Spanner Client library tracing config, use SpannerConfig (withSpannerConfig). + * Spanner Client library tracing can be enabled via the SpannerConfig passed to + * withSpannerConfig. */ - public ReadChangeStream withEnableOpenTelemetryTracing(boolean enableOpenTelemetryTracing) { - return toBuilder().setEnableOpenTelemetryTracing(enableOpenTelemetryTracing).build(); + public ReadChangeStream withPerChangeRecordOpenTelemetryTracing(boolean openTelemetryTracing) { + return toBuilder().setPerChangeRecordOpenTelemetryTracing(openTelemetryTracing).build(); } /** Specifies the change stream name. */ @@ -2356,7 +2357,7 @@ && getInclusiveStartAt().toSqlTimestamp().after(getInclusiveEndAt().toSqlTimesta metrics, getRealTimeCheckpointInterval(), getCancelQueryOnHeartbeat(), - getEnableOpenTelemetryTracing()); + getPerChangeRecordOpenTelemetryTracing()); final PostProcessingMetricsDoFn postProcessingMetricsDoFn = new PostProcessingMetricsDoFn(metrics); diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/ActionFactory.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/ActionFactory.java index 575bcc866303..c3bfc78eee56 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/action/ActionFactory.java +++ b/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( diff --git a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFn.java b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFn.java index d935261be2d3..9d45ac089950 100644 --- a/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFn.java +++ b/sdks/java/io/google-cloud-platform/src/main/java/org/apache/beam/sdk/io/gcp/spanner/changestreams/dofn/ReadChangeStreamPartitionDoFn.java @@ -78,7 +78,7 @@ public class ReadChangeStreamPartitionDoFn extends DoFn(); - this.enableOpenTelemetryTracing = enableOpenTelemetryTracing; + this.enablePerChangeRecordOpenTelemetryTracing = enablePerChangeRecordOpenTelemetryTracing; } @GetInitialWatermarkEstimatorState @@ -241,7 +241,7 @@ public void setup(PipelineOptions options) { metrics, isMutableChangeStream, realTimeCheckpointInterval, - enableOpenTelemetryTracing + enablePerChangeRecordOpenTelemetryTracing ? options.as(SdkHarnessOptions.class).getOpenTelemetry() : OpenTelemetry.noop()); }