feat(bqjdbc): Implement Correlated OpenTelemetry Logging Bridge

java-bigquery/google-cloud-bigquery-jdbc/pom.xml

@@ -182,6 +182,11 @@
       <artifactId>google-cloud-bigquerystorage</artifactId>
       <version>3.28.0-SNAPSHOT</version><!-- {x-version-update:google-cloud-bigquerystorage:current} -->
     </dependency>
+    <dependency>
+      <groupId>com.google.cloud</groupId>
+      <artifactId>google-cloud-logging</artifactId>
+      <version>3.33.0-SNAPSHOT</version><!-- {x-version-update:google-cloud-logging:current} -->
+    </dependency>
     <dependency>
       <groupId>com.google.http-client</groupId>
       <artifactId>google-http-client-apache-v5</artifactId>

java-bigquery/google-cloud-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryConnection.java

@@ -954,6 +954,7 @@ private void closeImpl() throws SQLException {
     } finally {
       BigQueryJdbcMdc.removeInstance(this);
       BigQueryJdbcRootLogger.closeConnectionHandler(this.connectionId);
+      BigQueryJdbcOpenTelemetry.unregisterConnection(this.connectionId);
     }
     this.isClosed = true;
   }
@@ -1056,6 +1057,12 @@ private BigQuery getBigQueryConnection() {
     OpenTelemetry openTelemetry =
         BigQueryJdbcOpenTelemetry.getOpenTelemetry(
             this.enableGcpTraceExporter, this.enableGcpLogExporter, this.customOpenTelemetry);
+
+    if (this.enableGcpLogExporter || this.customOpenTelemetry != null) {
+      BigQueryJdbcOpenTelemetry.registerConnection(
+          this.connectionId, openTelemetry, null, this.enableGcpLogExporter);
+    }
+
     if (this.enableGcpTraceExporter || this.customOpenTelemetry != null) {
       this.tracer = BigQueryJdbcOpenTelemetry.getTracer(openTelemetry);
       bigQueryOptions.setOpenTelemetryTracer(this.tracer);

java-bigquery/google-cloud-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryDriver.java

@@ -157,7 +157,9 @@ public Connection connect(String url, Properties info) throws SQLException {
         if (logPath == null) {
           logPath = System.getenv(BigQueryJdbcUrlUtility.LOG_PATH_ENV_VAR);
         }
-        if (logPath == null) {
+
+        // Fallback to default path only if not specified and not in Cloud-Only mode
+        if (logPath == null && !ds.getEnableGcpLogExporter()) {
           logPath = BigQueryJdbcUrlUtility.DEFAULT_LOG_PATH;
         }
 

java-bigquery/google-cloud-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryJdbcOpenTelemetry.java

@@ -16,15 +16,77 @@
 
 package com.google.cloud.bigquery.jdbc;
 
+import com.google.cloud.logging.Logging;
 import io.opentelemetry.api.OpenTelemetry;
 import io.opentelemetry.api.trace.Tracer;
+import java.util.Collection;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.logging.Handler;
+import java.util.logging.Logger;
 
 public class BigQueryJdbcOpenTelemetry {
 
   static final String INSTRUMENTATION_SCOPE_NAME = "com.google.cloud.bigquery.jdbc";
+  static final String BIGQUERY_NAMESPACE = "com.google.cloud.bigquery";
+  public static final String CONNECTION_ID_BAGGAGE_KEY = "jdbc.connection_id";
+
+  static class TelemetryConfig {
+    final OpenTelemetry openTelemetry;
+    final Logging loggingClient;
+    final boolean useDirectGcpLogging;
+
+    TelemetryConfig(
+        OpenTelemetry openTelemetry, Logging loggingClient, boolean useDirectGcpLogging) {
+      this.openTelemetry = openTelemetry;
+      this.loggingClient = loggingClient;
+      this.useDirectGcpLogging = useDirectGcpLogging;
+    }
+  }
+
+  private static final ConcurrentHashMap<String, TelemetryConfig> connectionConfigs =
+      new ConcurrentHashMap<>();
 
   private BigQueryJdbcOpenTelemetry() {}
 
+  static {
+    ensureGlobalHandlerAttached();
+  }
+
+  public static void ensureGlobalHandlerAttached() {
+    Logger logger = Logger.getLogger(BIGQUERY_NAMESPACE);
+    boolean present = false;
+    for (Handler h : logger.getHandlers()) {
+      if (h instanceof OpenTelemetryJulHandler) {
+        present = true;
+        break;
+      }
+    }
+    if (!present) {
+      logger.addHandler(new OpenTelemetryJulHandler());
+    }
+  }
+
+  public static void registerConnection(
+      String connectionId,
+      OpenTelemetry openTelemetry,
+      Logging loggingClient,
+      boolean useDirectGcpLogging) {
+    connectionConfigs.put(
+        connectionId, new TelemetryConfig(openTelemetry, loggingClient, useDirectGcpLogging));
+  }
+
+  public static void unregisterConnection(String connectionId) {
+    connectionConfigs.remove(connectionId);
+  }
+
+  public static TelemetryConfig getConnectionConfig(String connectionId) {
+    return connectionConfigs.get(connectionId);
+  }
+
+  public static Collection<TelemetryConfig> getRegisteredConfigs() {
+    return connectionConfigs.values();
+  }
+
   /**
    * Initializes or returns the OpenTelemetry instance based on hybrid logic. Prefer
    * customOpenTelemetry if provided; fallback to an auto-configured GCP exporter if requested.

java-bigquery/google-cloud-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/BigQueryJdbcRootLogger.java

@@ -138,15 +138,17 @@ public static Logger getRootLogger() {
 
   public static void setLevel(Level level, String logPath) throws IOException {
     if (level != Level.OFF) {
-      setPath(logPath, level);
-      logger.setLevel(level);
+      if (logPath != null) {
+        setPath(logPath, level);
+      }
     } else {
       for (Handler h : logger.getHandlers()) {
         h.close();
         logger.removeHandler(h);
       }
       fileHandler = null;
     }
+    logger.setLevel(level);
   }
 
   static void setPath(String logPath, Level level) {

java-bigquery/google-cloud-bigquery-jdbc/src/main/java/com/google/cloud/bigquery/jdbc/OpenTelemetryJulHandler.java

@@ -0,0 +1,186 @@
+/*
+ * Copyright 2026 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *       http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.cloud.bigquery.jdbc;
+
+import com.google.cloud.logging.LogEntry;
+import com.google.cloud.logging.Logging;
+import com.google.cloud.logging.Payload;
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.api.baggage.Baggage;
+import io.opentelemetry.api.common.AttributeKey;
+import io.opentelemetry.api.logs.LogRecordBuilder;
+import io.opentelemetry.api.logs.Logger;
+import io.opentelemetry.api.logs.Severity;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanContext;
+import io.opentelemetry.context.Context;
+import java.time.Instant;
+import java.util.Collections;
+import java.util.logging.Handler;
+import java.util.logging.Level;
+import java.util.logging.LogRecord;
+
+/**
+ * Custom logging handler that bridges java.util.logging records to OpenTelemetry or Google Cloud
+ * Logging. Extracts TraceId, SpanId, and Connection UUID from context.
+ */
+public class OpenTelemetryJulHandler extends Handler {
+
+  public OpenTelemetryJulHandler() {}
+
+  @Override
+  public void publish(LogRecord record) {
+    if (!isLoggable(record)) {
+      return;
+    }
+
+    try {
+      // Extract connection ID from baggage
+      String connectionId =
+          Baggage.fromContext(Context.current())
+              .getEntryValue(BigQueryJdbcOpenTelemetry.CONNECTION_ID_BAGGAGE_KEY);
+
+      // Fallback to MDC if not in baggage (if MDC is available and used)
+      if (connectionId == null) {
+        connectionId = BigQueryJdbcMdc.getConnectionId();
+      }
+
+      if (connectionId == null) {
+        return;
+      }
+
+      BigQueryJdbcOpenTelemetry.TelemetryConfig config =
+          BigQueryJdbcOpenTelemetry.getConnectionConfig(connectionId);
+      if (config == null) {
+        return;
+      }
+
+      if (config.useDirectGcpLogging && config.loggingClient != null) {
+        publishToGcp(record, connectionId, config.loggingClient);
+      } else if (config.openTelemetry != null) {
+        publishToOTel(record, connectionId, config.openTelemetry);
+      }
+    } catch (Throwable t) {
+      // Ignore exceptions to prevent breaking application logging or other handlers
+    }
+  }
+
+  private void publishToGcp(LogRecord record, String connectionId, Logging loggingClient) {
+    Context context = Context.current();
+    SpanContext spanContext = Span.fromContext(context).getSpanContext();
+    String traceId = spanContext.isValid() ? spanContext.getTraceId() : null;
+    String spanId = spanContext.isValid() ? spanContext.getSpanId() : null;
+
+    // TODO(b/491238299): May require refinement for structured logging or error handling
+
+    LogEntry.Builder builder =
+        LogEntry.newBuilder(Payload.StringPayload.of(formatMessage(record)))
+            .setSeverity(mapGcpSeverity(record.getLevel()))
+            .setTimestamp(record.getMillis());
+
+    if (traceId != null) {
+      builder.setTrace(traceId);
+    }
+    if (spanId != null) {
+      builder.setSpanId(spanId);
+    }
+    if (connectionId != null) {
+      builder.addLabel(BigQueryJdbcOpenTelemetry.CONNECTION_ID_BAGGAGE_KEY, connectionId);
+    }
+
+    loggingClient.write(Collections.singleton(builder.build()));
+  }
+
+  private com.google.cloud.logging.Severity mapGcpSeverity(Level level) {
+    if (level == Level.SEVERE) return com.google.cloud.logging.Severity.ERROR;
+    if (level == Level.WARNING) return com.google.cloud.logging.Severity.WARNING;
+    if (level == Level.INFO) return com.google.cloud.logging.Severity.INFO;
+    if (level == Level.CONFIG) return com.google.cloud.logging.Severity.INFO;
+    if (level == Level.FINE) return com.google.cloud.logging.Severity.DEBUG;
+    return com.google.cloud.logging.Severity.DEBUG;
+  }
+
+  private void publishToOTel(LogRecord record, String connectionId, OpenTelemetry openTelemetry) {
+    String loggerName = record.getLoggerName();
+    Logger logger =
+        openTelemetry
+            .getLogsBridge()
+            .get(
+                loggerName != null
+                    ? loggerName
+                    : BigQueryJdbcOpenTelemetry.INSTRUMENTATION_SCOPE_NAME);
+
+    LogRecordBuilder builder =
+        logger
+            .logRecordBuilder()
+            .setBody(formatMessage(record))
+            .setSeverity(mapSeverity(record.getLevel()))
+            .setTimestamp(Instant.ofEpochMilli(record.getMillis()))
+            .setContext(Context.current());
+
+    if (connectionId != null) {
+      builder.setAttribute(
+          AttributeKey.stringKey(BigQueryJdbcOpenTelemetry.CONNECTION_ID_BAGGAGE_KEY),
+          connectionId);
+    }
+
+    builder.emit();
+  }
+
+  private Severity mapSeverity(Level level) {
+    if (level == Level.SEVERE) return Severity.ERROR;
+    if (level == Level.WARNING) return Severity.WARN;
+    if (level == Level.INFO) return Severity.INFO;
+    if (level == Level.CONFIG) return Severity.INFO;
+    if (level == Level.FINE) return Severity.DEBUG;
+    if (level == Level.FINER) return Severity.TRACE;
+    if (level == Level.FINEST) return Severity.TRACE;
+    return Severity.TRACE;
+  }
+
+  private String formatMessage(LogRecord record) {
+    String message = record.getMessage();
+    Object[] params = record.getParameters();
+    if (params != null && params.length > 0) {
+      try {
+        return java.text.MessageFormat.format(message, params);
+      } catch (IllegalArgumentException e) {
+        return message;
+      }
+    }
+    return message;
+  }
+
+  @Override
+  public void flush() {
+    for (BigQueryJdbcOpenTelemetry.TelemetryConfig config :
+        BigQueryJdbcOpenTelemetry.getRegisteredConfigs()) {
+      if (config.useDirectGcpLogging && config.loggingClient != null) {
+        try {
+          config.loggingClient.flush();
+        } catch (Exception e) {
+          // Ignore failures during flush to protect other connections
+        }
+      }
+    }
+  }
+
+  @Override
+  public void close() throws SecurityException {
+    // TODO(b/491238299): Implement with gcp exporter logic
+  }
+}

java-bigquery/google-cloud-bigquery-jdbc/src/test/java/com/google/cloud/bigquery/jdbc/BigQueryJdbcLoggingBaseTest.java

@@ -34,6 +34,7 @@ public abstract class BigQueryJdbcLoggingBaseTest extends BigQueryJdbcBaseTest {
   @BeforeEach
   public void setUpLogValidator() {
     logger = BigQueryJdbcRootLogger.getRootLogger();
+    logger.setLevel(java.util.logging.Level.ALL);
     capturedLogs.clear();
     threadId = Thread.currentThread().getId();
     handler =