Fix race condition accessing ServerDescription

This change guarantees that any operation accessing the ServerDescription
via ConnectionSource will see a Server Description that is valid
(i.e consistent with the state of the server when it was selected), though
perhaps stale, depending on how long the ConnectionSource is retained.

JAVA-3911

Author: jyemin

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

@@ -21,7 +21,6 @@
 import com.mongodb.annotations.Immutable;
 import com.mongodb.annotations.NotThreadSafe;
 import com.mongodb.event.ClusterListener;
-import com.mongodb.internal.connection.Cluster;
 import com.mongodb.internal.connection.ServerAddressHelper;
 import com.mongodb.internal.selector.LatencyMinimizingServerSelector;
 import com.mongodb.lang.Nullable;
@@ -389,7 +388,6 @@ public String getRequiredReplicaSetName() {
      * </ul>
      *
      * @return the server selector, which may be null
-     * @see Cluster#selectServer(com.mongodb.selector.ServerSelector)
      */
     @Nullable
     public ServerSelector getServerSelector() {

driver-core/src/main/com/mongodb/internal/async/client/ClientSessionBinding.java

@@ -25,7 +25,7 @@
 import com.mongodb.internal.binding.AsyncConnectionSource;
 import com.mongodb.internal.binding.AsyncReadWriteBinding;
 import com.mongodb.internal.connection.AsyncConnection;
-import com.mongodb.internal.connection.Server;
+import com.mongodb.internal.connection.ServerTuple;
 import com.mongodb.internal.selector.ReadPreferenceServerSelector;
 import com.mongodb.internal.session.ClientSessionContext;
 import com.mongodb.internal.session.SessionContext;
@@ -76,13 +76,13 @@ public SessionContext getSessionContext() {
     private void getPinnedConnectionSource(final SingleResultCallback<AsyncConnectionSource> callback) {
         if (session.getPinnedServerAddress() == null) {
             wrapped.getCluster().selectServerAsync(
-                    new ReadPreferenceServerSelector(wrapped.getReadPreference()), new SingleResultCallback<Server>() {
+                    new ReadPreferenceServerSelector(wrapped.getReadPreference()), new SingleResultCallback<ServerTuple>() {
                         @Override
-                        public void onResult(final Server server, final Throwable t) {
+                        public void onResult(final ServerTuple serverTuple, final Throwable t) {
                             if (t != null) {
                                 callback.onResult(null, t);
                             } else {
-                                session.setPinnedServerAddress(server.getDescription().getAddress());
+                                session.setPinnedServerAddress(serverTuple.getServerDescription().getAddress());
                                 wrapped.getConnectionSource(session.getPinnedServerAddress(), new WrappingCallback(callback));
                             }
                         }

driver-core/src/main/com/mongodb/internal/binding/AsyncClusterBinding.java

@@ -25,6 +25,7 @@
 import com.mongodb.internal.connection.Cluster;
 import com.mongodb.internal.connection.ReadConcernAwareNoOpSessionContext;
 import com.mongodb.internal.connection.Server;
+import com.mongodb.internal.connection.ServerTuple;
 import com.mongodb.internal.selector.ReadPreferenceServerSelector;
 import com.mongodb.internal.selector.ServerAddressSelector;
 import com.mongodb.internal.selector.WritableServerSelector;
@@ -96,29 +97,31 @@ public void getConnectionSource(final ServerAddress serverAddress, final SingleR
 
     private void getAsyncClusterBindingConnectionSource(final ServerSelector serverSelector,
                                                         final SingleResultCallback<AsyncConnectionSource> callback) {
-        cluster.selectServerAsync(serverSelector, new SingleResultCallback<Server>() {
+        cluster.selectServerAsync(serverSelector, new SingleResultCallback<ServerTuple>() {
             @Override
-            public void onResult(final Server result, final Throwable t) {
+            public void onResult(final ServerTuple result, final Throwable t) {
                 if (t != null) {
                     callback.onResult(null, t);
                 } else {
-                    callback.onResult(new AsyncClusterBindingConnectionSource(result), null);
+                    callback.onResult(new AsyncClusterBindingConnectionSource(result.getServer(), result.getServerDescription()), null);
                 }
             }
         });
     }
 
     private final class AsyncClusterBindingConnectionSource extends AbstractReferenceCounted implements AsyncConnectionSource {
         private final Server server;
+        private final ServerDescription serverDescription;
 
-        private AsyncClusterBindingConnectionSource(final Server server) {
+        private AsyncClusterBindingConnectionSource(final Server server, final ServerDescription serverDescription) {
             this.server = server;
+            this.serverDescription = serverDescription;
             AsyncClusterBinding.this.retain();
         }
 
         @Override
         public ServerDescription getServerDescription() {
-            return server.getDescription();
+            return serverDescription;
         }
 
         @Override

driver-core/src/main/com/mongodb/internal/binding/AsyncSingleServerBinding.java

@@ -24,6 +24,7 @@
 import com.mongodb.internal.connection.Connection;
 import com.mongodb.internal.connection.ReadConcernAwareNoOpSessionContext;
 import com.mongodb.internal.connection.Server;
+import com.mongodb.internal.connection.ServerTuple;
 import com.mongodb.internal.selector.ReadPreferenceServerSelector;
 import com.mongodb.internal.selector.ServerAddressSelector;
 import com.mongodb.internal.selector.WritableServerSelector;
@@ -98,15 +99,18 @@ public ConnectionSource getConnectionSource(final ServerAddress serverAddress) {
 
     private final class ClusterBindingConnectionSource extends AbstractReferenceCounted implements ConnectionSource {
         private final Server server;
+        private final ServerDescription serverDescription;
 
         private ClusterBindingConnectionSource(final ServerSelector serverSelector) {
-            this.server = cluster.selectServer(serverSelector);
+            ServerTuple serverTuple = cluster.selectServer(serverSelector);
+            this.server = serverTuple.getServer();
+            this.serverDescription = serverTuple.getServerDescription();
             ClusterBinding.this.retain();
         }
 
         @Override
         public ServerDescription getServerDescription() {
-            return server.getDescription();
+            return serverDescription;
         }
 
         @Override

driver-core/src/main/com/mongodb/internal/binding/ClusterBinding.java

@@ -22,7 +22,7 @@
 import com.mongodb.internal.connection.Cluster;
 import com.mongodb.internal.connection.Connection;
 import com.mongodb.internal.connection.NoOpSessionContext;
-import com.mongodb.internal.connection.Server;
+import com.mongodb.internal.connection.ServerTuple;
 import com.mongodb.internal.selector.ServerAddressSelector;
 import com.mongodb.internal.session.SessionContext;
 
@@ -86,16 +86,17 @@ public SingleServerBinding retain() {
     }
 
     private final class SingleServerBindingConnectionSource extends AbstractReferenceCounted implements ConnectionSource {
-        private final Server server;
+        private final ServerDescription serverDescription;
 
         private SingleServerBindingConnectionSource() {
             SingleServerBinding.this.retain();
-            server = cluster.selectServer(new ServerAddressSelector(serverAddress));
+            ServerTuple serverTuple = cluster.selectServer(new ServerAddressSelector(serverAddress));
+            serverDescription = serverTuple.getServerDescription();
         }
 
         @Override
         public ServerDescription getServerDescription() {
-            return server.getDescription();
+            return serverDescription;
         }
 
         @Override
@@ -105,7 +106,7 @@ public SessionContext getSessionContext() {
 
         @Override
         public Connection getConnection() {
-            return cluster.selectServer(new ServerAddressSelector(serverAddress)).getConnection();
+            return cluster.selectServer(new ServerAddressSelector(serverAddress)).getServer().getConnection();
         }
 
         @Override