JAVA-2824: Stop authenticating server monitor connections

This is a significant behavior change in the driver, as authentication
errors will be thrown much faster to applications, as a
MongoSecurityException rather than as a MongoTimeoutException after the
server selection timeout (which defaults to 30 seconds) has expired.

Author: jyemin

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

@@ -64,10 +64,12 @@ public ClusterableServer create(final ServerAddress serverAddress, final ServerL
         ConnectionPool connectionPool = new DefaultConnectionPool(new ServerId(clusterId, serverAddress),
                 new InternalStreamConnectionFactory(streamFactory, credentialList, applicationName,
                         mongoDriverInformation, compressorList, commandListener), connectionPoolSettings);
+
+        // no credentials, compressor list, or command listener for the server monitor factory
         ServerMonitorFactory serverMonitorFactory =
             new DefaultServerMonitorFactory(new ServerId(clusterId, serverAddress), serverSettings, clusterClock,
-                    new InternalStreamConnectionFactory(heartbeatStreamFactory, credentialList, applicationName,
-                            mongoDriverInformation, Collections.<MongoCompressor>emptyList(), null), connectionPool);
+                    new InternalStreamConnectionFactory(heartbeatStreamFactory, Collections.<MongoCredentialWithCache>emptyList(),
+                            applicationName, mongoDriverInformation, Collections.<MongoCompressor>emptyList(), null), connectionPool);
 
         return new DefaultServer(new ServerId(clusterId, serverAddress), clusterSettings.getMode(), connectionPool,
                 new DefaultConnectionFactory(), serverMonitorFactory, serverListener, commandListener, clusterClock);

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

