Java FIT performer: support two ClusterEnvironment modes

3.2.6 added an easy way for SDK users to configure
the SDK without having to take ownership of
ClusterEnvironment management.  It also allows
passing parameters in the connection string, which
is not allowed with those externally owned
ClusterEnvironments.

The performer will now use the "new" mode where
possible.  The features required by the new mode
are likely not to be required in the situations
where we are using the old mode.

Change-Id: I58755c369856bd977779fee4174ca54814189dea
Reviewed-on: https://review.couchbase.org/c/couchbase-jvm-clients/+/229357
Reviewed-by: David Nault <[email protected]>
Tested-by: Build Bot <[email protected]>

Author: programmatix

java-fit-performer/src/main/java/com/couchbase/JavaPerformer.java

@@ -231,19 +231,30 @@ public void clusterConnectionCreate(ClusterConnectionCreateRequest request,
                 });
             });
 
-            var clusterEnvironment = OptionsUtil.convertClusterConfig(request, getCluster, onClusterConnectionClose);
-
-            // [if:3.7.5] first version that allows specifying custom publishOn scheduler
-            var userExecutorAndScheduler = UserSchedulerUtil.userExecutorAndScheduler();
-            onClusterConnectionClose.add(userExecutorAndScheduler::dispose);
-            clusterEnvironment.publishOnScheduler(userExecutorAndScheduler::scheduler);
-            // [end]
+            // [if:3.2.6]
+            // 3.2.6 added an easy way for SDK users to configure the SDK without having to take ownership of
+            // ClusterEnvironment management.  It also allows passing parameters in the connection string, which
+            // is not allowed with those externally owned ClusterEnvironments.
+            var clusterEnvironment = OptionsUtil.convertClusterConfigToConsumer(request, getCluster, onClusterConnectionClose);
 
             var connection = new ClusterConnection(request.getClusterHostname(),
                     request.getClusterUsername(),
                     request.getClusterPassword(),
                     clusterEnvironment,
                     onClusterConnectionClose);
+            // [end]
+
+            // [if:<3.2.6]
+            // Support falling back to the original way of creating
+            //? var clusterEnvironment = OptionsUtil.convertClusterConfig(request, getCluster, onClusterConnectionClose);
+
+            //? var connection = new ClusterConnection(request.getClusterHostname(),
+            //?         request.getClusterUsername(),
+            //?         request.getClusterPassword(),
+            //?         clusterEnvironment,
+            //?         onClusterConnectionClose);
+            // [end]
+
             clusterConnections.put(clusterConnectionId, connection);
             logger.info("Created cluster connection {} for user {}, now have {}",
                     clusterConnectionId, request.getClusterUsername(), clusterConnections.size());

java-fit-performer/src/main/java/com/couchbase/utils/ClusterConnection.java

@@ -29,11 +29,12 @@
 import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
