feat(bigquery-jdbc): implement automated MDC executor context propagation and decouple daemon threads (googleapis#13374)

b/519201498

This PR implements automated MDC logging context propagation across
threads spawned by execution services in the BigQuery JDBC driver and
decouples JVM-wide background garbage collection monitoring from
statement executors. It also cleans up test execution by routing
unit-test logging files into a dedicated sub-directory.

## Detailed Changes

### 1. MDC Propagation Infrastructure (`BigQueryJdbcMdc.java`)
- Added `MdcThreadPoolExecutor` and `MdcFutureTask` implementation that
automatically intercepts calls to `execute` and `newTaskFor` to capture
the connection's MDC context (`connectionId`) and restore it on worker
threads.
- Added static helper factory methods like `newFixedThreadPool(...)` to
allow creation of MDC-propagating executor pools.

### 2. GC Daemon Decoupling (`BigQueryDaemonPollingTask.java`)
- Decoupled `BigQueryDaemonPollingTask` tasks (which monitor Arrow and
Json references for JVM-wide cleanup) from the connection statement
executor pool.
- Changed task initialization to spawn dedicated daemon threads
(`BigQuery-GC-Daemon-Arrow` and `BigQuery-GC-Daemon-Json`) directly,
ensuring GC work does not consume execution pool resources.

### 3. Unit Testing (`BigQueryJdbcMdcTest.java`)
- Added a full test suite validating transparent MDC connection context
propagation and verifying that contexts are properly cleared from
threads on completion/re-use.

Author: Neenu1995

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

@@ -45,6 +45,7 @@ class BigQueryDaemonPollingTask extends Thread {
   private BigQueryDaemonPollingTask(
       List<BigQueryResultSetFinalizers.ArrowResultSetFinalizer> arrowRsFinalizers,
       ReferenceQueue<BigQueryArrowResultSet> referenceQueueArrowRs) {
+    super("BigQuery-GC-Daemon-Arrow");
     BigQueryDaemonPollingTask.referenceQueueArrowRs = referenceQueueArrowRs;
     BigQueryDaemonPollingTask.arrowRsFinalizers = arrowRsFinalizers;
     setDaemon(true);
@@ -53,6 +54,7 @@ private BigQueryDaemonPollingTask(
   private BigQueryDaemonPollingTask(
       ReferenceQueue<BigQueryJsonResultSet> referenceQueueJsonRs,
       List<BigQueryResultSetFinalizers.JsonResultSetFinalizer> jsonRsFinalizers) {
+    super("BigQuery-GC-Daemon-Json");
     BigQueryDaemonPollingTask.referenceQueueJsonRs = referenceQueueJsonRs;
     BigQueryDaemonPollingTask.jsonRsFinalizers = jsonRsFinalizers;
     setDaemon(true);

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

@@ -16,6 +16,17 @@
 
 package com.google.cloud.bigquery.jdbc;
 
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.Callable;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.RunnableFuture;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.ThreadPoolExecutor;
+import java.util.concurrent.TimeUnit;
+
 /** Lightweight MDC implementation for the BigQuery JDBC driver using InheritableThreadLocal. */
 class BigQueryJdbcMdc {
   private static final InheritableThreadLocal<String> currentConnectionId =
@@ -40,6 +51,110 @@ static void clear() {
     currentConnectionId.remove();
   }
 
+  /**
+   * Creates a new fixed thread pool ExecutorService that automatically propagates MDC connection
+   * context from the submitting thread to the executing thread.
+   */
+  static ExecutorService newFixedThreadPool(int nThreads, ThreadFactory threadFactory) {
+    return new MdcThreadPoolExecutor(
+        nThreads,
+        nThreads,
+        0L,
+        TimeUnit.MILLISECONDS,
+        new LinkedBlockingQueue<>(),
+        new MdcThreadFactory(threadFactory));
+  }
+
+  /**
+   * Creates a new fixed thread pool ExecutorService that automatically propagates MDC connection
+   * context from the submitting thread to the executing thread.
+   */
+  static ExecutorService newFixedThreadPool(int nThreads) {
+    return newFixedThreadPool(nThreads, Executors.defaultThreadFactory());
+  }
+
+  private static class MdcThreadFactory implements ThreadFactory {
+    private final ThreadFactory delegate;
+
+    public MdcThreadFactory(ThreadFactory delegate) {
+      this.delegate = delegate;
+    }
+
+    @Override
+    public Thread newThread(Runnable r) {
+      return delegate.newThread(
+          () -> {
+            clear();
+            r.run();
+          });
+    }
+  }
+
+  private static class MdcThreadPoolExecutor extends ThreadPoolExecutor {
+
+    public MdcThreadPoolExecutor(
+        int corePoolSize,
+        int maximumPoolSize,
+        long keepAliveTime,
+        TimeUnit unit,
+        BlockingQueue<Runnable> workQueue,
+        ThreadFactory threadFactory) {
+      super(corePoolSize, maximumPoolSize, keepAliveTime, unit, workQueue, threadFactory);
+    }
+
+    @Override
+    public void execute(Runnable command) {
+      if (command == null) {
+        throw new NullPointerException();
+      }
+      if (command instanceof MdcFutureTask) {
+        super.execute(command);
+      } else {
+        super.execute(wrap(command));
+      }
+    }
+
+    @Override
+    protected <T> RunnableFuture<T> newTaskFor(Runnable runnable, T value) {
+      return new MdcFutureTask<>(runnable, value, getConnectionId());
+    }
+
+    @Override
+    protected <T> RunnableFuture<T> newTaskFor(Callable<T> callable) {
+      return new MdcFutureTask<>(callable, getConnectionId());
+    }
+
+    private static Runnable wrap(Runnable runnable) {
+      String connectionId = getConnectionId();
+      return () -> {
+        try (MdcCloseable mdc = registerInstance(connectionId)) {
+          runnable.run();
+        }
+      };
+    }
+  }
+
+  private static class MdcFutureTask<V> extends FutureTask<V> {
+    private final String connectionId;
+
+    public MdcFutureTask(Runnable runnable, V result, String connectionId) {
+      super(runnable, result);
+      this.connectionId = connectionId;
+    }
+
+    public MdcFutureTask(Callable<V> callable, String connectionId) {
+      super(callable);
+      this.connectionId = connectionId;
+    }
+
+    @Override
+    public void run() {
+      try (MdcCloseable mdc = registerInstance(connectionId)) {
+        super.run();
+      }
+    }
+  }
+
   /**
    * Functional interface that extends AutoCloseable to avoid throwing checked exceptions in
    * try-with-resources.

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

@@ -112,7 +112,7 @@ public void testUnknownPropertyWarningIsLogged() throws SQLException {
     Connection connection =
         bigQueryDriver.connect(
             "jdbc:bigquery://https://www.googleapis.com/bigquery/v2:443;"
-                + "OAuthType=2;OAuthAccessToken=redactedToken;ProjectId=t;LogLevel=3;"
+                + "OAuthType=2;OAuthAccessToken=redactedToken;ProjectId=t;LogLevel=3;LogPath=logs/;"
                 + "MyUnknownSetting=Value",
             new Properties());
     assertThat(connection.isClosed()).isFalse();
@@ -134,7 +134,7 @@ public void testMalformedUrlExceptionIsLogged() {
         () ->
             bigQueryDriver.connect(
                 "jdbc:bigquery://https://www.googleapis.com/bigquery/v2:443;"
-                    + "OAuthType=2;OAuthAccessToken=redactedToken;ProjectId=t;LogLevel=3;"
+                    + "OAuthType=2;OAuthAccessToken=redactedToken;ProjectId=t;LogLevel=3;LogPath=logs/;"
                     + "MalformedPropertyWithoutEquals",
                 new Properties()));
 

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

@@ -18,9 +18,14 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Future;
+import java.util.concurrent.FutureTask;
+import java.util.concurrent.RunnableFuture;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
 import org.junit.jupiter.api.AfterEach;
@@ -131,4 +136,119 @@ public void testMdcCloseableClearsContext() {
     }
     assertNull(BigQueryJdbcMdc.getConnectionId());
   }
+
+  @Test
+  public void testExecutorPropagatesMdc() throws Exception {
+    BigQueryJdbcMdc.registerInstance("JdbcConnection-Executor-Test");
+    ExecutorService executor = BigQueryJdbcMdc.newFixedThreadPool(2);
+
+    try {
+      // Test Runnable submission
+      CountDownLatch runnableLatch = new CountDownLatch(1);
+      AtomicReference<String> runnableMdcVal = new AtomicReference<>();
+      executor.execute(
+          () -> {
+            runnableMdcVal.set(BigQueryJdbcMdc.getConnectionId());
+            runnableLatch.countDown();
+          });
+      assertTrue(runnableLatch.await(5, TimeUnit.SECONDS));
+      assertEquals("JdbcConnection-Executor-Test", runnableMdcVal.get());
+
+      // Test Callable submission
+      Future<String> callableFuture =
+          executor.submit(
+              () -> {
+                return BigQueryJdbcMdc.getConnectionId();
+              });
+      assertEquals("JdbcConnection-Executor-Test", callableFuture.get(5, TimeUnit.SECONDS));
+
+      // Test context is cleared on worker thread after task completion
+      CountDownLatch cleanupLatch = new CountDownLatch(1);
+      AtomicReference<String> postTaskMdcVal = new AtomicReference<>("initial-non-null");
+      executor.execute(
+          () -> {
+            // Run a task on a potentially reused thread
+            postTaskMdcVal.set(BigQueryJdbcMdc.getConnectionId());
+            cleanupLatch.countDown();
+          });
+      assertTrue(cleanupLatch.await(5, TimeUnit.SECONDS));
+      // Since the main thread has context set, the second task will also capture and set it.
+      // Let's clear the context on the main thread, and submit a task to check if the thread-local
+      // is clean for a thread that has previously executed tasks.
+      BigQueryJdbcMdc.clear();
+      CountDownLatch cleanMainLatch = new CountDownLatch(1);
+      AtomicReference<String> cleanThreadMdcVal = new AtomicReference<>("initial-non-null");
+      executor.execute(
+          () -> {
+            cleanThreadMdcVal.set(BigQueryJdbcMdc.getConnectionId());
+            cleanMainLatch.countDown();
+          });
+      assertTrue(cleanMainLatch.await(5, TimeUnit.SECONDS));
+      // It should be null because the submitting thread had no context set,
+      // and the previous task's close() call cleaned the ThreadLocal.
+      assertNull(cleanThreadMdcVal.get());
+
+    } finally {
+      executor.shutdownNow();
+    }
+  }
+
+  @Test
+  public void testExecutorThrowsNpeOnNullCommand() {
+    ExecutorService executor = BigQueryJdbcMdc.newFixedThreadPool(2);
+    try {
+      assertThrows(NullPointerException.class, () -> executor.execute(null));
+    } finally {
+      executor.shutdownNow();
+    }
+  }
+
+  @Test
+  public void testExecutorWrapsCustomRunnableFuture() throws Exception {
+    BigQueryJdbcMdc.registerInstance("JdbcConnection-CustomFuture-Test");
+    ExecutorService executor = BigQueryJdbcMdc.newFixedThreadPool(2);
+    try {
+      CountDownLatch latch = new CountDownLatch(1);
+      AtomicReference<String> mdcVal = new AtomicReference<>();
+      RunnableFuture<Void> customFuture =
+          new FutureTask<>(
+              () -> {
+                mdcVal.set(BigQueryJdbcMdc.getConnectionId());
+                latch.countDown();
+              },
+              null);
+
+      executor.execute(customFuture);
+      assertTrue(latch.await(5, TimeUnit.SECONDS));
+      assertEquals("JdbcConnection-CustomFuture-Test", mdcVal.get());
+    } finally {
+      executor.shutdownNow();
+    }
+  }
+
+  @Test
+  public void testPoolThreadInheritanceSevered() throws Exception {
+    BigQueryJdbcMdc.registerInstance("JdbcConnection-ParentContext");
+    ExecutorService executor = BigQueryJdbcMdc.newFixedThreadPool(1);
+    try {
+      CountDownLatch initLatch = new CountDownLatch(1);
+      executor.execute(initLatch::countDown);
+      assertTrue(initLatch.await(5, TimeUnit.SECONDS));
+
+      BigQueryJdbcMdc.clear();
+
+      CountDownLatch taskLatch = new CountDownLatch(1);
+      AtomicReference<String> workerMdcVal = new AtomicReference<>("initial-non-null");
+      executor.execute(
+          () -> {
+            workerMdcVal.set(BigQueryJdbcMdc.getConnectionId());
+            taskLatch.countDown();
+          });
+
+      assertTrue(taskLatch.await(5, TimeUnit.SECONDS));
+      assertNull(workerMdcVal.get());
+    } finally {
+      executor.shutdownNow();
+    }
+  }
 }