Now using bound executor pool with a limit of 100.

Author: vverman

google-auth-library-java/oauth2_http/java/com/google/auth/oauth2/RegionalAccessBoundaryManager.java

@@ -88,16 +88,22 @@ final class RegionalAccessBoundaryManager {
   // on concurrent refresh tasks, while threadCount provides unique names
   // for easier debugging.
   private static final AtomicInteger threadCount = new AtomicInteger(0);
+  
+  // Bounded executor service ensures hard limits on concurrent refresh tasks and queued tasks
+  // to avoid resource exhaustion.
+  private static final int EXECUTOR_POOL_SIZE = 5;
+  private static final int EXECUTOR_QUEUE_CAPACITY = 100;
+  
   private static final ExecutorService EXECUTOR;
 
   static {
     ThreadPoolExecutor executor =
         new ThreadPoolExecutor(
-            5, // corePoolSize: threads to keep alive
-            5, // maximumPoolSize: max threads allowed
+            EXECUTOR_POOL_SIZE, // corePoolSize: threads to keep alive
+            EXECUTOR_POOL_SIZE, // maximumPoolSize: max threads allowed
             1, // keepAliveTime: time to wait before terminating idle threads
             TimeUnit.HOURS, // unit for keepAliveTime
-            new LinkedBlockingQueue<>(), // work queue
+            new LinkedBlockingQueue<>(EXECUTOR_QUEUE_CAPACITY), // work queue with bound
             r -> {
               Thread t = new Thread(r, "RAB-refresh-" + threadCount.getAndIncrement());
               t.setDaemon(true);

google-auth-library-java/oauth2_http/javatests/com/google/auth/oauth2/RegionalAccessBoundaryTest.java

@@ -45,6 +45,7 @@
 import java.io.ObjectInputStream;
 import java.io.ObjectOutputStream;
 import java.util.Collections;
+import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.atomic.AtomicLong;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -224,6 +225,57 @@ public void testManagerTriggersRefreshInGracePeriod() throws InterruptedExceptio
     assertEquals(newerEncoded, resultRab.getEncodedLocations());
   }
 
+  @Test
+  public void testExecutorQueueCapacityLimit() throws Exception {
+    final String url = "https://example.com/rab";
+    final AccessToken token = new AccessToken("token", new java.util.Date(System.currentTimeMillis() + 3600000L));
+    RegionalAccessBoundaryProvider provider = () -> url;
+
+    int poolSize = 5;
+    int queueCapacity = 100;
+    int totalCapacity = poolSize + queueCapacity;
+
+    CountDownLatch latch = new CountDownLatch(1);
+    
+    java.io.InputStream blockingStream = new java.io.InputStream() {
+      private final java.io.InputStream delegate = new ByteArrayInputStream("{\"encodedLocations\": \"encoded\", \"locations\": [\"loc\"]}".getBytes());
+      private boolean blocked = false;
+
+      @Override
+      public int read() throws java.io.IOException {
+        if (!blocked) {
+          try {
+            latch.await();
+          } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+          }
+          blocked = true;
+        }
+        return delegate.read();
+      }
+    };
+
+    MockHttpTransport transport = new MockHttpTransport.Builder()
+        .setLowLevelHttpResponse(new MockLowLevelHttpResponse().setContent(blockingStream).setContentType("application/json"))
+        .build();
+    HttpTransportFactory transportFactory = () -> transport;
+
+    RegionalAccessBoundaryManager[] managers = new RegionalAccessBoundaryManager[totalCapacity];
+    for (int i = 0; i < totalCapacity; i++) {
+      managers[i] = new RegionalAccessBoundaryManager(testClock);
+      managers[i].triggerAsyncRefresh(transportFactory, provider, token);
+    }
+
+    RegionalAccessBoundaryManager extraManager = new RegionalAccessBoundaryManager(testClock);
+    assertFalse(extraManager.isCooldownActive());
+    
+    extraManager.triggerAsyncRefresh(transportFactory, provider, token);
+
+    assertTrue(extraManager.isCooldownActive(), "106th task should have been rejected and entered cooldown");
+
+    latch.countDown();
+  }
+
   private static class TestClock implements Clock {
     private final AtomicLong currentTime = new AtomicLong(System.currentTimeMillis());