+import java.util.function.Consumer;
 
 
 public class ClusterConnection {
     private final Cluster cluster;
-    @Nullable private final ClusterEnvironment config;
+    @Nullable private final ClusterEnvironment environmentOwnedByCaller;
     public final String username;
     // Commands to run when this ClusterConnection is being closed.  Allows closing other related resources that have
     // the same lifetime.
@@ -49,16 +50,38 @@ public ClusterConnection(String hostname,
 
         var co = ClusterOptions.clusterOptions(username, password);
         if (config != null) {
-            this.config = config.build();
-            co.environment(this.config);
+            this.environmentOwnedByCaller = config.build();
+            co.environment(this.environmentOwnedByCaller);
         }
         else {
-            this.config = null;
+            this.environmentOwnedByCaller = null;
         }
 
         this.cluster = Cluster.connect(hostname, co);
     }
 
+    // [if:3.2.6]
+    public ClusterConnection(String hostname,
+                             String username,
+                             String password,
+                             @Nullable Consumer<ClusterEnvironment.Builder> config,
+                             ArrayList<Runnable> onClusterConnectionClose)  {
+        this.username = username;
+        this.onClusterConnectionClose = onClusterConnectionClose;
+
+        var co = ClusterOptions.clusterOptions(username, password)
+                .environment(env -> {
+                    if (config != null) {
+                        config.accept(env);
+                    }
+                });
+        // The SDK will manage things instead in this mode
+        this.environmentOwnedByCaller = null;
+
+        this.cluster = Cluster.connect(hostname, co);
+    }
+    // [end]
+
     public Cluster cluster(){
         return cluster;
     }
@@ -104,8 +127,8 @@ public Collection collection(com.couchbase.client.protocol.shared.Collection col
 
     public void close() {
         cluster.disconnect();
-        if (config != null) {
-            config.shutdown();
+        if (environmentOwnedByCaller != null) {
+            environmentOwnedByCaller.shutdown();
         }
         onClusterConnectionClose.forEach(Runnable::run);
     }

java-fit-performer/src/main/java/com/couchbase/utils/OptionsUtil.java

@@ -30,7 +30,6 @@
 import com.couchbase.client.core.env.SecurityConfig;
 import com.couchbase.client.core.env.TimeoutConfig;
 import com.couchbase.client.core.msg.kv.DurabilityLevel;
-import com.couchbase.client.java.ClusterOptions;
 import com.couchbase.client.java.env.ClusterEnvironment;
 import com.couchbase.client.java.json.JsonArray;
 import com.couchbase.client.java.json.JsonObject;
@@ -72,7 +71,6 @@
 import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
-import javax.net.ssl.TrustManagerFactory;
 import java.io.ByteArrayInputStream;
 import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
@@ -85,6 +83,7 @@
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.TimeUnit;
+import java.util.function.Consumer;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
 
@@ -93,6 +92,23 @@ public class OptionsUtil {
 
     private OptionsUtil() {}
 
+    public static Consumer<ClusterEnvironment.Builder> convertClusterConfigToConsumer(ClusterConnectionCreateRequest request,
+                                                                                      Supplier<ClusterConnection> getCluster,
+                                                                                      ArrayList<Runnable> onClusterConnectionClose) {
+        return (builder) -> {
+            if (request.hasClusterConfig()) {
+                var cc = request.getClusterConfig();
+                applyClusterConfig(builder, cc, onClusterConnectionClose);
+
+                // [if:3.3.0]
+                if (request.getClusterConfig().hasTransactionsConfig()) {
+                    applyTransactionsConfig(request, getCluster, builder);
+                }
+                // [end]
+            }
+        };
+    }
+
     public static
     ClusterEnvironment.Builder convertClusterConfig(ClusterConnectionCreateRequest request,
                                                     Supplier<ClusterConnection> getCluster,
@@ -102,14 +118,10 @@ ClusterEnvironment.Builder convertClusterConfig(ClusterConnectionCreateRequest r
         if (request.hasClusterConfig()) {
             var cc = request.getClusterConfig();
 
-            // [if:3.3.0]
-            if (request.getClusterConfig().hasTransactionsConfig()) {
-                applyTransactionsConfig(request, getCluster, clusterEnvironment);
-            }
-            // [end]
-
             applyClusterConfig(clusterEnvironment, cc, onClusterConnectionClose);
 
+            // No need to support transactions here, as this code is now only executed in <3.2.6 mode (transactions
+            // was introduced in 3.3.0)
         }
 
         return clusterEnvironment;
@@ -295,6 +307,12 @@ private static void applyClusterConfig(ClusterEnvironment.Builder clusterEnviron
             clusterEnvironment.disableAppTelemetry(!cc.getEnableAppTelemetry());
             // [end]
         }
+
+        // [if:3.7.5] first version that allows specifying custom publishOn scheduler
+        var userExecutorAndScheduler = UserSchedulerUtil.userExecutorAndScheduler();
+        onClusterConnectionClose.add(userExecutorAndScheduler::dispose);
+        clusterEnvironment.publishOnScheduler(userExecutorAndScheduler::scheduler);
+        // [end]
     }
 
     // [if:3.5.1]