Make maxConnecting configurable (#822)

Note that `maxConnecting` is duplicated in `MongoClientOptions` similarly to other fields.
https://jira.mongodb.org/browse/JAVA-4404 was filed to refactor the whole class.

JAVA-4390

Author: stIncMale

driver-core/src/main/com/mongodb/ConnectionString.java

@@ -16,6 +16,7 @@
 
 package com.mongodb;
 
+import com.mongodb.connection.ConnectionPoolSettings;
 import com.mongodb.diagnostics.logging.Logger;
 import com.mongodb.diagnostics.logging.Loggers;
 import com.mongodb.internal.dns.DefaultDnsResolver;
@@ -126,6 +127,7 @@
  * <li>{@code maxPoolSize=n}: The maximum number of connections in the connection pool.</li>
  * <li>{@code waitQueueTimeoutMS=ms}: The maximum wait time in milliseconds that a thread may wait for a connection to
  * become available.</li>
+ * <li>{@code maxConnecting=n}: The maximum number of connections a pool may be establishing concurrently.</li>
  * </ul>
  * <p>Write concern configuration:</p>
  * <ul>
@@ -265,6 +267,7 @@ public class ConnectionString {
     private Integer maxWaitTime;
     private Integer maxConnectionIdleTime;
     private Integer maxConnectionLifeTime;
+    private Integer maxConnecting;
     private Integer connectTimeout;
     private Integer socketTimeout;
     private Boolean sslEnabled;
@@ -446,6 +449,7 @@ public ConnectionString(final String connectionString) {
         GENERAL_OPTIONS_KEYS.add("connecttimeoutms");
         GENERAL_OPTIONS_KEYS.add("maxidletimems");
         GENERAL_OPTIONS_KEYS.add("maxlifetimems");
+        GENERAL_OPTIONS_KEYS.add("maxconnecting");
         GENERAL_OPTIONS_KEYS.add("sockettimeoutms");
 
         // Order matters here: Having tls after ssl means than the tls option will supersede the ssl option when both are set
@@ -542,6 +546,8 @@ private void translateOptions(final Map<String, List<String>> optionsMap) {
                 maxConnectionIdleTime = parseInteger(value, "maxidletimems");
             } else if (key.equals("maxlifetimems")) {
                 maxConnectionLifeTime = parseInteger(value, "maxlifetimems");
+            } else if (key.equals("maxconnecting")) {
+                maxConnecting = parseInteger(value, "maxConnecting");
             } else if (key.equals("waitqueuetimeoutms")) {
                 maxWaitTime = parseInteger(value, "waitqueuetimeoutms");
             } else if (key.equals("connecttimeoutms")) {
@@ -1300,6 +1306,18 @@ public Integer getMaxConnectionLifeTime() {
         return maxConnectionLifeTime;
     }
 
+    /**
+     * Gets the maximum number of connections a pool may be establishing concurrently specified in the connection string.
+     * @return The maximum number of connections a pool may be establishing concurrently
+     * if the {@code maxConnecting} option is specified in the connection string, or {@code null} otherwise.
+     * @see ConnectionPoolSettings#getMaxConnecting()
+     * @since 4.4
+     */
+    @Nullable
+    public Integer getMaxConnecting() {
+        return maxConnecting;
+    }
+
     /**
      * Gets the socket connect timeout specified in the connection string.
      * @return the socket connect timeout
@@ -1445,6 +1463,7 @@ public boolean equals(final Object o) {
                 && Objects.equals(maxWaitTime, that.maxWaitTime)
                 && Objects.equals(maxConnectionIdleTime, that.maxConnectionIdleTime)
                 && Objects.equals(maxConnectionLifeTime, that.maxConnectionLifeTime)
+                && Objects.equals(maxConnecting, that.maxConnecting)
                 && Objects.equals(connectTimeout, that.connectTimeout)
                 && Objects.equals(socketTimeout, that.socketTimeout)
                 && Objects.equals(sslEnabled, that.sslEnabled)
@@ -1462,8 +1481,8 @@ public boolean equals(final Object o) {
     public int hashCode() {
         return Objects.hash(credential, isSrvProtocol, hosts, database, collection, directConnection, readPreference,
                 writeConcern, retryWrites, retryReads, readConcern, minConnectionPoolSize, maxConnectionPoolSize, maxWaitTime,
-                maxConnectionIdleTime, maxConnectionLifeTime, connectTimeout, socketTimeout, sslEnabled, sslInvalidHostnameAllowed,
-                requiredReplicaSetName, serverSelectionTimeout, localThreshold, heartbeatFrequency, applicationName, compressorList,
-                uuidRepresentation);
+                maxConnectionIdleTime, maxConnectionLifeTime, maxConnecting, connectTimeout, socketTimeout, sslEnabled,
+                sslInvalidHostnameAllowed, requiredReplicaSetName, serverSelectionTimeout, localThreshold, heartbeatFrequency,
+                applicationName, compressorList, uuidRepresentation);
     }
 }

driver-core/src/main/com/mongodb/connection/ConnectionPoolSettings.java

@@ -19,7 +19,9 @@
 import com.mongodb.ConnectionString;
 import com.mongodb.annotations.Immutable;
 import com.mongodb.annotations.NotThreadSafe;
+import com.mongodb.event.ConnectionCreatedEvent;
 import com.mongodb.event.ConnectionPoolListener;
+import com.mongodb.event.ConnectionReadyEvent;
 
 import java.util.ArrayList;
 import java.util.List;
@@ -46,6 +48,7 @@ public class ConnectionPoolSettings {
     private final long maxConnectionIdleTimeMS;
     private final long maintenanceInitialDelayMS;
     private final long maintenanceFrequencyMS;
+    private final int maxConnecting;
 
     /**
      * Gets a Builder for creating a new ConnectionPoolSettings instance.
@@ -80,6 +83,7 @@ public static final class Builder {
         private long maxConnectionIdleTimeMS;
         private long maintenanceInitialDelayMS;
         private long maintenanceFrequencyMS = MILLISECONDS.convert(1, MINUTES);
+        private int maxConnecting = 2;
 
         Builder() {
         }
@@ -103,6 +107,7 @@ public Builder applySettings(final ConnectionPoolSettings connectionPoolSettings
             maxConnectionIdleTimeMS = connectionPoolSettings.maxConnectionIdleTimeMS;
             maintenanceInitialDelayMS = connectionPoolSettings.maintenanceInitialDelayMS;
             maintenanceFrequencyMS = connectionPoolSettings.maintenanceFrequencyMS;
+            maxConnecting = connectionPoolSettings.maxConnecting;
             return this;
         }
 
@@ -210,6 +215,19 @@ public Builder addConnectionPoolListener(final ConnectionPoolListener connection
             return this;
         }
 
+        /**
+         * The maximum number of connections a pool may be establishing concurrently.
+         *
+         * @param maxConnecting The maximum number of connections a pool may be establishing concurrently. Must be positive.
+         * @return {@code this}.
+         * @see ConnectionPoolSettings#getMaxConnecting()
+         * @since 4.4
+         */
+        public Builder maxConnecting(final int maxConnecting) {
+            this.maxConnecting = maxConnecting;
+            return this;
+        }
+
         /**
          * Creates a new ConnectionPoolSettings object with the settings initialised on this builder.
          *
@@ -251,6 +269,11 @@ public Builder applyConnectionString(final ConnectionString connectionString) {
                 maxConnectionLifeTime(maxConnectionLifeTime, MILLISECONDS);
             }
 
+            Integer maxConnecting = connectionString.getMaxConnecting();
+            if (maxConnecting != null) {
+                maxConnecting(maxConnecting);
+            }
+
             return this;
         }
     }
@@ -343,6 +366,21 @@ public List<ConnectionPoolListener> getConnectionPoolListeners() {
         return connectionPoolListeners;
     }
 
+    /**
+     * The maximum number of connections a pool may be establishing concurrently.
+     * Establishment of a connection is a part of its life cycle
+     * starting after a {@link ConnectionCreatedEvent} and ending before a {@link ConnectionReadyEvent}.
+     * <p>
+     * Default is 2.</p>
+     *
+     * @return The maximum number of connections a pool may be establishing concurrently.
+     * @see Builder#maxConnecting(int)
+     * @since 4.4
+     */
+    public int getMaxConnecting() {
+        return maxConnecting;
+    }
+
     @Override
     public boolean equals(final Object o) {
         if (this == o) {
@@ -378,7 +416,9 @@ public boolean equals(final Object o) {
         if (!connectionPoolListeners.equals(that.connectionPoolListeners)) {
             return false;
         }
-
+        if (maxConnecting != that.maxConnecting) {
+            return false;
+        }
         return true;
     }
 
@@ -392,6 +432,7 @@ public int hashCode() {
         result = 31 * result + (int) (maintenanceInitialDelayMS ^ (maintenanceInitialDelayMS >>> 32));
         result = 31 * result + (int) (maintenanceFrequencyMS ^ (maintenanceFrequencyMS >>> 32));
         result = 31 * result + connectionPoolListeners.hashCode();
+        result = 31 * result + maxConnecting;
         return result;
     }
 
@@ -406,6 +447,7 @@ public String toString() {
                 + ", maintenanceInitialDelayMS=" + maintenanceInitialDelayMS
                 + ", maintenanceFrequencyMS=" + maintenanceFrequencyMS
                 + ", connectionPoolListeners=" + connectionPoolListeners
+                + ", maxConnecting=" + maxConnecting
                 + '}';
     }
 
@@ -417,6 +459,7 @@ public String toString() {
         isTrue("maxConnectionIdleTime >= 0", builder.maxConnectionIdleTimeMS >= 0);
         isTrue("sizeMaintenanceFrequency > 0", builder.maintenanceFrequencyMS > 0);
         isTrue("maxSize >= minSize", builder.maxSize >= builder.minSize);
+        isTrue("maxConnecting > 0", builder.maxConnecting > 0);
 
         maxSize = builder.maxSize;
         minSize = builder.minSize;
@@ -426,5 +469,6 @@ public String toString() {
         maintenanceInitialDelayMS = builder.maintenanceInitialDelayMS;
         maintenanceFrequencyMS = builder.maintenanceFrequencyMS;
         connectionPoolListeners = unmodifiableList(builder.connectionPoolListeners);
+        maxConnecting = builder.maxConnecting;
     }
 }

driver-core/src/main/com/mongodb/internal/connection/DefaultConnectionPool.java

@@ -100,10 +100,6 @@
 @SuppressWarnings("deprecation")
 class DefaultConnectionPool implements ConnectionPool {
     private static final Logger LOGGER = Loggers.getLogger("connection");
-    /**
-     * Is package-access for the purpose of testing and must not be used for any other purpose outside of this class.
-     */
-    static final int MAX_CONNECTING = 2;
 
     private final ConcurrentPool<UsageTrackingInternalConnection> pool;
     private final ConnectionPoolSettings settings;
@@ -145,7 +141,7 @@ class DefaultConnectionPool implements ConnectionPool {
         this.connectionPoolListener = getConnectionPoolListener(settings);
         backgroundMaintenance = new BackgroundMaintenanceManager();
         connectionPoolCreated(connectionPoolListener, serverId, settings);
-        openConcurrencyLimiter = new OpenConcurrencyLimiter(MAX_CONNECTING);
+        openConcurrencyLimiter = new OpenConcurrencyLimiter(settings.getMaxConnecting());
         asyncWorkManager = new AsyncWorkManager(internalSettings.isPrestartAsyncWorkManager());
         stateAndGeneration = new StateAndGeneration();
         connectionGenerationSupplier = new ConnectionGenerationSupplier() {

driver-core/src/test/functional/com/mongodb/internal/connection/DefaultConnectionPoolTest.java

@@ -58,7 +58,6 @@
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.stream.Stream;
 
-import static com.mongodb.internal.connection.DefaultConnectionPool.MAX_CONNECTING;
 import static java.lang.Long.MAX_VALUE;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.MINUTES;
@@ -81,6 +80,7 @@
 public class DefaultConnectionPoolTest {
     private static final ServerId SERVER_ID = new ServerId(new ClusterId(), new ServerAddress());
     private static final long TEST_WAIT_TIMEOUT_MILLIS = SECONDS.toMillis(5);
+    private static final int DEFAULT_MAX_CONNECTING = ConnectionPoolSettings.builder().build().getMaxConnecting();
 
     private TestInternalConnectionFactory connectionFactory;
 
@@ -331,11 +331,15 @@ private static Stream<Arguments> concurrentUsageArguments() {
         return Stream.of(// variants marked with (*) have proved their usefulness by detecting bugs
                 Arguments.of(0, 1, true, 8, true, false, 0.02f, 0, 0),
                 Arguments.of(0, 1, false, 8, false, true, 0.02f, 0, 0), // (*)
-                Arguments.of(MAX_CONNECTING, MAX_CONNECTING, true, 8, true, true, 0, 0, 0),
-                Arguments.of(MAX_CONNECTING + 1, MAX_CONNECTING + 5, true, 2 * (MAX_CONNECTING + 5), true, true, 0.02f, 0, 0),
-                Arguments.of(MAX_CONNECTING + 5, MAX_CONNECTING + 5, false, 2 * (MAX_CONNECTING + 5), true, true, 0.02f, 0, 0), // (*)
-                Arguments.of(MAX_CONNECTING + 1, MAX_CONNECTING + 5, false, 2 * (MAX_CONNECTING + 5), true, true, 0.3f, 0.1f, 0.1f),
-                Arguments.of(MAX_CONNECTING + 1, MAX_CONNECTING + 5, true, 2 * (MAX_CONNECTING + 5), true, true, 0, 0.5f, 0.05f));
+                Arguments.of(DEFAULT_MAX_CONNECTING, DEFAULT_MAX_CONNECTING, true, 8, true, true, 0, 0, 0),
+                Arguments.of(DEFAULT_MAX_CONNECTING + 1, DEFAULT_MAX_CONNECTING + 5, true, 2 * (DEFAULT_MAX_CONNECTING + 5),
+                        true, true, 0.02f, 0, 0),
+                Arguments.of(DEFAULT_MAX_CONNECTING + 5, DEFAULT_MAX_CONNECTING + 5, false, 2 * (DEFAULT_MAX_CONNECTING + 5),
+                        true, true, 0.02f, 0, 0), // (*)
+                Arguments.of(DEFAULT_MAX_CONNECTING + 1, DEFAULT_MAX_CONNECTING + 5, false, 2 * (DEFAULT_MAX_CONNECTING + 5),
+                        true, true, 0.3f, 0.1f, 0.1f),
+                Arguments.of(DEFAULT_MAX_CONNECTING + 1, DEFAULT_MAX_CONNECTING + 5, true, 2 * (DEFAULT_MAX_CONNECTING + 5),
+                        true, true, 0, 0.5f, 0.05f));
     }
 
     @Test
@@ -346,14 +350,15 @@ public void callbackShouldNotBlockCheckoutIfOpenAsyncWorksNotInCurrentThread() t
         TestConnectionPoolListener listener = new TestConnectionPoolListener();
         provider = new DefaultConnectionPool(SERVER_ID, controllableConnFactory.factory,
                 ConnectionPoolSettings.builder()
-                    .maxSize(MAX_CONNECTING + maxAvailableConnections)
+                    .maxSize(DEFAULT_MAX_CONNECTING + maxAvailableConnections)
                     .addConnectionPoolListener(listener)
                     .maxWaitTime(TEST_WAIT_TIMEOUT_MILLIS, MILLISECONDS)
                     .maintenanceInitialDelay(MAX_VALUE, NANOSECONDS)
                     .build(),
                 mockSdamProvider());
         provider.ready();
-        acquireOpenPermits(provider, MAX_CONNECTING, InfiniteCheckoutEmulation.INFINITE_CALLBACK, controllableConnFactory, listener);
+        acquireOpenPermits(provider, DEFAULT_MAX_CONNECTING, InfiniteCheckoutEmulation.INFINITE_CALLBACK,
+                controllableConnFactory, listener);
         assertUseConcurrently(provider, 2 * maxAvailableConnections,
                 true, true,
                 0.02f, 0, 0,
@@ -364,7 +369,7 @@ public void callbackShouldNotBlockCheckoutIfOpenAsyncWorksNotInCurrentThread() t
      * The idea of this test is as follows:
      * <ol>
      *     <li>Check out some connections from the pool
-     *     ({@link DefaultConnectionPool#MAX_CONNECTING} connections must not be checked out to make the next step possible).</li>
+     *     ({@link #DEFAULT_MAX_CONNECTING} connections must not be checked out to make the next step possible).</li>
      *     <li>Acquire all permits to open a connection and leave them acquired.</li>
      *     <li>Check in the checked out connections and concurrently check them out again.</li>
      * </ol>
@@ -380,7 +385,7 @@ public void checkoutHandOverMechanism() throws InterruptedException, TimeoutExce
         TestConnectionPoolListener listener = new TestConnectionPoolListener();
         provider = new DefaultConnectionPool(SERVER_ID, controllableConnFactory.factory,
                 ConnectionPoolSettings.builder()
-                    .maxSize(MAX_CONNECTING
+                    .maxSize(DEFAULT_MAX_CONNECTING
                             + openConnectionsCount
                             /* This wiggle room is needed to open opportunities to create new connections from the standpoint of
                              * the max pool size, and then check that no connections were created nonetheless. */
@@ -395,7 +400,7 @@ public void checkoutHandOverMechanism() throws InterruptedException, TimeoutExce
         for (int i = 0; i < openConnectionsCount; i++) {
             connections.add(provider.get(0, NANOSECONDS));
         }
-        acquireOpenPermits(provider, MAX_CONNECTING, InfiniteCheckoutEmulation.INFINITE_OPEN, controllableConnFactory, listener);
+        acquireOpenPermits(provider, DEFAULT_MAX_CONNECTING, InfiniteCheckoutEmulation.INFINITE_OPEN, controllableConnFactory, listener);
         int previousIdx = 0;
         // concurrently check in / check out and assert the hand-over mechanism works
         for (int idx = 0; idx < connections.size(); idx += maxConcurrentlyHandedOver) {
@@ -542,7 +547,7 @@ private static void acquireOpenPermits(final DefaultConnectionPool pool, final i
                                            final InfiniteCheckoutEmulation infiniteEmulation,
                                            final ControllableConnectionFactory controllableConnFactory,
                                            final TestConnectionPoolListener listener) throws TimeoutException, InterruptedException {
-        assertTrue(openPermitsCount <= MAX_CONNECTING);
+        assertTrue(openPermitsCount <= DEFAULT_MAX_CONNECTING);
         int initialCreatedEventCount = listener.countEvents(ConnectionCreatedEvent.class);
         switch (infiniteEmulation) {
             case INFINITE_CALLBACK: {