Skip to content

fix(bigquery-jdbc): fix minor issues and run integration tests for otel #13866

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
keshavdandeva merged 6 commits into from
Jul 23, 2026
Merged

fix(bigquery-jdbc): fix minor issues and run integration tests for otel #13866

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
a6a224c
test: testing presubmits
keshavdandeva Jul 22, 2026
359ce44
increase maxAttempts to wait around ~60 seconds
keshavdandeva Jul 22, 2026
bcdee02
chore: fix issues as per feedback
keshavdandeva Jul 22, 2026
1a21a66
address pr feedback
keshavdandeva Jul 22, 2026
11d0614
address feedback nits
keshavdandeva Jul 22, 2026
0f42545
fix ADC support
keshavdandeva Jul 23, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
20 changes: 15 additions & 5 deletions java-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryConnection.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -1101,6 +1101,8 @@ private void closeImpl() throws SQLException {
BigQueryJdbcMdc.clear();
BigQueryJdbcRootLogger.closeConnectionHandler(this.connectionId);
BigQueryJdbcOpenTelemetry.unregisterConnection(this.connectionId);
BigQueryJdbcOpenTelemetry.releaseSdk(this.openTelemetry);
Comment thread
keshavdandeva marked this conversation as resolved.
this.openTelemetry = null;
}
if (exceptionToThrow != null) {
throw exceptionToThrow;
Expand Down Expand Up @@ -1211,13 +1213,21 @@ private OpenTelemetry getOpenTelemetryInstance() {
}

private String resolveEffectiveCredentials() {
String creds = this.gcpTelemetryCredentials;
if (this.gcpTelemetryCredentials != null) {
return this.gcpTelemetryCredentials;
}

String authTypeStr = this.authProperties.get(BigQueryJdbcUrlUtility.OAUTH_TYPE_PROPERTY_NAME);
if (creds == null
&& BigQueryJdbcOAuthUtility.AuthType.GOOGLE_SERVICE_ACCOUNT.name().equals(authTypeStr)) {
return this.authProperties.get(BigQueryJdbcUrlUtility.OAUTH_PVT_KEY_PROPERTY_NAME);
if (!BigQueryJdbcOAuthUtility.AuthType.GOOGLE_SERVICE_ACCOUNT.name().equals(authTypeStr)) {
return null;
}
return creds;

String pvtKey = this.authProperties.get(BigQueryJdbcUrlUtility.OAUTH_PVT_KEY_PROPERTY_NAME);
if (pvtKey != null) {
return pvtKey;
}

return this.authProperties.get(BigQueryJdbcUrlUtility.OAUTH_PVT_KEY_PATH_PROPERTY_NAME);
}

private void validateTraceConfiguration(boolean isTraceEnabled, String effectiveCredentials) {
Expand Down
205 changes: 126 additions & 79 deletions ...bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryJdbcOpenTelemetry.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@
package com.google.cloud.bigquery.jdbc;

import com.google.auth.Credentials;
import com.google.auth.oauth2.GoogleCredentials;
import com.google.cloud.bigquery.exception.BigQueryJdbcRuntimeException;
import com.google.cloud.logging.Logging;
import com.google.cloud.logging.LoggingOptions;
Expand Down Expand Up @@ -44,6 +45,8 @@
import java.util.Objects;
import java.util.concurrent.Callable;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.logging.Handler;
import java.util.logging.Logger;

Expand Down Expand Up @@ -104,7 +107,16 @@ public int hashCode() {
}
}

private static final ConcurrentHashMap<SdkCacheKey, OpenTelemetrySdk> sdkCache =
private static final class CachedSdk {
final OpenTelemetrySdk sdk;
final AtomicInteger refCount = new AtomicInteger(1);

CachedSdk(OpenTelemetrySdk sdk) {
this.sdk = sdk;
}
}
Comment thread
keshavdandeva marked this conversation as resolved.

private static final ConcurrentHashMap<SdkCacheKey, CachedSdk> sdkCache =
new ConcurrentHashMap<>();

static class TelemetryConfig {
Expand All @@ -127,20 +139,6 @@ private BigQueryJdbcOpenTelemetry() {}

static {
ensureGlobalHandlerAttached();
Runtime.getRuntime()
.addShutdownHook(
new Thread(
() -> {
for (OpenTelemetrySdk sdk : sdkCache.values()) {
try {
sdk.close();
} catch (Exception e) {
// Ignore failures during shutdown to ensure all SDKs are attempted to be
// closed. Logging is avoided here because the logging system might have
// already been shut down by the JVM.
}
}
}));
}

public static synchronized void ensureGlobalHandlerAttached() {
Expand Down Expand Up @@ -207,6 +205,44 @@ public static Logging createLoggingClient(
}
}

public static void releaseSdk(OpenTelemetry openTelemetry) {
if (openTelemetry == null
|| openTelemetry == GlobalOpenTelemetry.get()
|| openTelemetry == OpenTelemetry.noop()) {
return;
}

AtomicBoolean shouldClose = new AtomicBoolean(false);
for (Map.Entry<SdkCacheKey, CachedSdk> entry : sdkCache.entrySet()) {
if (entry.getValue().sdk == openTelemetry) {
sdkCache.computeIfPresent(
entry.getKey(),
(k, cachedSdk) -> {
if (cachedSdk.sdk != openTelemetry) {
return cachedSdk;
}
if (cachedSdk.refCount.decrementAndGet() <= 0) {
shouldClose.set(true);
return null;
}
return cachedSdk;
});
Comment thread
keshavdandeva marked this conversation as resolved.
break;
}
}
Comment thread
keshavdandeva marked this conversation as resolved.

if (shouldClose.get() && openTelemetry instanceof OpenTelemetrySdk) {
try {
((OpenTelemetrySdk) openTelemetry).close();
} catch (Exception e) {
// Swallow exceptions so that telemetry shutdown failures (e.g. flush timeouts)
// do not propagate and disrupt the core BigQueryConnection.close() logic.
// Throwing here could prevent the core connection from cleaning up correctly.
LOG.warning("Failed to close OpenTelemetry SDK: %s", e.getMessage());
}
}
}

private static Credentials resolveCredentialsFromString(String credsString) {
Map<String, String> authProperties = new java.util.HashMap<>();
authProperties.put(
Expand Down Expand Up @@ -295,70 +331,81 @@ public static OpenTelemetry getOpenTelemetry(
gcpTelemetryProjectId,
getCredentialsIdentifier(gcpTelemetryCredentials),
enableGcpTraceExporter);
return sdkCache.computeIfAbsent(
key,
k -> {
Map<String, String> props = new HashMap<>();

if (enableGcpTraceExporter) {
props.put(OTEL_TRACES_EXPORTER, EXPORTER_OTLP);
props.put(OTEL_EXPORTER_OTLP_ENDPOINT, OTLP_ENDPOINT_VALUE);
} else {
props.put(OTEL_TRACES_EXPORTER, EXPORTER_NONE);
}

// Logs are handled directly via GCP logging
props.put(OTEL_LOGS_EXPORTER, EXPORTER_NONE);
// Metrics are deferred to a future phase
props.put(OTEL_METRICS_EXPORTER, EXPORTER_NONE);

if (gcpTelemetryProjectId != null) {
props.put(GOOGLE_CLOUD_PROJECT, gcpTelemetryProjectId);
}

// Set safe, generous default limits on attribute value lengths (32KB) to protect
// customers from GCP Cloud Trace 64KB span ingestion failures when logging massive
// exception stack traces or database schema metadata.
// Respect any existing user configuration overrides.
if (!props.containsKey(OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT)) {
props.put(OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, DEFAULT_ATTRIBUTE_LENGTH_LIMIT);
}
if (!props.containsKey(OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT)) {
props.put(OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT, DEFAULT_ATTRIBUTE_LENGTH_LIMIT);
}

AutoConfiguredOpenTelemetrySdk autoConfigured =
AutoConfiguredOpenTelemetrySdk.builder()
.addPropertiesSupplier(() -> props)
.addSpanExporterCustomizer(
(spanExporter, configProperties) -> {
if (gcpTelemetryCredentials == null) {
return spanExporter;
}
try {
Credentials credentials =
resolveCredentialsFromString(gcpTelemetryCredentials);
if (spanExporter instanceof OtlpHttpSpanExporter) {
return ((OtlpHttpSpanExporter) spanExporter)
.toBuilder().setHeaders(() -> getAuthHeaders(credentials)).build();
}
if (spanExporter instanceof OtlpGrpcSpanExporter) {
return ((OtlpGrpcSpanExporter) spanExporter)
.toBuilder().setHeaders(() -> getAuthHeaders(credentials)).build();
}
} catch (Exception e) {
LOG.warning(
e,
"Failed to resolve telemetry credentials. Telemetry will be exported using default OpenTelemetry configuration (custom authentication headers will not be injected).");
}
return spanExporter;
})
.build();

OpenTelemetrySdk sdk = autoConfigured.getOpenTelemetrySdk();

return sdk;
});
return sdkCache.compute(
key,
(k, cachedSdk) -> {
if (cachedSdk != null) {
cachedSdk.refCount.incrementAndGet();
return cachedSdk;
}

Map<String, String> props = new HashMap<>();

if (enableGcpTraceExporter) {
props.put(OTEL_TRACES_EXPORTER, EXPORTER_OTLP);
props.put(OTEL_EXPORTER_OTLP_ENDPOINT, OTLP_ENDPOINT_VALUE);
} else {
props.put(OTEL_TRACES_EXPORTER, EXPORTER_NONE);
}

// Logs are handled directly via GCP logging
props.put(OTEL_LOGS_EXPORTER, EXPORTER_NONE);
// Metrics are deferred to a future phase
props.put(OTEL_METRICS_EXPORTER, EXPORTER_NONE);

if (gcpTelemetryProjectId != null) {
props.put(GOOGLE_CLOUD_PROJECT, gcpTelemetryProjectId);
}

// Set safe, generous default limits on attribute value lengths (32KB) to protect
// customers from GCP Cloud Trace 64KB span ingestion failures when logging massive
// exception stack traces or database schema metadata.
// Respect any existing user configuration overrides.
if (!props.containsKey(OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT)) {
props.put(OTEL_SPAN_ATTRIBUTE_VALUE_LENGTH_LIMIT, DEFAULT_ATTRIBUTE_LENGTH_LIMIT);
}
if (!props.containsKey(OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT)) {
props.put(OTEL_ATTRIBUTE_VALUE_LENGTH_LIMIT, DEFAULT_ATTRIBUTE_LENGTH_LIMIT);
}

AutoConfiguredOpenTelemetrySdk autoConfigured =
AutoConfiguredOpenTelemetrySdk.builder()
.addPropertiesSupplier(() -> props)
.addSpanExporterCustomizer(
(spanExporter, configProperties) -> {
try {
Credentials credentials;
if (gcpTelemetryCredentials != null) {
credentials = resolveCredentialsFromString(gcpTelemetryCredentials);
} else {
credentials = GoogleCredentials.getApplicationDefault();
}
if (spanExporter instanceof OtlpHttpSpanExporter) {
return ((OtlpHttpSpanExporter) spanExporter)
.toBuilder()
.setHeaders(() -> getAuthHeaders(credentials))
.build();
}
if (spanExporter instanceof OtlpGrpcSpanExporter) {
return ((OtlpGrpcSpanExporter) spanExporter)
.toBuilder()
.setHeaders(() -> getAuthHeaders(credentials))
.build();
}
} catch (Exception e) {
LOG.warning(
e,
"Failed to resolve telemetry credentials. Telemetry will be exported using default OpenTelemetry configuration (custom authentication headers will not be injected).");
}
return spanExporter;
})
.build();

OpenTelemetrySdk sdk = autoConfigured.getOpenTelemetrySdk();

return new CachedSdk(sdk);
})
.sdk;
}

/** Gets a Tracer for the JDBC driver instrumentation scope. */
Expand Down
5 changes: 3 additions & 2 deletions java-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/it/ITOpenTelemetryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -306,8 +306,8 @@ private Trace verifyAndFetchTrace(String traceId) throws Exception {

private <T> T pollWithRetry(java.util.concurrent.Callable<T> task) throws InterruptedException {
int attempts = 0;
int maxAttempts = 30; // 30 attempts * 500ms = 15 seconds max delay
long delayMs = 500; // 500ms linear polling
int maxAttempts = 20; // 20 attempts * 3000ms = 60 seconds max delay
long delayMs = 3000; // 3000ms linear polling

while (attempts < maxAttempts) {
attempts++;
Expand All @@ -321,6 +321,7 @@ private <T> T pollWithRetry(java.util.concurrent.Callable<T> task) throws Interr
throw new RuntimeException("Test execution interrupted", e);
} catch (Exception e) {
// Ignore exceptions during remote lookup and retry
e.printStackTrace();
}
if (attempts < maxAttempts) {
Thread.sleep(delayMs);
Expand Down
Loading