Test new enhancements

Author: ashwin2002

src/main/java/RestServer/TaskRequest.java

@@ -702,7 +702,7 @@ public ResponseEntity<Map<String, Object>> doc_load() {
         }
 
         // Calculate effective number of workers needed
-        int effectiveWorkers = Math.min(ws.workers,
+        int effectiveWorkers = Math.min(ws.workers, 
             (int)((totalDocsToProcess + docsPerWorker - 1) / docsPerWorker));  // ceil division
 
         System.out.println("Smart worker counting: Total docs=" + totalDocsToProcess +

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

@@ -1,7 +1,5 @@
 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.atomic.AtomicInteger;

src/main/java/couchbase/sdk/SharedClusterManager.java

@@ -23,55 +23,39 @@
  */
 public class SharedClusterManager {
     private static Logger logger = LogManager.getLogger(SharedClusterManager.class);
-
+    
     // Common KV connections setting for massively parallel collection loads
     // Increased from default 5 to 500 to support 5,000 collections loading in parallel
-    private static final int DEFAULT_KV_CONNECTIONS = 5;
-
+    private static final int DEFAULT_KV_CONNECTIONS = 500;
+    
     // Shared ClusterEnvironment with optimized connection settings
     private static ClusterEnvironment sharedEnvironment;
-
-    // Track whether environment has been shutdown
-    private static volatile boolean environmentShutdown = false;
-
-    private static final Object environmentLock = new Object();
-
+    
     // Store cluster instances per server connection string
     private static ConcurrentHashMap<String, ClusterWrapper> clusterMap = new ConcurrentHashMap<>();
-
-    // Initialize the shared environment (lazy initialization with recreation)
+    
+    // Initialize the shared environment once (lazy initialization)
     private static void initializeSharedEnvironment() {
-        if (sharedEnvironment == null || environmentShutdown) {
-            synchronized (environmentLock) {
-                // Double-check under lock
-                if (sharedEnvironment == null || environmentShutdown) {
-                    try {
-                        if (sharedEnvironment != null && environmentShutdown) {
-                            logger.info("Shared Cluster Environment was shutdown, recreating");
-                        }
-                        
-                        sharedEnvironment = ClusterEnvironment.builder()
-                                .timeoutConfig(TimeoutConfig.builder().kvTimeout(Duration.ofSeconds(10)))
-                                .securityConfig(SecurityConfig.enableTls(true)
-                                        .trustManagerFactory(InsecureTrustManagerFactory.INSTANCE))
-                                .ioConfig(IoConfig.enableDnsSrv(true))
-                                .ioConfig(IoConfig.numKvConnections(DEFAULT_KV_CONNECTIONS))
-                                .build();
-                        environmentShutdown = false;
-                        logger.info("Shared Cluster Environment initialized with " + DEFAULT_KV_CONNECTIONS + " KV connections for massively parallel collection loads");
-                    } catch (Exception e) {
-                        logger.error("Failed to initialize shared Cluster Environment", e);
-                    }
-                }
+        if (sharedEnvironment == null) {
+            try {
+                sharedEnvironment = ClusterEnvironment.builder()
+                        .timeoutConfig(TimeoutConfig.builder().kvTimeout(Duration.ofSeconds(10)))
+                        .securityConfig(SecurityConfig.enableTls(true)
+                                .trustManagerFactory(InsecureTrustManagerFactory.INSTANCE))
+                        .ioConfig(IoConfig.enableDnsSrv(true))
+                        .ioConfig(IoConfig.numKvConnections(DEFAULT_KV_CONNECTIONS))
+                        .build();
+                logger.info("Shared Cluster Environment initialized with " + DEFAULT_KV_CONNECTIONS + " KV connections for massively parallel collection loads");
+            } catch (Exception e) {
+                logger.error("Failed to initialize shared Cluster Environment", e);
             }
         }
     }
-
+    
     /**
      * Get or create a shared Cluster instance for the given server connection
      */
     public static synchronized Cluster getCluster(Server server) throws AuthenticationFailureException {
-        initializeSharedEnvironment();
         String clusterKey = getClusterKey(server);
 
         ClusterWrapper wrapper = clusterMap.get(clusterKey);
@@ -88,27 +72,27 @@ public static synchronized Cluster getCluster(Server server) throws Authenticati
 
         return wrapper.cluster;
     }
-
+    
     /**
      * Release reference to the shared Cluster instance
      */
     public static synchronized void releaseCluster(Server server) {
         String clusterKey = getClusterKey(server);
         ClusterWrapper wrapper = clusterMap.get(clusterKey);
-
+        
         if (wrapper != null) {
             int refCount = wrapper.decrementRefCount();
-            logger.debug("Released Cluster instance for server: " + server.ip +
+            logger.debug("Released Cluster instance for server: " + server.ip + 
                         " (ref count: " + refCount + ")");
-
+            
             if (refCount == 0) {
                 logger.info("No more references, disconnecting Cluster for server: " + server.ip);
                 wrapper.cluster.disconnect();
                 clusterMap.remove(clusterKey);
             }
         }
     }
-
+    
     /**
      * Shutdown all cluster instances and the shared environment
      */
@@ -123,11 +107,10 @@ public static synchronized void shutdownAll() {
 
         if (sharedEnvironment != null) {
             sharedEnvironment.shutdown();
-            environmentShutdown = true;
             logger.info("Shared Cluster Environment shutdown complete");
         }
     }
-
+    
     private static Cluster createCluster(Server server) throws AuthenticationFailureException {
         ClusterOptions clusterOptions;
         try {
@@ -138,52 +121,52 @@ private static Cluster createCluster(Server server) throws AuthenticationFailure
                 clusterOptions = ClusterOptions.clusterOptions(server.rest_username, server.rest_password)
                         .environment(createNonTLSEnvironment());
             }
-
+            
             Cluster cluster = Cluster.connect(server.ip, clusterOptions);
             logger.info("Cluster connection successful: " + server.ip);
             return cluster;
         } catch (AuthenticationFailureException e) {
-            logger.error("Authentication failed for server: " + server.ip +
+            logger.error("Authentication failed for server: " + server.ip + 
                         " with user: " + server.rest_username);
             throw e;
         } catch (Exception e) {
             logger.error("Failed to connect Cluster to server: " + server.ip, e);
             throw new RuntimeException("Cluster connection failed", e);
         }
     }
-
+    
     private static ClusterEnvironment createNonTLSEnvironment() {
         return ClusterEnvironment.builder()
                 .timeoutConfig(TimeoutConfig.builder().kvTimeout(Duration.ofSeconds(10)))
                 .ioConfig(IoConfig.enableDnsSrv(true))
                 .ioConfig(IoConfig.numKvConnections(DEFAULT_KV_CONNECTIONS))
                 .build();
     }
-
+    
     private static String getClusterKey(Server server) {
         return server.ip + ":" + server.memcached_port;
     }
-
+    
     /**
      * Wrapper class to track reference count for shared Cluster instances
      */
     private static class ClusterWrapper {
         Cluster cluster;
         AtomicInteger refCount;
-
+        
         ClusterWrapper(Cluster cluster) {
             this.cluster = cluster;
             this.refCount = new AtomicInteger(1);
         }
-
+        
         void incrementRefCount() {
             refCount.incrementAndGet();
         }
-
+        
         int decrementRefCount() {
             return refCount.decrementAndGet();
         }
-
+        
         int getRefCount() {
             return refCount.get();
         }