JAVA-2788: Prohibit unacknowledged writes within an explicit session

Author: jyemin

driver-async/src/test/functional/com/mongodb/async/client/MongoClientSessionSpecification.groovy

@@ -38,8 +38,10 @@ import spock.lang.IgnoreIf
 
 import java.util.concurrent.TimeUnit
 
+import static com.mongodb.ClusterFixture.isDiscoverableReplicaSet
 import static com.mongodb.ClusterFixture.isStandalone
 import static com.mongodb.ClusterFixture.serverVersionAtLeast
+import static com.mongodb.async.client.Fixture.getMongoClient
 import static com.mongodb.async.client.TestHelper.run
 
 class MongoClientSessionSpecification extends FunctionalSpecification {
@@ -283,10 +285,45 @@ class MongoClientSessionSpecification extends FunctionalSpecification {
         then:
         def pingCommandStartedEvent = commandListener.events.get(0)
         !(pingCommandStartedEvent as CommandStartedEvent).command.containsKey('lsid')
+
         cleanup:
         client?.close()
     }
 
+    @IgnoreIf({ !serverVersionAtLeast(3, 6) || isStandalone() })
+    def 'should throw exception if unacknowledged write used with explicit session'() {
+        given:
+        def session = run(getMongoClient().&startSession)
+
+        when:
+        run(getMongoClient().getDatabase(getDatabaseName()).getCollection(getCollectionName())
+                .withWriteConcern(WriteConcern.UNACKNOWLEDGED).&insertOne, session, new Document())
+
+        then:
+        thrown(MongoClientException)
+
+        cleanup:
+        session?.close()
+    }
+
+
+    @IgnoreIf({ !serverVersionAtLeast(3, 7) || !isDiscoverableReplicaSet() })
+    def 'should ignore unacknowledged write concern when in a transaction'() {
+        given:
+        def session = run(getMongoClient().&startSession)
+        session.startTransaction()
+
+        when:
+        run(getMongoClient().getDatabase(getDatabaseName()).getCollection(getCollectionName())
+                .withWriteConcern(WriteConcern.UNACKNOWLEDGED).&insertOne, session, new Document())
+
+        then:
+        noExceptionThrown()
+
+        cleanup:
+        session.close()
+    }
+
     // This test attempts attempts to demonstrate that causal consistency works correctly by inserting a document and then immediately
     // searching for that document on a secondary by its _id and failing the test if the document is not found.  Without causal consistency
     // enabled the expectation is that eventually that test would fail since generally the find will execute on the secondary before

driver-async/src/test/unit/com/mongodb/async/client/ClientSessionBindingSpecification.groovy

@@ -163,7 +163,7 @@ class ClientSessionBindingSpecification extends Specification {
         where:
         ownsSession << [true, false]
     }
-    
+
     private AsyncReadWriteBinding createStubBinding() {
         def cluster = Mock(Cluster) {
             selectServerAsync(_, _) >> {

driver-core/src/main/com/mongodb/operation/BulkWriteBatch.java

@@ -17,6 +17,7 @@
 package com.mongodb.operation;
 
 import com.mongodb.MongoBulkWriteException;
+import com.mongodb.MongoClientException;
 import com.mongodb.MongoInternalException;
 import com.mongodb.MongoNamespace;
 import com.mongodb.WriteConcern;
@@ -97,6 +98,10 @@ public static BulkWriteBatch createBulkWriteBatch(final MongoNamespace namespace
                                                       final Boolean bypassDocumentValidation, final boolean retryWrites,
                                                       final List<? extends WriteRequest> writeRequests,
                                                       final SessionContext sessionContext) {
+        if (sessionContext.hasSession() && !sessionContext.isImplicitSession() && !sessionContext.hasActiveTransaction()
+                && !writeConcern.isAcknowledged()) {
+            throw new MongoClientException("Unacknowledged writes are not supported when using an explicit session");
+        }
         boolean canRetryWrites = isRetryableWrite(retryWrites, writeConcern, serverDescription, connectionDescription, sessionContext);
         List<WriteRequestWithIndex> writeRequestsWithIndex = new ArrayList<WriteRequestWithIndex>();
         boolean writeRequestsAreRetryable = true;

driver-legacy/src/test/functional/com/mongodb/MongoClientSessionSpecification.groovy

@@ -33,10 +33,11 @@ import java.util.concurrent.TimeUnit
 import static Fixture.getDefaultDatabaseName
 import static Fixture.getMongoClientURI
 import static com.mongodb.ClusterFixture.isAuthenticated
+import static com.mongodb.ClusterFixture.isDiscoverableReplicaSet
 import static com.mongodb.ClusterFixture.isStandalone
 import static com.mongodb.ClusterFixture.serverVersionAtLeast
-import static com.mongodb.Fixture.getMongoClient
 import static com.mongodb.MongoCredential.createCredential
+import static com.mongodb.Fixture.getMongoClient
 
 class MongoClientSessionSpecification extends FunctionalSpecification {
 
@@ -404,4 +405,39 @@ class MongoClientSessionSpecification extends FunctionalSpecification {
         cleanup:
         client?.close()
     }
+
+    @IgnoreIf({ !serverVersionAtLeast(3, 6) || isStandalone() })
+    def 'should throw exception if unacknowledged write used with explicit session'() {
+        given:
+        def session = getMongoClient().startSession()
+
+        when:
+        getMongoClient().getDatabase(getDatabaseName()).getCollection(getCollectionName())
+                .withWriteConcern(WriteConcern.UNACKNOWLEDGED)
+                .insertOne(session, new Document())
+
+        then:
+        thrown(MongoClientException)
+
+        cleanup:
+        session?.close()
+    }
+
+    @IgnoreIf({ !serverVersionAtLeast(3, 7) || !isDiscoverableReplicaSet() })
+    def 'should ignore unacknowledged write concern when in a transaction'() {
+        given:
+        def session = getMongoClient().startSession()
+        session.startTransaction()
+
+        when:
+        getMongoClient().getDatabase(getDatabaseName()).getCollection(getCollectionName())
+                .withWriteConcern(WriteConcern.UNACKNOWLEDGED)
+                .insertOne(session, new Document())
+
+        then:
+        noExceptionThrown()
+
+        cleanup:
+        session.close()
+    }
 }

driver-sync/src/test/functional/com/mongodb/client/MongoClientSessionSpecification.groovy

@@ -37,6 +37,7 @@ import spock.lang.IgnoreIf
 
 import java.util.concurrent.TimeUnit
 
+import static com.mongodb.ClusterFixture.isDiscoverableReplicaSet
 import static com.mongodb.ClusterFixture.isStandalone
 import static com.mongodb.ClusterFixture.serverVersionAtLeast
 import static com.mongodb.client.Fixture.getDefaultDatabaseName
@@ -321,7 +322,7 @@ class MongoClientSessionSpecification extends FunctionalSpecification {
     }
 
     @IgnoreIf({ !serverVersionAtLeast(3, 6) })
-    def 'should not use a session for an unacknowledged write'() {
+    def 'should not use an implicit session for an unacknowledged write'() {
         given:
         def commandListener = new TestCommandListener()
         def settings = MongoClientSettings.builder(getMongoClientSettings()).commandListenerList([commandListener]).build()
@@ -339,4 +340,40 @@ class MongoClientSessionSpecification extends FunctionalSpecification {
         cleanup:
         client?.close()
     }
+
+    @IgnoreIf({ !serverVersionAtLeast(3, 6) || isStandalone() })
+    def 'should throw exception if unacknowledged write used with explicit session'() {
+        given:
+        def session = getMongoClient().startSession()
+
+        when:
+        getMongoClient().getDatabase(getDatabaseName()).getCollection(getCollectionName())
+                .withWriteConcern(WriteConcern.UNACKNOWLEDGED)
+                .insertOne(session, new Document())
+
+        then:
+        thrown(MongoClientException)
+
+        cleanup:
+        session?.close()
+    }
+
+
+    @IgnoreIf({ !serverVersionAtLeast(3, 7) || !isDiscoverableReplicaSet() })
+    def 'should ignore unacknowledged write concern when in a transaction'() {
+        given:
+        def session = getMongoClient().startSession()
+        session.startTransaction()
+
+        when:
+        getMongoClient().getDatabase(getDatabaseName()).getCollection(getCollectionName())
+                .withWriteConcern(WriteConcern.UNACKNOWLEDGED)
+                .insertOne(session, new Document())
+
+        then:
+        noExceptionThrown()
+
+        cleanup:
+        session.close()
+    }
 }