[java-bigquerystorage] Accept an OpenTelemetry instance per stream writer instead of the cached GlobalOpenTelemetry

Is your feature request related to a problem? Please describe.

The Storage Write client's OpenTelemetry metrics (StreamWriter.Builder.setEnableOpenTelemetry(true)) can only be sent to the process-wide GlobalOpenTelemetry.
TelemetryMetrics takes its MeterProvider from Singletons.getOpenTelemetry(), which caches GlobalOpenTelemetry.get() in a static field on first use and has no setter:

google-cloud-java/java-bigquerystorage/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/Singletons.java

Lines 30 to 36 in 165f02e

    
           public static OpenTelemetry getOpenTelemetry() { 
        
             if (openTelemetry == null) { 
        
               openTelemetry = GlobalOpenTelemetry.get(); 
        
               log.info("BigQueryStorage initialized Open Telemetry"); 
        
             } 
        
             return openTelemetry; 
        
           }

google-cloud-java/java-bigquerystorage/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/TelemetryMetrics.java

Lines 136 to 139 in 165f02e

    
           MeterProvider meterProvider = Singletons.getOpenTelemetry().getMeterProvider(); 
        
           writeMeter = 
        
               meterProvider 
        
                   .meterBuilder("com.google.cloud.bigquery.storage.v1.write")

This makes the metrics hard to use from a framework that hosts many writers in one JVM.
We maintain an Apache Flink connector for BigQuery (flink-gcp/flink-connector-gcp) and want to forward each writer's metrics into Flink's own metric system, which then exports them to whatever reporter the deployment uses (Prometheus, Datadog, OTLP).
With only the global instance:

several jobs or subtasks share one TaskManager JVM, so their writers cannot report to separate meter providers;
a library cannot call GlobalOpenTelemetry.set(...) without taking that global over from the application;
when the client is shaded into a connector jar, io.opentelemetry is relocated, so the relocated global is not one the application or a Java agent can set at all.

Describe the solution you'd like

An optional per-writer OpenTelemetry (or MeterProvider) on the writer builders, for example StreamWriter.Builder.setOpenTelemetry(OpenTelemetry), with the same setter on JsonStreamWriter.Builder and SchemaAwareStreamWriter.Builder.
When it is unset, the current GlobalOpenTelemetry behaviour would stay as it is.

Other clients in this repository already offer this:

