JAVA-1285: When server selection fails to find a match, await only for connectRetryFrequency, instead of maxWaitTime, so that attempts to re-connect to each server happen more frequently.

Author: jyemin

src/main/com/mongodb/BaseCluster.java

@@ -59,34 +59,40 @@ public BaseCluster(final String clusterId, final ClusterSettings settings, final
     }
 
     @Override
-    public Server getServer(final ServerSelector serverSelector, final long maxWaitTime, final TimeUnit timeUnit) {
+    public Server getServer(final ServerSelector serverSelector, final long maxWaitTime, final TimeUnit maxWaitTimeTimeUnit,
+                            final long connectRetryFrequency, final TimeUnit connectRetryFrequencyTimeUnit) {
         isTrue("open", !isClosed());
 
         try {
             CountDownLatch currentPhase = phase.get();
             ClusterDescription curDescription = description;
             List<ServerDescription> serverDescriptions = serverSelector.choose(curDescription);
-            final long endTime = System.nanoTime() + NANOSECONDS.convert(maxWaitTime, timeUnit);
+            final long endTime = System.nanoTime() + NANOSECONDS.convert(maxWaitTime, maxWaitTimeTimeUnit);
+            boolean selectionFailureLogged = false;
             while (true) {
                 throwIfIncompatible(curDescription);
+
                 if (!serverDescriptions.isEmpty()) {
                     ClusterableServer server = getRandomServer(new ArrayList<ServerDescription>(serverDescriptions));
                     if (server != null) {
                         return new WrappedServer(server);
                     }
                 }
 
-                final long timeout = endTime - System.nanoTime();
-
-                LOGGER.info(format("No server chosen by %s from cluster description %s. Waiting for %d ms before timing out",
-                                   serverSelector, curDescription, MILLISECONDS.convert(timeout, NANOSECONDS)));
+                if (System.nanoTime() > endTime) {
+                    throw new MongoTimeoutException(format("Timed out while waiting for a server that matches %s after %d ms",
+                                                           serverSelector, MILLISECONDS.convert(maxWaitTime, maxWaitTimeTimeUnit)));
+                }
 
+                if (!selectionFailureLogged) {
+                    LOGGER.info(format("No server chosen by %s from cluster description %s. Waiting for %d ms before timing out",
+                                       serverSelector, curDescription, MILLISECONDS.convert(maxWaitTime, maxWaitTimeTimeUnit)));
+                    selectionFailureLogged = true;
+                }
                 connect();
 
-                if (!currentPhase.await(timeout, NANOSECONDS)) {
-                    throw new MongoTimeoutException(format("Timed out while waiting for a server that matches %s after %d ms",
-                                                           serverSelector, MILLISECONDS.convert(maxWaitTime, timeUnit)));
-                }
+                currentPhase.await(connectRetryFrequency, connectRetryFrequencyTimeUnit);
+
                 currentPhase = phase.get();
                 curDescription = description;
                 serverDescriptions = serverSelector.choose(curDescription);
@@ -97,25 +103,32 @@ public Server getServer(final ServerSelector serverSelector, final long maxWaitT
     }
 
     @Override
-    public ClusterDescription getDescription(final long maxWaitTime, final TimeUnit timeUnit) {
+    public ClusterDescription getDescription(final long maxWaitTime, final TimeUnit maxWaitTimeTimeUnit,
+                                             final long connectRetryFrequency, final TimeUnit connectRetryFrequencyTimeUnit) {
         isTrue("open", !isClosed());
 
         try {
             CountDownLatch currentPhase = phase.get();
             ClusterDescription curDescription = description;
-            final long endTime = System.nanoTime() + NANOSECONDS.convert(maxWaitTime, timeUnit);
+            final long endTime = System.nanoTime() + NANOSECONDS.convert(maxWaitTime, maxWaitTimeTimeUnit);
+            boolean selectionFailureLogged = false;
             while (curDescription.getType() == ClusterType.Unknown) {
-                final long timeout = endTime - System.nanoTime();
 
-                LOGGER.info(format("Cluster description not yet available. Waiting for %d ms before timing out",
-                                   MILLISECONDS.convert(timeout, NANOSECONDS)));
+                if (System.nanoTime() > endTime) {
+                    throw new MongoTimeoutException(format("Timed out while waiting to connect after %d ms",
+                                                           MILLISECONDS.convert(maxWaitTime, maxWaitTimeTimeUnit)));
+                }
+
+                if (!selectionFailureLogged) {
+                    LOGGER.info(format("Cluster description not yet available. Waiting for %d ms before timing out",
+                                       MILLISECONDS.convert(maxWaitTime, maxWaitTimeTimeUnit)));
+                    selectionFailureLogged = true;
+                }
 
                 connect();
 
-                if (!currentPhase.await(timeout, NANOSECONDS)) {
-                    throw new MongoTimeoutException(format("Timed out while waiting to connect after %d ms",
-                                                           MILLISECONDS.convert(maxWaitTime, timeUnit)));
-                }
+                currentPhase.await(connectRetryFrequency, connectRetryFrequencyTimeUnit);
+
                 currentPhase = phase.get();
                 curDescription = description;
             }

src/main/com/mongodb/Cluster.java

@@ -20,9 +20,11 @@
 
 interface Cluster {
 
-    ClusterDescription getDescription(final long maxWaitTime, final TimeUnit timeUnit);
+    ClusterDescription getDescription(final long maxWaitTime, final TimeUnit maxWaitTimeTimeUnit,
+                                      long connectRetryFrequency, TimeUnit connectRetryFrequencyTimeUnit);
 
-    Server getServer(ServerSelector serverSelector, final long maxWaitTime, final TimeUnit timeUnit);
+    Server getServer(ServerSelector serverSelector, final long maxWaitTime, final TimeUnit maxWaitTimeTimeUnit,
+                     long connectRetryFrequency, TimeUnit connectRetryFrequencyTimeUnit);
 
     void close();
 

src/main/com/mongodb/DBTCPConnector.java

@@ -393,7 +393,8 @@ private boolean shouldRetryQuery(ReadPreference readPreference, final DBCollecti
     }
 
     private ClusterDescription getClusterDescription() {
-        return cluster.getDescription(getClusterWaitTimeMS(), MILLISECONDS);
+        return cluster.getDescription(getClusterWaitTimeMS(), MILLISECONDS,
+                                      _mongo.getMongoOptions().heartbeatConnectRetryFrequencyMS, MILLISECONDS);
     }
 
     private int getClusterWaitTimeMS() {
@@ -651,6 +652,7 @@ MyPort getMyPort() {
     }
 
     private Server getServer(final ServerSelector serverSelector) {
-        return cluster.getServer(serverSelector, getClusterWaitTimeMS(), MILLISECONDS);
+        return cluster.getServer(serverSelector, getClusterWaitTimeMS(), MILLISECONDS,
+                                 _mongo.getMongoOptions().heartbeatConnectRetryFrequencyMS, MILLISECONDS);
     }
 }

src/main/com/mongodb/ServerMonitor.java

@@ -115,8 +115,7 @@ public void run() {
                     if (!isClosed) {
                         try {
                             logStateChange(currentServerDescription, throwable);
-                            serverStateListener.stateChanged(new ChangeEvent<ServerDescription>(currentServerDescription,
-                                                                                                serverDescription));
+                            sendStateChangedEvent(currentServerDescription);
                         } catch (Throwable t) {
                             LOGGER.log(Level.WARNING, "Exception in monitor thread during notification of server state change", t);
                         }
@@ -130,6 +129,13 @@ public void run() {
             }
         }
 
+        private void sendStateChangedEvent(final ServerDescription currentServerDescription) {
+            if (!currentServerDescription.equals(serverDescription) ||
+                currentServerDescription.getAverageLatencyNanos() != serverDescription.getAverageLatencyNanos()) {
+                serverStateListener.stateChanged(new ChangeEvent<ServerDescription>(currentServerDescription, serverDescription));
+            }
+        }
+
         private void logStateChange(final ServerDescription currentServerDescription, final Throwable throwable) {
             // Note that the ServerDescription.equals method does not include the average ping time as part of the comparison,
             // so this will not spam the logs too hard.

src/test/com/mongodb/MultiServerClusterSpecification.groovy

@@ -280,7 +280,7 @@ class MultiServerClusterSpecification extends Specification {
         cluster.close()
 
         when:
-        cluster.getServer(new ReadPreferenceServerSelector(ReadPreference.primary()), 1, SECONDS)
+        cluster.getServer(new ReadPreferenceServerSelector(ReadPreference.primary()), 1, SECONDS, 1, SECONDS)
 
         then:
         thrown(IllegalStateException)
@@ -428,7 +428,7 @@ class MultiServerClusterSpecification extends Specification {
     }
 
     def getClusterDescription(MultiServerCluster cluster) {
-        cluster.getDescription(1, MILLISECONDS)
+        cluster.getDescription(1, MILLISECONDS, 1, MILLISECONDS)
     }
 
     def getServerDescription(ServerAddress server) {

src/test/com/mongodb/SingleServerClusterSpecification.groovy

@@ -151,7 +151,7 @@ class SingleServerClusterSpecification extends Specification {
         sendNotification(firstServer, getBuilder(firstServer).minWireVersion(1000).maxWireVersion(1000).build())
 
         when:
-        cluster.getServer(new ReadPreferenceServerSelector(ReadPreference.primary()), 1, SECONDS)
+        cluster.getServer(new ReadPreferenceServerSelector(ReadPreference.primary()), 1, SECONDS, 1, SECONDS)
 
         then:
         thrown(MongoIncompatibleDriverException)
@@ -186,7 +186,7 @@ class SingleServerClusterSpecification extends Specification {
     }
 
     def getClusterDescription(Cluster cluster) {
-        cluster.getDescription(1, MILLISECONDS)
+        cluster.getDescription(1, MILLISECONDS, 1, MILLISECONDS)
     }
 
     def getServerDescriptions() {