This is an automated email from the ASF dual-hosted git repository.
stankiewicz pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 56c3aa231b5 [OpenTelemetry] Spanner change streams otel configuration
cleanup (#40552)
56c3aa231b5 is described below
commit 56c3aa231b588c8ab69416020a6ae005251fc217
Author: Radosław Stankiewicz <[email protected]>
AuthorDate: Wed Oct 14 11:53:04 2026 +0200
[OpenTelemetry] Spanner change streams otel configuration cleanup (#40552)
* spanner_cdc_traces_cleanup
---
.../apache/beam/sdk/io/gcp/spanner/SpannerIO.java | 26 ++++++++++++----------
.../changestreams/action/ActionFactory.java | 1 +
.../action/QueryChangeStreamAction.java | 7 +++++-
.../dofn/ReadChangeStreamPartitionDoFn.java | 11 +++++++--
.../dofn/ReadChangeStreamPartitionDoFnTest.java | 3 ++-
5 files changed, 32 insertions(+), 16 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 d2ea020c471..4c709963fc6 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
@@ -552,6 +552,7 @@ public class SpannerIO {
.setRealTimeCheckpointInterval(DEFAULT_REAL_TIME_CHECKPOINT_INTERVAL)
.setHeartbeatMillis(DEFAULT_HEARTBEAT_MILLIS)
.setCancelQueryOnHeartbeat(false)
+ .setPerChangeRecordOpenTelemetryTracing(false)
.build();
}
@@ -2044,6 +2045,8 @@ public class SpannerIO {
abstract boolean getCancelQueryOnHeartbeat();
+ abstract boolean getPerChangeRecordOpenTelemetryTracing();
+
abstract Builder toBuilder();
@AutoValue.Builder
@@ -2079,6 +2082,8 @@ public class SpannerIO {
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
@@ -2132,17 +2137,13 @@ public class SpannerIO {
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. */
@@ -2447,7 +2448,8 @@ public class SpannerIO {
actionFactory,
metrics,
getRealTimeCheckpointInterval(),
- getCancelQueryOnHeartbeat());
+ getCancelQueryOnHeartbeat(),
+ 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 575bcc86630..c3bfc78eee5 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 class ActionFactory implements Serializable {
*
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/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 7a5780144ea..dc02e3d0227 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.ErrorCode;
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 class QueryChangeStreamAction {
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(
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 5901f60c9d1..9d45ac08995 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<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,6 +105,7 @@ 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,
@@ -110,7 +113,8 @@ public class ReadChangeStreamPartitionDoFn extends
DoFn<PartitionMetadata, DataC
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 class ReadChangeStreamPartitionDoFn extends
DoFn<PartitionMetadata, DataC
this.realTimeCheckpointInterval = realTimeCheckpointInterval;
this.cancelQueryOnHeartbeat = cancelQueryOnHeartbeat;
this.throughputEstimator = new NullThroughputEstimator<>();
+ this.enablePerChangeRecordOpenTelemetryTracing =
enablePerChangeRecordOpenTelemetryTracing;
}
@GetInitialWatermarkEstimatorState
@@ -236,7 +241,9 @@ public class ReadChangeStreamPartitionDoFn extends
DoFn<PartitionMetadata, DataC
metrics,
isMutableChangeStream,
realTimeCheckpointInterval,
- options.as(SdkHarnessOptions.class).getOpenTelemetry());
+ enablePerChangeRecordOpenTelemetryTracing
+ ? 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 73da42ba48d..2af71539bf5 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 class ReadChangeStreamPartitionDoFnTest {
actionFactory,
metrics,
Duration.standardMinutes(2),
- CANCEL_QUERY_ON_HEARTBEAT);
+ CANCEL_QUERY_ON_HEARTBEAT,
+ false);
doFn.setThroughputEstimator(throughputEstimator);
partition =
