Fix SDK client starvation under high thread count (200-300 collection loaders)

- SDKClientPool: replace ConcurrentLinkedQueue with LinkedBlockingQueue for idle/busy pools
- SDKClientPool: get_client_for_bucket now blocks via poll(30, MINUTES) instead of returning null immediately
- SDKClientPool: remove idlePool.isEmpty() pre-check that caused spurious null returns under contention
- WorkLoadGenerate: remove busy-poll retry loop (while retries < 30) — blocking now handled in pool
- WorkLoadGenerate: simplify run() client acquisition to single call with InterruptedException handling

Root cause: with 200-300 threads competing for a finite client pool and long-running
doc loads holding clients, the 30×1s busy-poll cap expired before clients were released,
causing threads to fail with "Failed to acquire SDK client after 30 retries".
Blocking in the pool parks threads efficiently and removes the arbitrary time ceiling.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: ashwin2002

src/main/java/couchbase/loadgen/WorkLoadGenerate.java

@@ -474,30 +474,23 @@ else if(ops < dg.ws.ops/dg.ws.workers && flag) {
     @Override
     public void run() {
         if (this.sdkClientPool != null) {
-            int retries = 0;
-            while (this.sdk == null && retries < 30) {
-                try {
-                    this.sdk = this.sdkClientPool.get_client_for_bucket(
+            try {
+                // Pool blocks internally until a client is available (up to its configured timeout).
+                // No busy-poll needed here — blocking in the pool is cheaper and has no retry cap.
+                this.sdk = this.sdkClientPool.get_client_for_bucket(
                         this.bucket_name, this.scope, this.collection);
-                } catch (Exception e) {
-                    logger.error("Error getting SDK client from pool for bucket "
-                            + this.bucket_name + ": " + e.getMessage());
-                }
-                if (this.sdk == null) {
-                    try {
-                        TimeUnit.SECONDS.sleep(1);
-                    } catch (InterruptedException e) {
-                        logger.error("Interrupted while waiting for SDK client", e);
-                        Thread.currentThread().interrupt();
-                        this.result = false;
-                        return;
-                    }
-                    retries++;
-                }
+            } catch (InterruptedException e) {
+                logger.error("Interrupted while waiting for SDK client for bucket "
+                        + this.bucket_name, e);
+                Thread.currentThread().interrupt();
+                this.result = false;
+                return;
+            } catch (Exception e) {
+                logger.error("Error acquiring SDK client for bucket "
+                        + this.bucket_name + ": " + e.getMessage(), e);
             }
             if (this.sdk == null) {
-                logger.error("Failed to acquire SDK client for bucket "
-                        + this.bucket_name + " after " + retries + " retries");
+                logger.error("Failed to acquire SDK client for bucket " + this.bucket_name);
                 this.result = false;
                 return;
             }

src/main/java/couchbase/sdk/SDKClientPool.java

@@ -1,9 +1,8 @@
 package couchbase.sdk;
 
-import java.util.ArrayList;
-import java.util.ConcurrentModificationException;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ConcurrentLinkedQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
 import org.apache.log4j.LogManager;
@@ -16,9 +15,12 @@ public class SDKClientPool {
     // Thread-safe client collection cache
     private ConcurrentHashMap<String, ClientInfo> clientCache = new ConcurrentHashMap<>();
 
+    // Block up to this long waiting for an idle client before giving up
+    private static final int CLIENT_WAIT_TIMEOUT_MINUTES = 30;
+
     // Thread-safe client pools by bucket
-    private ConcurrentHashMap<String, ConcurrentLinkedQueue<SDKClient>> idleClients = new ConcurrentHashMap<>();
-    private ConcurrentHashMap<String, ConcurrentLinkedQueue<SDKClient>> busyClients = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<String, LinkedBlockingQueue<SDKClient>> idleClients = new ConcurrentHashMap<>();
+    private ConcurrentHashMap<String, LinkedBlockingQueue<SDKClient>> busyClients = new ConcurrentHashMap<>();
 
     public SDKClientPool() {
         super();
@@ -29,8 +31,8 @@ public void shutdown() {
 
         // Process all buckets
         for (String bucketName : idleClients.keySet()) {
-            ConcurrentLinkedQueue<SDKClient> idle = idleClients.get(bucketName);
-            ConcurrentLinkedQueue<SDKClient> busy = busyClients.get(bucketName);
+            LinkedBlockingQueue<SDKClient> idle = idleClients.get(bucketName);
+            LinkedBlockingQueue<SDKClient> busy = busyClients.get(bucketName);
 
             if (idle != null) {
                 for (SDKClient client : idle) {
@@ -54,8 +56,8 @@ public void shutdown() {
     }
 
     public void force_close_clients_for_bucket(String bucket_name) {
-        ConcurrentLinkedQueue<SDKClient> idle = idleClients.get(bucket_name);
-        ConcurrentLinkedQueue<SDKClient> busy = busyClients.get(bucket_name);
+        LinkedBlockingQueue<SDKClient> idle = idleClients.get(bucket_name);
+        LinkedBlockingQueue<SDKClient> busy = busyClients.get(bucket_name);
 
         if (idle != null) {
             for (SDKClient client : idle) {
@@ -74,19 +76,20 @@ public void force_close_clients_for_bucket(String bucket_name) {
 
     public void create_clients(String bucket_name, Server server, int req_clients) throws Exception {
         // Initialize thread-safe client pools for this bucket if not already present
-        idleClients.computeIfAbsent(bucket_name, k -> new ConcurrentLinkedQueue<>());
-        busyClients.computeIfAbsent(bucket_name, k -> new ConcurrentLinkedQueue<>());
-        
-        ConcurrentLinkedQueue<SDKClient> idlePool = idleClients.get(bucket_name);
+        idleClients.computeIfAbsent(bucket_name, k -> new LinkedBlockingQueue<>());
+        busyClients.computeIfAbsent(bucket_name, k -> new LinkedBlockingQueue<>());
 
+        LinkedBlockingQueue<SDKClient> idlePool = idleClients.get(bucket_name);
+
         for (int i = 0; i < req_clients; i++) {
             SDKClient client = new SDKClient(server, bucket_name);
             client.initialiseSDK();
             idlePool.add(client);
         }
     }
 
-    public SDKClient get_client_for_bucket(String bucket_name, String scope, String collection) {
+    public SDKClient get_client_for_bucket(String bucket_name, String scope, String collection)
+            throws InterruptedException {
         String cache_key = bucket_name + ":" + scope + ":" + collection;
 
         // Check if client is already cached for this collection
@@ -95,28 +98,32 @@ public SDKClient get_client_for_bucket(String bucket_name, String scope, String
             existing.counter.incrementAndGet();
             return existing.client;
         }
-        
+
         // Get idle client pool for this bucket
-        ConcurrentLinkedQueue<SDKClient> idlePool = idleClients.get(bucket_name);
-        if (idlePool == null || idlePool.isEmpty()) {
+        LinkedBlockingQueue<SDKClient> idlePool = idleClients.get(bucket_name);
+        if (idlePool == null) {
             return null;
         }
-        
-        // Get client from idle pool atomically
-        SDKClient client = idlePool.poll();
+
+        // Block until a client becomes available or timeout expires.
+        // With 200-300 threads sharing a finite pool, spinning with a fixed retry cap
+        // causes spurious failures on long-running loads — blocking here is cheaper and correct.
+        SDKClient client = idlePool.poll(CLIENT_WAIT_TIMEOUT_MINUTES, TimeUnit.MINUTES);
         if (client == null) {
+            logger.error("Timed out waiting " + CLIENT_WAIT_TIMEOUT_MINUTES
+                    + " min for idle SDK client for bucket " + bucket_name);
             return null;
         }
-        
+
         // Configure client for this collection
         client.selectCollection(scope, collection);
-        
+
         // Add to busy pool atomically
-        busyClients.computeIfAbsent(bucket_name, k -> new ConcurrentLinkedQueue<>()).add(client);
-        
+        busyClients.computeIfAbsent(bucket_name, k -> new LinkedBlockingQueue<>()).add(client);
+
         // Cache client reference with thread-safe counter
         clientCache.put(cache_key, new ClientInfo(client, new AtomicInteger(1)));
-        
+
         return client;
     }
 
@@ -142,8 +149,8 @@ public void release_client(SDKClient client) {
             clientCache.remove(cache_key);
 
             // Remove from busy pool and add to idle pool atomically
-            ConcurrentLinkedQueue<SDKClient> busyPool = busyClients.get(bucket_key);
-            ConcurrentLinkedQueue<SDKClient> idlePool = idleClients.get(bucket_key);
+            LinkedBlockingQueue<SDKClient> busyPool = busyClients.get(bucket_key);
+            LinkedBlockingQueue<SDKClient> idlePool = idleClients.get(bucket_key);
 
             if (busyPool != null) {
                 busyPool.remove(client);