@@ -79,7 +79,7 @@ public Connection getConnection() {
         try {
             return connectionFactory.create(connectionPool.get(), new DefaultServerProtocolExecutor(), clusterConnectionMode);
         } catch (MongoSecurityException e) {
-            invalidate();
+            connectionPool.invalidate();
             throw e;
         } catch (MongoSocketException e) {
             invalidate();
@@ -93,7 +93,9 @@ public void getConnectionAsync(final SingleResultCallback<AsyncConnection> callb
         connectionPool.getAsync(new SingleResultCallback<InternalConnection>() {
             @Override
             public void onResult(final InternalConnection result, final Throwable t) {
-                if (t instanceof MongoSecurityException || t instanceof MongoSocketException) {
+                if (t instanceof MongoSecurityException) {
+                    connectionPool.invalidate();
+                } else if (t instanceof MongoSocketException) {
                     invalidate();
                 }
                 if (t != null) {

driver-core/src/test/functional/com/mongodb/operation/UserOperationsSpecification.groovy

@@ -19,8 +19,8 @@ package com.mongodb.operation
 import category.Slow
 import com.mongodb.MongoCredential
 import com.mongodb.MongoNamespace
+import com.mongodb.MongoSecurityException
 import com.mongodb.MongoServerException
-import com.mongodb.MongoTimeoutException
 import com.mongodb.MongoWriteConcernException
 import com.mongodb.OperationFunctionalSpecification
 import com.mongodb.ReadPreference
@@ -135,12 +135,13 @@ class UserOperationsSpecification extends OperationFunctionalSpecification {
         given:
         def credential = createCredential('user', databaseName, 'pencil' as char[])
         def cluster = getCluster(credential, ClusterSettings.builder().serverSelectionTimeout(1, TimeUnit.SECONDS))
+        def server = cluster.selectServer(new WritableServerSelector())
 
         when:
-        cluster.selectServer(new WritableServerSelector())
+        server.getConnection()
 
         then:
-        thrown(MongoTimeoutException)
+        thrown(MongoSecurityException)
 
         cleanup:
         cluster?.close()
@@ -155,12 +156,13 @@ class UserOperationsSpecification extends OperationFunctionalSpecification {
         execute(new CreateUserOperation(credential, true), async)
         execute(new DropUserOperation(databaseName, credential.userName), async)
         def cluster = getCluster(ClusterSettings.builder().serverSelectionTimeout(1, TimeUnit.SECONDS))
+        def server = cluster.selectServer(new WritableServerSelector())
 
         when:
-        cluster.selectServer(new WritableServerSelector())
+        server.getConnection()
 
         then:
-        thrown(MongoTimeoutException)
+        thrown(MongoSecurityException)
 
         cleanup:
         cluster?.close()

driver-core/src/test/unit/com/mongodb/connection/DefaultServerSpecification.groovy

@@ -156,14 +156,41 @@ class DefaultServerSpecification extends Specification {
 
         where:
         exceptionToThrow << [
-                new MongoSecurityException(createCredential('jeff', 'admin', '123'.toCharArray()), 'Auth failed'),
                 new MongoSocketOpenException('open failed', new ServerAddress(), new IOException()),
                 new MongoSocketWriteException('Write failed', new ServerAddress(), new IOException()),
                 new MongoSocketReadException('Read failed', new ServerAddress(), new IOException()),
                 new MongoSocketReadTimeoutException('Read timed out', new ServerAddress(), new IOException()),
         ]
     }
 
+    def 'failed authentication should invalidate the connection pool'() {
+        given:
+        def clusterTime = new ClusterClock()
+        def connectionPool = Mock(ConnectionPool)
+        def connectionFactory = Mock(ConnectionFactory)
+        def serverMonitorFactory = Stub(ServerMonitorFactory)
+        def serverMonitor = Mock(ServerMonitor)
+        connectionPool.get() >> { throw exceptionToThrow }
+        serverMonitorFactory.create(_) >> { serverMonitor }
+
+        def server = new DefaultServer(serverId, SINGLE, connectionPool, connectionFactory, serverMonitorFactory,
+                NO_OP_SERVER_LISTENER, null, clusterTime)
+
+        when:
+        server.getConnection()
+
+        then:
+        def e = thrown(MongoSecurityException)
+        e.is(exceptionToThrow)
+        1 * connectionPool.invalidate()
+        0 * serverMonitor.connect()
+
+        where:
+        exceptionToThrow << [
+                new MongoSecurityException(createCredential('jeff', 'admin', '123'.toCharArray()), 'Auth failed'),
+        ]
+    }
+
     def 'failed open should invalidate the server asynchronously'() {
         given:
         def clusterTime = new ClusterClock()
@@ -192,14 +219,45 @@ class DefaultServerSpecification extends Specification {
 
         where:
         exceptionToThrow << [
-                new MongoSecurityException(createCredential('jeff', 'admin', '123'.toCharArray()), 'Auth failed'),
                 new MongoSocketOpenException('open failed', new ServerAddress(), new IOException()),
                 new MongoSocketWriteException('Write failed', new ServerAddress(), new IOException()),
                 new MongoSocketReadException('Read failed', new ServerAddress(), new IOException()),
                 new MongoSocketReadTimeoutException('Read timed out', new ServerAddress(), new IOException()),
         ]
     }
 
+    def 'failed auth should invalidate the connection pool asynchronously'() {
+        given:
+        def clusterTime = new ClusterClock()
+        def connectionPool = Mock(ConnectionPool)
+        def connectionFactory = Mock(ConnectionFactory)
+        def serverMonitorFactory = Stub(ServerMonitorFactory)
+        def serverMonitor = Mock(ServerMonitor)
+        connectionPool.getAsync(_) >> { it[0].onResult(null, exceptionToThrow) }
+        serverMonitorFactory.create(_) >> { serverMonitor }
+        def server = new DefaultServer(serverId, SINGLE, connectionPool, connectionFactory,
+                serverMonitorFactory, NO_OP_SERVER_LISTENER, null, clusterTime)
+
+        when:
+        def latch = new CountDownLatch(1)
+        def receivedConnection = null
+        def receivedThrowable = null
+        server.getConnectionAsync { result, throwable -> receivedConnection = result; receivedThrowable = throwable; latch.countDown() }
+        latch.await()
+
+        then:
+        !receivedConnection
+        receivedThrowable.is(exceptionToThrow)
+        1 * connectionPool.invalidate()
+        0 * serverMonitor.connect()
+
+
+        where:
+        exceptionToThrow << [
+                new MongoSecurityException(createCredential('jeff', 'admin', '123'.toCharArray()), 'Auth failed'),
+        ]
+    }
+
     def 'should invalidate on MongoNotPrimaryException'() {
         given:
         def clusterTime = new ClusterClock()