Ensure callback using cluster description is guarded (#878)

Getting the description from a cluster can throw a MongoTimeoutException
or a MongoInterruptedException, so in that case the exception should be 
passed to the callback.

JAVA-4432

Co-authored-by: Valentin Kovalenko <[email protected]>

Author: 2 people

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

@@ -58,19 +58,27 @@ public ReadPreference getReadPreference() {
 
     @Override
     public void getReadConnectionSource(final SingleResultCallback<AsyncConnectionSource> callback) {
-        if (isConnectionSourcePinningRequired()) {
-            getPinnedConnectionSource(true, callback);
-        } else {
-            wrapped.getReadConnectionSource(new WrappingCallback(callback));
-        }
+        isConnectionSourcePinningRequired((isConnectionSourcePinningRequired, t) -> {
+            if (t != null) {
+                callback.onResult(null, t);
+            } else if (isConnectionSourcePinningRequired) {
+                getPinnedConnectionSource(true, callback);
+            } else {
+                wrapped.getReadConnectionSource(new WrappingCallback(callback));
+            }
+        });
     }
 
     public void getWriteConnectionSource(final SingleResultCallback<AsyncConnectionSource> callback) {
-        if (isConnectionSourcePinningRequired()) {
-            getPinnedConnectionSource(false, callback);
-        } else {
-            wrapped.getWriteConnectionSource(new WrappingCallback(callback));
-        }
+        isConnectionSourcePinningRequired((isConnectionSourcePinningRequired, t) -> {
+            if (t != null) {
+                callback.onResult(null, t);
+            } else if (isConnectionSourcePinningRequired) {
+                getPinnedConnectionSource(false, callback);
+            } else {
+                wrapped.getWriteConnectionSource(new WrappingCallback(callback));
+            }
+        });
     }
 
     @Override
@@ -138,6 +146,14 @@ public int release() {
         return count;
     }
 
+    private void isConnectionSourcePinningRequired(final SingleResultCallback<Boolean> callback) {
+        try {
+            callback.onResult(isConnectionSourcePinningRequired(), null);
+        } catch (RuntimeException e) {
+            callback.onResult(null, e);
+        }
+    }
+
     private boolean isConnectionSourcePinningRequired() {
         ClusterType clusterType = wrapped.getCluster().getDescription().getType();
         return session.hasActiveTransaction() && (clusterType == ClusterType.SHARDED || clusterType == LOAD_BALANCED);

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

@@ -51,6 +51,7 @@ public interface Cluster extends Closeable {
      *
      * @return a ClusterDescription representing the current state of the cluster
      * @throws com.mongodb.MongoTimeoutException if the timeout has been reached before the cluster type is known
+     * @throws com.mongodb.MongoInterruptedException if interrupted when getting the cluster description
      */
     ClusterDescription getDescription();