Limit server selection deprioritization to system overloaded errors on replica sets

- Gate server deprioritization in onAttemptFailure: sharded clusters deprioritize on any error, all other topologies only on SystemOverloadedError
- Pass ClusterType to updateCandidate so onAttemptFailure can distinguish topology types
- Parameterize ServerDeprioritizationTest by cluster type and error type to cover conditional deprioritization
- Add retryable reads prose tests 3.1 and 3.2

JAVA-6105

Author: vbabanin

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

@@ -163,7 +163,7 @@ public ServerTuple selectServer(final ServerSelector serverSelector, final Opera
             if (serverTuple != null) {
                 ServerAddress serverAddress = serverTuple.getServerDescription().getAddress();
                 logServerSelectionSucceeded(operationContext, clusterId, serverAddress, serverSelector, currentDescription);
-                serverDeprioritization.updateCandidate(serverAddress);
+                serverDeprioritization.updateCandidate(serverAddress, currentDescription.getType());
                 return serverTuple;
             }
             computedServerSelectionTimeout.onExpired(() ->
@@ -306,7 +306,7 @@ private boolean handleServerSelectionRequest(
                 if (serverTuple != null) {
                     ServerAddress serverAddress = serverTuple.getServerDescription().getAddress();
                     logServerSelectionSucceeded(operationContext, clusterId, serverAddress, request.originalSelector, description);
-                    serverDeprioritization.updateCandidate(serverAddress);
+                    serverDeprioritization.updateCandidate(serverAddress, description.getType());
                     request.onResult(serverTuple, null);
                     return true;
                 }

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

@@ -17,11 +17,13 @@
 
 import com.mongodb.Function;
 import com.mongodb.MongoConnectionPoolClearedException;
+import com.mongodb.MongoException;
 import com.mongodb.ReadConcern;
 import com.mongodb.RequestContext;
 import com.mongodb.ServerAddress;
 import com.mongodb.ServerApi;
 import com.mongodb.connection.ClusterDescription;
+import com.mongodb.connection.ClusterType;
 import com.mongodb.connection.ServerDescription;
 import com.mongodb.internal.IgnorableRequestContext;
 import com.mongodb.internal.TimeoutContext;
@@ -215,6 +217,7 @@ public OperationContext withOverride(final TimeoutContextOverride timeoutContext
     public static final class ServerDeprioritization {
         @Nullable
         private ServerAddress candidate;
+        private ClusterType clusterType;
         private final Set<ServerAddress> deprioritized;
 
         private ServerDeprioritization() {
@@ -236,16 +239,21 @@ ServerSelector apply(final ServerSelector wrappedSelector) {
         }
 
         @VisibleForTesting(otherwise = PACKAGE)
-        public void updateCandidate(final ServerAddress serverAddress) {
-            candidate = serverAddress;
+        public void updateCandidate(final ServerAddress serverAddress, final ClusterType clusterType) {
+            this.candidate = serverAddress;
+            this.clusterType = clusterType;
         }
 
         public void onAttemptFailure(final Throwable failure) {
             if (candidate == null || failure instanceof MongoConnectionPoolClearedException) {
                 candidate = null;
                 return;
             }
-            deprioritized.add(candidate);
+
+            boolean isSystemOverloadedError = failure instanceof MongoException && ((MongoException) failure).hasErrorLabel("SystemOverloadedError");
+            if (clusterType == ClusterType.SHARDED || isSystemOverloadedError) {
+                deprioritized.add(candidate);
+            }
         }
 
         /**

driver-core/src/test/unit/com/mongodb/connection/ServerSelectionSelectionTest.java

@@ -67,11 +67,10 @@
 
 import static com.mongodb.ClusterFixture.TIMEOUT_SETTINGS;
 import static com.mongodb.connection.ServerDescription.MIN_DRIVER_WIRE_VERSION;
-import static java.lang.String.format;
+import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
 import static org.junit.Assume.assumeFalse;
-import static org.junit.Assert.assertNotNull;
 import static org.mockito.Mockito.when;
 
 /**
@@ -299,8 +298,12 @@ private OperationContext createOperationContext() {
                         new TimeoutContext(TIMEOUT_SETTINGS.withServerSelectionTimeoutMS(0)));
         OperationContext.ServerDeprioritization serverDeprioritization = operationContext.getServerDeprioritization();
         for (ServerAddress address : extractDeprioritizedServerAddresses(definition)) {
-            serverDeprioritization.updateCandidate(address);
-            serverDeprioritization.onAttemptFailure(new MongoException("test"));
+            serverDeprioritization.updateCandidate(address, clusterDescription.getType());
+            // The spec defines deprioritized_servers as a pre-populated list to feed into the selection mechanism - not as "simulate the
+            // failure that caused deprioritization." Thus, SystemOverloadedError used unconditionally regardless of the cluster type.
+            MongoException error = new MongoException("test");
+            error.addLabel("SystemOverloadedError");
+            serverDeprioritization.onAttemptFailure(error);
         }
         return operationContext;
     }
@@ -317,7 +320,7 @@ private static Cluster.ServersSnapshot createServersSnapshot(
         return serverMap::get;
     }
 
-        private static void validateTestDescriptionFields(final Set<String> actualFields, final Set<String> knownFields) {
+    private static void validateTestDescriptionFields(final Set<String> actualFields, final Set<String> knownFields) {
         Set<String> unknownFields = new HashSet<>(actualFields);
         unknownFields.removeAll(knownFields);
         if (!unknownFields.isEmpty()) {