Client Backpressure

driver-core/src/main/com/mongodb/MongoException.java

@@ -39,6 +39,7 @@ public class MongoException extends RuntimeException {
      *
      * @see #hasErrorLabel(String)
      * @since 3.8
+     * @mongodb.driver.manual core/transactions-in-applications/#std-label-transient-transaction-error
      */
     public static final String TRANSIENT_TRANSACTION_ERROR_LABEL = "TransientTransactionError";
 
@@ -47,9 +48,32 @@ public class MongoException extends RuntimeException {
      *
      * @see #hasErrorLabel(String)
      * @since 3.8
+     * @mongodb.driver.manual core/transactions-in-applications/#std-label-unknown-transaction-commit-result
      */
     public static final String UNKNOWN_TRANSACTION_COMMIT_RESULT_LABEL = "UnknownTransactionCommitResult";
 
+    /**
+     * Server is overloaded and shedding load.
+     * If you retry, use exponential backoff because the server has indicated overload.
+     * This label on its own does not mean that the operation can be safely retried.
+     *
+     * @see #hasErrorLabel(String)
+     * @since 5.7
+     * @mongodb.server.release 8.3
+     */
+    // TODO-BACKPRESSURE Valentin Add a @mongodb.driver.manual link or something similar, see `content/atlas/source/overload-errors.txt` in https://github.com/10gen/docs-mongodb-internal/pull/17281
+    public static final String SYSTEM_OVERLOADED_ERROR_LABEL = "SystemOverloadedError";
+
+    /**
+     * The operation was not executed and is safe to retry.
+     *
+     * @see #hasErrorLabel(String)
+     * @since 5.7
+     * @mongodb.server.release 8.3
+     */
+    // TODO-BACKPRESSURE Valentin Add a @mongodb.driver.manual link or something similar, see `content/atlas/source/overload-errors.txt` in https://github.com/10gen/docs-mongodb-internal/pull/17281
+    public static final String RETRYABLE_ERROR_LABEL = "RetryableError";
+
     private static final long serialVersionUID = -4415279469780082174L;
 
     private final int code;

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

@@ -172,7 +172,8 @@ private InternalConnectionInitializationDescription createInitializationDescript
 
     private BsonDocument createHelloCommand(final Authenticator authenticator, final InternalConnection connection) {
         BsonDocument helloCommandDocument = new BsonDocument(getHandshakeCommandName(), new BsonInt32(1))
-                .append("helloOk", BsonBoolean.TRUE);
+                .append("helloOk", BsonBoolean.TRUE)
+                .append("backpressure", BsonBoolean.TRUE);
         if (clientMetadataDocument != null) {
             helloCommandDocument.append("client", clientMetadataDocument);
         }

driver-core/src/main/com/mongodb/internal/event/CommandListenerMulticaster.java

@@ -16,6 +16,7 @@
 
 package com.mongodb.internal.event;
 
+import com.mongodb.annotations.ThreadSafe;
 import com.mongodb.event.CommandFailedEvent;
 import com.mongodb.event.CommandListener;
 import com.mongodb.event.CommandStartedEvent;
@@ -29,7 +30,11 @@
 import static com.mongodb.assertions.Assertions.isTrue;
 import static java.lang.String.format;
 
-
+/**
+ * This {@link CommandListener} is {@linkplain ThreadSafe thread-safe},
+ * provided that the recipient listeners passed to {@link #CommandListenerMulticaster(List)} are.
+ */
+@ThreadSafe
 final class CommandListenerMulticaster implements CommandListener {
     private static final Logger LOGGER = Loggers.getLogger("protocol.event");
 

driver-core/src/main/com/mongodb/internal/operation/CommandOperationHelper.java

@@ -51,7 +51,7 @@
 import static java.util.Arrays.asList;
 
 @SuppressWarnings("overloads")
-final class CommandOperationHelper {
+public final class CommandOperationHelper {
     static WriteConcern validateAndGetEffectiveWriteConcern(final WriteConcern writeConcernSetting, final SessionContext sessionContext)
             throws MongoClientException {
         boolean activeTransaction = sessionContext.hasActiveTransaction();
@@ -223,8 +223,8 @@ static boolean isRetryWritesEnabled(@Nullable final BsonDocument command) {
                 || command.getFirstKey().equals("commitTransaction") || command.getFirstKey().equals("abortTransaction")));
     }
 
-    static final String RETRYABLE_WRITE_ERROR_LABEL = "RetryableWriteError";
-    private static final String NO_WRITES_PERFORMED_ERROR_LABEL = "NoWritesPerformed";
+    public static final String RETRYABLE_WRITE_ERROR_LABEL = "RetryableWriteError";
+    public static final String NO_WRITES_PERFORMED_ERROR_LABEL = "NoWritesPerformed";
 
     private static boolean decideRetryableAndAddRetryableWriteErrorLabel(final Throwable t, @Nullable final Integer maxWireVersion) {
         if (!(t instanceof MongoException)) {

driver-core/src/test/functional/com/mongodb/internal/connection/TestCommandListener.java

@@ -17,6 +17,7 @@
 package com.mongodb.internal.connection;
 
 import com.mongodb.MongoTimeoutException;
+import com.mongodb.annotations.ThreadSafe;
 import com.mongodb.client.TestListener;
 import com.mongodb.event.CommandEvent;
 import com.mongodb.event.CommandFailedEvent;
@@ -56,6 +57,7 @@
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 
+@ThreadSafe
 public class TestCommandListener implements CommandListener {
     private final List<String> eventTypes;
     private final List<String> ignoredCommandMonitoringEvents;

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

@@ -201,6 +201,7 @@ class InternalStreamConnectionInitializerSpecification extends Specification {
         def initializer = new InternalStreamConnectionInitializer(SINGLE, null, clientMetadataDocument, [], null)
         def expectedHelloCommandDocument = new BsonDocument(LEGACY_HELLO, new BsonInt32(1))
                 .append('helloOk', BsonBoolean.TRUE)
+                .append('backpressure', BsonBoolean.TRUE)
                 .append('\$db', new BsonString('admin'))
         if (clientMetadataDocument != null) {
              expectedHelloCommandDocument.append('client', clientMetadataDocument)
@@ -233,6 +234,7 @@ class InternalStreamConnectionInitializerSpecification extends Specification {
         def initializer = new InternalStreamConnectionInitializer(SINGLE, null, null, compressors, null)
         def expectedHelloCommandDocument = new BsonDocument(LEGACY_HELLO, new BsonInt32(1))
                 .append('helloOk', BsonBoolean.TRUE)
+                .append('backpressure', BsonBoolean.TRUE)
                 .append('\$db', new BsonString('admin'))
 
         def compressionArray = new BsonArray()
@@ -403,7 +405,8 @@ class InternalStreamConnectionInitializerSpecification extends Specification {
         ((SpeculativeAuthenticator) authenticator).getSpeculativeAuthenticateResponse() == null
         ((SpeculativeAuthenticator) authenticator)
                 .createSpeculativeAuthenticateCommand(internalConnection) == null
-        BsonDocument.parse("{$LEGACY_HELLO: 1, helloOk: true, '\$db': 'admin'}") == decodeCommand(internalConnection.getSent()[0])
+        BsonDocument.parse("{$LEGACY_HELLO: 1, helloOk: true, backpressure: true, '\$db': 'admin'}") ==
+                decodeCommand(internalConnection.getSent()[0])
 
         where:
         async << [true, false]
@@ -500,7 +503,7 @@ class InternalStreamConnectionInitializerSpecification extends Specification {
 
     def createHelloCommand(final String firstClientChallenge, final String mechanism,
                               final boolean hasSaslSupportedMechs) {
-        String hello = "{$LEGACY_HELLO: 1, helloOk: true, " +
+        String hello = "{$LEGACY_HELLO: 1, helloOk: true, backpressure: true, " +
                 (hasSaslSupportedMechs ? 'saslSupportedMechs: "database.user", ' : '') +
                 (mechanism == 'MONGODB-X509' ?
                         'speculativeAuthenticate: { authenticate: 1, ' +

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/ContextProviderTest.java

@@ -19,6 +19,7 @@
 import com.mongodb.ContextProvider;
 import com.mongodb.RequestContext;
 import com.mongodb.WriteConcern;
+import com.mongodb.annotations.NotThreadSafe;
 import com.mongodb.event.CommandFailedEvent;
 import com.mongodb.event.CommandListener;
 import com.mongodb.event.CommandStartedEvent;
@@ -186,6 +187,7 @@ public void contextShouldBeAvailableInCommandEvents() {
         }
     }
 
+    @NotThreadSafe
     private static final class TestCommandListener implements CommandListener {
         private int numCommandStartedEventsWithExpectedContext;
         private int numCommandSucceededEventsWithExpectedContext;

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/RetryableReadsProseTest.java

@@ -22,32 +22,27 @@
 import org.bson.Document;
 import org.junit.jupiter.api.Test;
 
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeoutException;
-
 import static com.mongodb.client.model.Filters.eq;
 
 /**
- * See
- * <a href="https://github.com/mongodb/specifications/tree/master/source/retryable-reads/tests">Retryable Reads Tests</a>.
+ * <a href="https://github.com/mongodb/specifications/blob/master/source/retryable-reads/tests/README.md#prose-tests">
+ * Prose Tests</a>.
  */
 final class RetryableReadsProseTest {
     /**
-     * See
-     * <a href="https://github.com/mongodb/specifications/tree/master/source/retryable-reads/tests#poolclearederror-retryability-test">
-     * PoolClearedError Retryability Test</a>.
+     * <a href="https://github.com/mongodb/specifications/blob/master/source/retryable-reads/tests/README.md#1-poolclearederror-retryability-test">
+     * 1. PoolClearedError Retryability Test</a>.
      */
     @Test
-    void poolClearedExceptionMustBeRetryable() throws InterruptedException, ExecutionException, TimeoutException {
+    void poolClearedExceptionMustBeRetryable() throws Exception {
         RetryableWritesProseTest.poolClearedExceptionMustBeRetryable(
                 SyncMongoClient::new,
                 mongoCollection -> mongoCollection.find(eq(0)).iterator().hasNext(), "find", false);
     }
 
     /**
-     * See
-     * <a href="https://github.com/mongodb/specifications/tree/master/source/retryable-reads/tests#21-retryable-reads-are-retried-on-a-different-mongos-when-one-is-available">
-     * Retryable Reads Are Retried on a Different mongos When One is Available</a>.
+     * <a href="https://github.com/mongodb/specifications/blob/master/source/retryable-reads/tests/README.md#21-retryable-reads-are-retried-on-a-different-mongos-when-one-is-available">
+     * 2.1 Retryable Reads Are Retried on a Different mongos When One is Available</a>.
      */
     @Test
     void retriesOnDifferentMongosWhenAvailable() {
@@ -61,9 +56,8 @@ void retriesOnDifferentMongosWhenAvailable() {
     }
 
     /**
-     * See
-     * <a href="https://github.com/mongodb/specifications/tree/master/source/retryable-reads/tests#22-retryable-reads-are-retried-on-the-same-mongos-when-no-others-are-available">
-     * Retryable Reads Are Retried on the Same mongos When No Others are Available</a>.
+     * <a href="https://github.com/mongodb/specifications/blob/master/source/retryable-reads/tests/README.md#22-retryable-reads-are-retried-on-the-same-mongos-when-no-others-are-available">
+     * 2.2 Retryable Reads Are Retried on the Same mongos When No Others are Available</a>.
      */
     @Test
     void retriesOnSameMongosWhenAnotherNotAvailable() {

driver-reactive-streams/src/test/functional/com/mongodb/reactivestreams/client/RetryableWritesProseTest.java

@@ -16,68 +16,87 @@
 
 package com.mongodb.reactivestreams.client;
 
-import com.mongodb.client.test.CollectionHelper;
 import com.mongodb.reactivestreams.client.syncadapter.SyncMongoClient;
 import org.bson.Document;
-import org.bson.codecs.DocumentCodec;
-import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 
-import java.util.concurrent.ExecutionException;
-import java.util.concurrent.TimeoutException;
-
 /**
- * See
- * <a href="https://github.com/mongodb/specifications/blob/master/source/retryable-writes/tests/README.md#prose-tests">Retryable Write Prose Tests</a>.
+ * <a href="https://github.com/mongodb/specifications/blob/master/source/retryable-writes/tests/README.md#prose-tests">
+ * Prose Tests</a>.
  */
-public class RetryableWritesProseTest extends DatabaseTestCase {
-    private CollectionHelper<Document> collectionHelper;
-
-    @BeforeEach
-    @Override
-    public void setUp() {
-        super.setUp();
-
-        collectionHelper = new CollectionHelper<>(new DocumentCodec(), collection.getNamespace());
-        collectionHelper.create();
-    }
-
+final class RetryableWritesProseTest {
     /**
-     * Prose test #2.
+     * <a href="https://github.com/mongodb/specifications/blob/master/source/retryable-writes/tests/README.md#2-test-that-drivers-properly-retry-after-encountering-poolclearederrors">
+     * 2. Test that drivers properly retry after encountering PoolClearedErrors</a>.
      */
     @Test
-    public void poolClearedExceptionMustBeRetryable() throws InterruptedException, ExecutionException, TimeoutException {
+    void poolClearedExceptionMustBeRetryable() throws Exception {
         com.mongodb.client.RetryableWritesProseTest.poolClearedExceptionMustBeRetryable(
                 SyncMongoClient::new,
                 mongoCollection -> mongoCollection.insertOne(new Document()), "insert", true);
     }
 
     /**
-     * Prose test #3.
+     * <a href="https://github.com/mongodb/specifications/blob/master/source/retryable-writes/tests/README.md#3-test-that-drivers-return-the-original-error-after-encountering-a-writeconcernerror-with-a-retryablewriteerror-label">
+     * 3. Test that drivers return the original error after encountering a WriteConcernError with a RetryableWriteError label</a>.
      */
     @Test
-    public void originalErrorMustBePropagatedIfNoWritesPerformed() throws InterruptedException {
+    void originalErrorMustBePropagatedIfNoWritesPerformed() throws Exception {
         com.mongodb.client.RetryableWritesProseTest.originalErrorMustBePropagatedIfNoWritesPerformed(
                 SyncMongoClient::new);
     }
 
     /**
-     * Prose test #4.
+     * <a href="https://github.com/mongodb/specifications/blob/master/source/retryable-writes/tests/README.md#4-test-that-in-a-sharded-cluster-writes-are-retried-on-a-different-mongos-when-one-is-available">
+     * 4. Test that in a sharded cluster writes are retried on a different mongos when one is available</a>.
      */
     @Test
-    public void retriesOnDifferentMongosWhenAvailable() {
+    void retriesOnDifferentMongosWhenAvailable() {
         com.mongodb.client.RetryableWritesProseTest.retriesOnDifferentMongosWhenAvailable(
                 SyncMongoClient::new,
                 mongoCollection -> mongoCollection.insertOne(new Document()), "insert", true);
     }
 
     /**
-     * Prose test #5.
+     * <a href="https://github.com/mongodb/specifications/blob/master/source/retryable-writes/tests/README.md#5-test-that-in-a-sharded-cluster-writes-are-retried-on-the-same-mongos-when-no-others-are-available">
+     * 5. Test that in a sharded cluster writes are retried on the same mongos when no others are available</a>.
      */
     @Test
-    public void retriesOnSameMongosWhenAnotherNotAvailable() {
+    void retriesOnSameMongosWhenAnotherNotAvailable() {
         com.mongodb.client.RetryableWritesProseTest.retriesOnSameMongosWhenAnotherNotAvailable(
                 SyncMongoClient::new,
                 mongoCollection -> mongoCollection.insertOne(new Document()), "insert", true);
     }
+
+    /**
+     * <a href="https://github.com/mongodb/specifications/blob/master/source/retryable-writes/tests/README.md#case-1-test-that-drivers-return-the-correct-error-when-receiving-only-errors-without-nowritesperformed">
+     * 6. Test error propagation after encountering multiple errors.
+     * Case 1: Test that drivers return the correct error when receiving only errors without NoWritesPerformed</a>.
+     */
+    @Test
+    @Disabled("TODO-BACKPRESSURE Valentin Enable when implementing JAVA-6055")
+    void errorPropagationAfterEncounteringMultipleErrorsCase1() throws Exception {
+        com.mongodb.client.RetryableWritesProseTest.errorPropagationAfterEncounteringMultipleErrorsCase1(SyncMongoClient::new);
+    }
+
+    /**
+     * <a href="https://github.com/mongodb/specifications/blob/master/source/retryable-writes/tests/README.md#case-2-test-that-drivers-return-the-correct-error-when-receiving-only-errors-with-nowritesperformed">
+     * 6. Test error propagation after encountering multiple errors.
+     * Case 2: Test that drivers return the correct error when receiving only errors with NoWritesPerformed</a>.
+     */
+    @Test
+    void errorPropagationAfterEncounteringMultipleErrorsCase2() throws Exception {
+        com.mongodb.client.RetryableWritesProseTest.errorPropagationAfterEncounteringMultipleErrorsCase2(SyncMongoClient::new);
+    }
+
+    /**
+     * <a href="https://github.com/mongodb/specifications/blob/master/source/retryable-writes/tests/README.md#case-3-test-that-drivers-return-the-correct-error-when-receiving-some-errors-with-nowritesperformed-and-some-without-nowritesperformed">
+     * 6. Test error propagation after encountering multiple errors.
+     * Case 3: Test that drivers return the correct error when receiving some errors with NoWritesPerformed and some without NoWritesPerformed</a>.
+     */
+    @Test
+    void errorPropagationAfterEncounteringMultipleErrorsCase3() throws Exception {
+        com.mongodb.client.RetryableWritesProseTest.errorPropagationAfterEncounteringMultipleErrorsCase3(SyncMongoClient::new);
+    }
 }

driver-sync/src/examples/documentation/TransactionExample.java

@@ -77,7 +77,8 @@ private void runTransactionWithRetry(final Runnable transactional) {
                 System.out.println("Transaction aborted. Caught exception during transaction.");
 
                 if (e.hasErrorLabel(MongoException.TRANSIENT_TRANSACTION_ERROR_LABEL)) {
-                    System.out.println("TransientTransactionError, aborting transaction and retrying ...");
+                    System.out.printf("%s, aborting transaction and retrying ...%n",
+                            MongoException.TRANSIENT_TRANSACTION_ERROR_LABEL);
                 } else {
                     throw e;
                 }
@@ -94,7 +95,8 @@ private void commitWithRetry(final ClientSession clientSession) {
             } catch (MongoException e) {
                 // can retry commit
                 if (e.hasErrorLabel(MongoException.UNKNOWN_TRANSACTION_COMMIT_RESULT_LABEL)) {
-                    System.out.println("UnknownTransactionCommitResult, retrying commit operation ...");
+                    System.out.printf("%s, retrying commit operation ...%n",
+                            MongoException.UNKNOWN_TRANSACTION_COMMIT_RESULT_LABEL);
                 } else {
                     System.out.println("Exception during commit ...");
                     throw e;

driver-sync/src/test/functional/com/mongodb/client/AbstractClientSideOperationsTimeoutProseTest.java

@@ -48,6 +48,7 @@
 import com.mongodb.event.ConnectionCreatedEvent;
 import com.mongodb.event.ConnectionReadyEvent;
 
+import static com.mongodb.MongoException.TRANSIENT_TRANSACTION_ERROR_LABEL;
 import static com.mongodb.internal.connection.CommandHelper.HELLO;
 import static com.mongodb.internal.connection.CommandHelper.LEGACY_HELLO;
 
@@ -687,7 +688,7 @@ public void test10CustomTestWithTransactionUsesASingleTimeoutWithLock() {
                 + "    blockConnection: true,"
                 + "    blockTimeMS: " + 25
                 + "    errorCode: " + 24
-                + "    errorLabels: [\"TransientTransactionError\"]"
+                + "    errorLabels: [\"" + TRANSIENT_TRANSACTION_ERROR_LABEL + "\"]"
                 + "  }"
                 + "}");
 

driver-sync/src/test/functional/com/mongodb/client/ContextProviderTest.java

@@ -19,6 +19,7 @@
 import com.mongodb.ContextProvider;
 import com.mongodb.RequestContext;
 import com.mongodb.WriteConcern;
+import com.mongodb.annotations.NotThreadSafe;
 import com.mongodb.event.CommandFailedEvent;
 import com.mongodb.event.CommandListener;
 import com.mongodb.event.CommandStartedEvent;
@@ -206,6 +207,7 @@ public void contextShouldBeAvailableInCommandEvents() {
         }
     }
 
+    @NotThreadSafe
     private static final class TestCommandListener implements CommandListener {
         private int numCommandStartedEventsWithExpectedContext;
         private int numCommandSucceededEventsWithExpectedContext;