Spanner: SpannerOptions.Builder.setClientMetricsProvider(MetricsProvider) (feat(spanner): support user-provided OpenTelemetry for client metrics export #13741).
Bigtable: BigtableDataSettings.Builder.setMetricsProvider(MetricsProvider), which accepts a CustomOpenTelemetryMetricsProvider.

Describe alternatives you've considered

GlobalOpenTelemetry.set(...) once per JVM: it cannot separate the writers of different jobs, and it does not reach a relocated copy of the client.
Reading the meters reflectively, or replacing Singletons: neither is a supported surface.

Additional context

Libraries-Bom: 26.90.0
- google-cloud-bigquerystorage 3.35.0
The same code is on main at 165f02e.

One addition, found after filing: a per-writer setter as proposed above would not reach a writer built with setEnableConnectionPool(true).

In pool mode the ConnectionWorkerPool is shared per (location, credentials) across every StreamWriter in the JVM, and it is built from the builder of whichever writer arrives first:

google-cloud-java/java-bigquerystorage/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/StreamWriter.java

Lines 346 to 362 in f01245a

    
           this.singleConnectionOrConnectionPool = 
        
               SingleConnectionOrConnectionPool.ofConnectionPool( 
        
                   connectionPoolMap.computeIfAbsent( 
        
                       ConnectionPoolKey.create( 
        
                           location, 
        
                           credentialsProvider != null ? credentialsProvider.getCredentials() : null), 
        
                       (key) -> { 
        
                         return new ConnectionWorkerPool( 
        
                             builder.maxInflightRequest, 
        
                             builder.maxInflightBytes, 
        
                             builder.maxRetryDuration, 
        
                             builder.limitExceededBehavior, 
        
                             builder.compressorName, 
        
                             client.getSettings(), 
        
                             builder.retrySettings, 
        
                             builder.enableRequestProfiler, 
        
                             builder.enableOpenTelemetry);

Its connections are created with that pool-wide enableOpenTelemetry flag, and each ConnectionWorker owns one TelemetryMetrics keyed by the worker's own writer_id, not by the StreamWriter that issued a request:

google-cloud-java/java-bigquerystorage/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorkerPool.java

Lines 408 to 424 in f01245a

    
           ConnectionWorker connectionWorker = 
        
               new ConnectionWorker( 
        
                   streamName, 
        
                   location, 
        
                   writeSchema, 
        
                   maxInflightRequests, 
        
                   maxInflightBytes, 
        
                   maxRetryDuration, 
        
                   limitExceededBehavior, 
        
                   clientId, 
        
                   traceId, 
        
                   compressorName, 
        
                   clientSettings, 
        
                   retrySettings, 
        
                   enableRequestProfiler, 
        
                   enableOpenTelemetry, 
        
                   /* isMultiplexing= */ true);

google-cloud-java/java-bigquerystorage/google-cloud-bigquerystorage/src/main/java/com/google/cloud/bigquery/storage/v1/ConnectionWorker.java

Lines 677 to 678 in f01245a

    
           this.telemetryMetrics = 
        
               new TelemetryMetrics(this, enableOpenTelemetry, getTableName(), writerId, fullTraceId);

So if the instance were captured the way enableOpenTelemetry is today, the first writer's OpenTelemetry would receive the measurements of every writer sharing the pool, and the recording side could not tell them apart.
For our connector the default-stream path uses the pool, so only its buffered-stream path (one connection per writer) would benefit from the setter alone.

For the pool case, it would help if measurements of a request (append_requests_acked, append_rows_acked, append_request_bytes_acked, network_response_latency) were recorded against the OpenTelemetry of the StreamWriter that appended it.
Connection-level instruments (connection_start_count, connection_end_count, the two gauges) belong to the pool, so a pool-level provider, or documenting that they go to the first writer's instance, would be enough there.

The line references above are on main at f01245a.

Navigation Menu

Repository navigation

[java-bigquerystorage] Accept an OpenTelemetry instance per stream writer instead of the cached GlobalOpenTelemetry #14765

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

	public static OpenTelemetry getOpenTelemetry() {
	if (openTelemetry == null) {
	openTelemetry = GlobalOpenTelemetry.get();
	log.info("BigQueryStorage initialized Open Telemetry");
	}
	return openTelemetry;
	}

	MeterProvider meterProvider = Singletons.getOpenTelemetry().getMeterProvider();
	writeMeter =
	meterProvider
	.meterBuilder("com.google.cloud.bigquery.storage.v1.write")

	this.singleConnectionOrConnectionPool =
	SingleConnectionOrConnectionPool.ofConnectionPool(
	connectionPoolMap.computeIfAbsent(
	ConnectionPoolKey.create(
	location,
	credentialsProvider != null ? credentialsProvider.getCredentials() : null),
	(key) -> {
	return new ConnectionWorkerPool(
	builder.maxInflightRequest,
	builder.maxInflightBytes,
	builder.maxRetryDuration,
	builder.limitExceededBehavior,
	builder.compressorName,
	client.getSettings(),
	builder.retrySettings,
	builder.enableRequestProfiler,
	builder.enableOpenTelemetry);

	ConnectionWorker connectionWorker =
	new ConnectionWorker(
	streamName,
	location,
	writeSchema,
	maxInflightRequests,
	maxInflightBytes,
	maxRetryDuration,
	limitExceededBehavior,
	clientId,
	traceId,
	compressorName,
	clientSettings,
	retrySettings,
	enableRequestProfiler,
	enableOpenTelemetry,
	/* isMultiplexing= */ true);

	this.telemetryMetrics =
	new TelemetryMetrics(this, enableOpenTelemetry, getTableName(), writerId, fullTraceId);

[java-bigquerystorage] Accept an OpenTelemetry instance per stream writer instead of the cached GlobalOpenTelemetry #14765

Description

Activity

laughingman7743 commented on Oct 28, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions