set maxThreadPoolSize instead of coreThreadPoolSize

Author: Neenu1995

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

@@ -60,13 +60,16 @@ static void clear() {
    * 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));
+    MdcThreadPoolExecutor executor =
+        new MdcThreadPoolExecutor(
+            nThreads,
+            nThreads,
+            60L,
+            TimeUnit.SECONDS,
+            new LinkedBlockingQueue<>(),
+            new MdcThreadFactory(threadFactory));
+    executor.allowCoreThreadTimeOut(true);
+    return executor;
   }
 
   /**
@@ -136,18 +139,18 @@ public MdcThreadPoolExecutor(
     private final AtomicBoolean warningLogged = new AtomicBoolean(false);
 
     private void monitorQueueSaturation(int queueSize) {
-      int corePoolSize = getCorePoolSize();
-      // Warn when queue size is >= corePoolSize * 5, with a minimum of 10 tasks to avoid false
+      int maxPoolSize = getMaximumPoolSize();
+      // Warn when queue size is >= maxPoolSize * 5, with a minimum of 10 tasks to avoid false
       // alerts for tiny pools
-      int warnThreshold = Math.max(10, corePoolSize * 5);
-      // Recovery reset threshold is corePoolSize * 2, with a minimum of 4 tasks
-      int recoveryThreshold = Math.max(4, corePoolSize * 2);
+      int warnThreshold = Math.max(10, maxPoolSize * 5);
+      // Recovery reset threshold is maxPoolSize * 2, with a minimum of 4 tasks
+      int recoveryThreshold = Math.max(4, maxPoolSize * 2);
 
       if (queueSize >= warnThreshold) {
         if (warningLogged.compareAndSet(false, true)) {
           LOG.warning(
-              "Thread pool is saturating. Core pool size: %d, Active threads: %d, Queued tasks: %d. Consider increasing the thread count property.",
-              corePoolSize, getActiveCount(), queueSize);
+              "Thread pool is saturating. Max pool size: %d, Active threads: %d, Queued tasks: %d. Consider increasing the thread count property.",
+              maxPoolSize, getActiveCount(), queueSize);
         }
       } else if (queueSize <= recoveryThreshold) {
         if (warningLogged.get()) {

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

@@ -1077,8 +1077,9 @@ public Integer getMetadataFetchThreadCount() {
 
   public void setMetadataFetchThreadCount(Integer metadataFetchThreadCount) {
     if (metadataFetchThreadCount != null) {
-      validateNonNegative(
+      validateMin(
           metadataFetchThreadCount,
+          1,
           BigQueryJdbcUrlUtility.METADATA_FETCH_THREAD_COUNT_PROPERTY_NAME);
     }
     this.metadataFetchThreadCount = metadataFetchThreadCount;
@@ -1388,11 +1389,7 @@ private static void validateNonNegative(long val, String propertyName) {
     }
   }
 
-  /**
-   * Validates that a property value is greater than or equal to a minimum threshold. For thread
-   * pools, a minimum of 2 is enforced to ensure there are enough threads to handle concurrent
-   * coordination and avoid deadlock or thread starvation.
-   */
+  /** Validates that a property value is greater than or equal to a minimum threshold. */
   private static void validateMin(long val, long min, String propertyName) {
     if (val < min) {
       throw new BigQueryJdbcRuntimeException(

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

@@ -253,6 +253,44 @@ public void testPoolThreadInheritanceSevered() throws Exception {
     }
   }
 
+  @Test
+  public void testNewFixedThreadPoolTimeout() {
+    ExecutorService exec2 = BigQueryJdbcMdc.newFixedThreadPool(2);
+    ExecutorService exec3 = BigQueryJdbcMdc.newFixedThreadPool(3);
+    ExecutorService exec4 = BigQueryJdbcMdc.newFixedThreadPool(4);
+    ExecutorService exec5 = BigQueryJdbcMdc.newFixedThreadPool(5);
+    ExecutorService exec10 = BigQueryJdbcMdc.newFixedThreadPool(10);
+
+    try {
+      assertEquals(2, ((ThreadPoolExecutor) exec2).getCorePoolSize());
+      assertEquals(2, ((ThreadPoolExecutor) exec2).getMaximumPoolSize());
+      assertTrue(((ThreadPoolExecutor) exec2).allowsCoreThreadTimeOut());
+      assertEquals(60L, ((ThreadPoolExecutor) exec2).getKeepAliveTime(TimeUnit.SECONDS));
+
+      assertEquals(3, ((ThreadPoolExecutor) exec3).getCorePoolSize());
+      assertEquals(3, ((ThreadPoolExecutor) exec3).getMaximumPoolSize());
+      assertTrue(((ThreadPoolExecutor) exec3).allowsCoreThreadTimeOut());
+
+      assertEquals(4, ((ThreadPoolExecutor) exec4).getCorePoolSize());
+      assertEquals(4, ((ThreadPoolExecutor) exec4).getMaximumPoolSize());
+      assertTrue(((ThreadPoolExecutor) exec4).allowsCoreThreadTimeOut());
+
+      assertEquals(5, ((ThreadPoolExecutor) exec5).getCorePoolSize());
+      assertEquals(5, ((ThreadPoolExecutor) exec5).getMaximumPoolSize());
+      assertTrue(((ThreadPoolExecutor) exec5).allowsCoreThreadTimeOut());
+
+      assertEquals(10, ((ThreadPoolExecutor) exec10).getCorePoolSize());
+      assertEquals(10, ((ThreadPoolExecutor) exec10).getMaximumPoolSize());
+      assertTrue(((ThreadPoolExecutor) exec10).allowsCoreThreadTimeOut());
+    } finally {
+      exec2.shutdownNow();
+      exec3.shutdownNow();
+      exec4.shutdownNow();
+      exec5.shutdownNow();
+      exec10.shutdownNow();
+    }
+  }
+
   @Test
   public void testConnectionScopedExecutorLifecycle() throws Exception {
     String url1 =
@@ -287,8 +325,10 @@ public void testConnectionScopedExecutorLifecycle() throws Exception {
       assertEquals(Integer.MAX_VALUE, ((ThreadPoolExecutor) exec2).getMaximumPoolSize());
       assertEquals(5, ((ThreadPoolExecutor) metadataExec1).getCorePoolSize());
       assertEquals(5, ((ThreadPoolExecutor) metadataExec1).getMaximumPoolSize());
+      assertTrue(((ThreadPoolExecutor) metadataExec1).allowsCoreThreadTimeOut());
       assertEquals(10, ((ThreadPoolExecutor) metadataExec2).getCorePoolSize());
       assertEquals(10, ((ThreadPoolExecutor) metadataExec2).getMaximumPoolSize());
+      assertTrue(((ThreadPoolExecutor) metadataExec2).allowsCoreThreadTimeOut());
 
       try (BigQueryJdbcMdc.MdcCloseable mdc =
           BigQueryJdbcMdc.registerInstance(conn1.getConnectionId())) {

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

@@ -16,12 +16,14 @@
 
 package com.google.cloud.bigquery.jdbc;
 
+import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import com.google.cloud.bigquery.exception.BigQueryJdbcException;
+import com.google.cloud.bigquery.exception.BigQueryJdbcRuntimeException;
 import java.sql.SQLException;
 import org.junit.jupiter.api.Test;
 
@@ -46,4 +48,39 @@ public void testWrapperMethods() throws SQLException {
             BigQueryJdbcException.class, () -> dataSource.unwrap(java.sql.Connection.class));
     assertTrue(e.getMessage().contains("Cannot unwrap to java.sql.Connection"));
   }
+
+  @Test
+  public void testMetadataFetchThreadCountValidation() {
+    DataSource dataSource = new DataSource();
+
+    // Setting positive values should succeed
+    dataSource.setMetadataFetchThreadCount(1);
+    assertEquals(1, dataSource.getMetadataFetchThreadCount());
+
+    dataSource.setMetadataFetchThreadCount(5);
+    assertEquals(5, dataSource.getMetadataFetchThreadCount());
+
+    dataSource.setMetadataFetchThreadCount(null); // Should fallback to default
+    assertEquals(
+        BigQueryJdbcUrlUtility.DEFAULT_METADATA_FETCH_THREAD_COUNT_VALUE,
+        dataSource.getMetadataFetchThreadCount());
+
+    // Setting 0 or negative should throw BigQueryJdbcRuntimeException
+    BigQueryJdbcRuntimeException ex0 =
+        assertThrows(
+            BigQueryJdbcRuntimeException.class, () -> dataSource.setMetadataFetchThreadCount(0));
+    assertTrue(
+        ex0.getMessage()
+            .contains(
+                "Invalid value for MetaDataFetchThreadCount. It must be greater than or equal to 1"));
+
+    BigQueryJdbcRuntimeException exNeg =
+        assertThrows(
+            BigQueryJdbcRuntimeException.class, () -> dataSource.setMetadataFetchThreadCount(-5));
+    assertTrue(
+        exNeg
+            .getMessage()
+            .contains(
+                "Invalid value for MetaDataFetchThreadCount. It must be greater than or equal to 1"));
+  }
 }