JAVA-2063: Obey ordered property in WriteProtocol when write concern is unacknowledged

Author: jyemin

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

@@ -91,7 +91,7 @@ public WriteConcernResult execute(final InternalConnection connection) {
                     sentCommandStartedEvent = true;
                 }
                 messageId = nextMessage.getId();
-                if (shouldAcknowledge(encodingMetadata)) {
+                if (shouldAcknowledge(encodingMetadata.getNextMessage())) {
                     CommandMessage getLastErrorMessage = new CommandMessage(new MongoNamespace(getNamespace().getDatabaseName(),
                                                                                                COMMAND_COLLECTION_NAME).getFullName(),
                                                                             createGetLastErrorCommandDocument(), false,
@@ -110,7 +110,7 @@ public WriteConcernResult execute(final InternalConnection connection) {
                 bsonOutput.close();
             }
 
-            if (shouldAcknowledge(encodingMetadata)) {
+            if (shouldAcknowledge(encodingMetadata.getNextMessage())) {
                 ResponseBuffers responseBuffers = null;
                 try {
                     responseBuffers = connection.receiveMessage(messageId);
@@ -128,7 +128,7 @@ public WriteConcernResult execute(final InternalConnection connection) {
                     }
                     if (writeConcern.isAcknowledged()) {
                         throw e;
-                    } else {
+                    } else if (ordered) {
                         break;
                     }
                 } catch (RuntimeException e) {
@@ -184,7 +184,7 @@ public void executeAsync(final InternalConnection connection, final SingleResult
             ByteBufferBsonOutput bsonOutput = new ByteBufferBsonOutput(connection);
             RequestMessage requestMessage = createRequestMessage(getMessageSettings(connection.getDescription()));
             RequestMessage nextMessage = encodeMessage(requestMessage, bsonOutput);
-            if (writeConcern.isAcknowledged()) {
+            if (shouldAcknowledge(nextMessage)) {
                 CommandMessage getLastErrorMessage = new CommandMessage(new MongoNamespace(getNamespace().getDatabaseName(),
                                                                                            COMMAND_COLLECTION_NAME).getFullName(),
                                                                         createGetLastErrorCommandDocument(), false,
@@ -233,8 +233,8 @@ protected BsonDocument getBaseCommandDocument() {
     protected abstract String getCommandName();
 
 
-    private boolean shouldAcknowledge(final RequestMessage.EncodingMetadata encodingMetadata) {
-        return writeConcern.isAcknowledged() || (isOrdered() && encodingMetadata.getNextMessage() != null);
+    private boolean shouldAcknowledge(final RequestMessage nextMessage) {
+        return writeConcern.isAcknowledged() || (isOrdered() && nextMessage != null);
     }
 
     private BsonDocument createGetLastErrorCommandDocument() {

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

@@ -18,6 +18,7 @@
 
 import com.mongodb.MongoNamespace;
 import com.mongodb.WriteConcern;
+import com.mongodb.WriteConcernException;
 import com.mongodb.WriteConcernResult;
 import com.mongodb.async.SingleResultCallback;
 import org.bson.BsonDocument;
@@ -52,9 +53,19 @@ protected void callCallback(final BsonDocument result, final Throwable t) {
             callback.onResult(null, t);
         } else {
             try {
-                WriteConcernResult writeConcernResult = ProtocolHelper.getWriteResult(result,
-                                                                                      connection.getDescription().getServerAddress());
-                if (nextMessage != null) {
+                WriteConcernResult writeConcernResult = null;
+                boolean shouldWriteNextMessage = true;
+                try {
+                    writeConcernResult = ProtocolHelper.getWriteResult(result, connection.getDescription().getServerAddress());
+                } catch (WriteConcernException e) {
+                    if (writeConcern.isAcknowledged()) {
+                        throw e;
+                    }
+                    if (ordered) {
+                        shouldWriteNextMessage = false;
+                    }
+                }
+                if (shouldWriteNextMessage && nextMessage != null) {
                     new GenericWriteProtocol(namespace, nextMessage, ordered, writeConcern).executeAsync(connection, callback);
                 } else {
                     callback.onResult(writeConcernResult, null);

driver-core/src/test/functional/com/mongodb/connection/WriteProtocolSpecification.groovy

@@ -28,11 +28,13 @@ import org.bson.BsonDocument
 import org.bson.BsonInt32
 import org.bson.BsonString
 import org.bson.codecs.BsonDocumentCodec
+import spock.lang.IgnoreIf
 import spock.lang.Shared
 
 import static com.mongodb.ClusterFixture.getCredentialList
 import static com.mongodb.ClusterFixture.getPrimary
 import static com.mongodb.ClusterFixture.getSslSettings
+import static com.mongodb.ClusterFixture.isSharded
 import static com.mongodb.WriteConcern.ACKNOWLEDGED
 import static com.mongodb.WriteConcern.UNACKNOWLEDGED
 import static com.mongodb.connection.ProtocolTestHelper.execute
@@ -104,6 +106,7 @@ class WriteProtocolSpecification extends OperationFunctionalSpecification {
         async << [false, true]
     }
 
+    @IgnoreIf({ isSharded() })
     def 'should execute split unacknowledged inserts'() {
         given:
         def binary = new BsonBinary(new byte[15000000])
@@ -150,7 +153,9 @@ class WriteProtocolSpecification extends OperationFunctionalSpecification {
         where:
         async << [false, true]
     }
-    def 'should ignore write errors on split unacknowledged inserts'() {
+
+    @IgnoreIf({ isSharded() })
+    def 'should stop writing on write error when an ordered unacknowledged inserts must be split'() {
         given:
         def binary = new BsonBinary(new byte[15000000])
         def documentOne = new BsonDocument('_id', new BsonInt32(1)).append('b', binary)
@@ -166,14 +171,40 @@ class WriteProtocolSpecification extends OperationFunctionalSpecification {
 
         when:
         execute(protocol, connection, async)
+        // force acknowledgement
+        new CommandProtocol(getDatabaseName(), new BsonDocument('ping', new BsonInt32(1)),
+                new NoOpFieldNameValidator(), new BsonDocumentCodec()).execute(connection)
 
         then:
         getCollectionHelper().count() == 1
 
-        cleanup:
+        where:
+        async << [false, true]
+    }
+
+    @IgnoreIf({ isSharded() })
+    def 'should continue writing on write error when an unordered unacknowledged inserts must be split'() {
+        given:
+        def binary = new BsonBinary(new byte[15000000])
+        def documentOne = new BsonDocument('_id', new BsonInt32(1)).append('b', binary)
+        def documentTwo = new BsonDocument('_id', new BsonInt32(2)).append('b', binary)
+        def documentThree = new BsonDocument('_id', new BsonInt32(3)).append('b', binary)
+        def documentFour = new BsonDocument('_id', new BsonInt32(4)).append('b', binary)
+
+        def insertRequest = [new InsertRequest(documentOne), new InsertRequest(documentTwo),
+                             new InsertRequest(documentThree), new InsertRequest(documentFour)]
+        def protocol = new InsertProtocol(getNamespace(), false, UNACKNOWLEDGED, insertRequest)
+
+        getCollectionHelper().insertDocuments(documentOne)
+
+        when:
+        execute(protocol, connection, async)
         // force acknowledgement
-        new CommandProtocol(getDatabaseName(), new BsonDocument('drop', new BsonString(getCollectionName())),
-                            new NoOpFieldNameValidator(), new BsonDocumentCodec()).execute(connection)
+        new CommandProtocol(getDatabaseName(), new BsonDocument('ping', new BsonInt32(1)),
+                new NoOpFieldNameValidator(), new BsonDocumentCodec()).execute(connection)
+
+        then:
+        getCollectionHelper().count() == 4
 
         where:
         async << [false, true]
@@ -203,4 +234,29 @@ class WriteProtocolSpecification extends OperationFunctionalSpecification {
         where:
         async << [false, true]
     }
+
+    @IgnoreIf({ isSharded() })
+    def 'should not report write errors on split unacknowledged inserts'() {
+        given:
+        def binary = new BsonBinary(new byte[15000000])
+        def documentOne = new BsonDocument('_id', new BsonInt32(1)).append('b', binary)
+        def documentTwo = new BsonDocument('_id', new BsonInt32(2)).append('b', binary)
+        def documentThree = new BsonDocument('_id', new BsonInt32(3)).append('b', binary)
+        def documentFour = new BsonDocument('_id', new BsonInt32(4)).append('b', binary)
+
+        def insertRequest = [new InsertRequest(documentOne), new InsertRequest(documentTwo),
+                             new InsertRequest(documentThree), new InsertRequest(documentFour)]
+        def protocol = new InsertProtocol(getNamespace(), true, UNACKNOWLEDGED, insertRequest)
+
+        getCollectionHelper().insertDocuments(documentOne)
+
+        when:
+        execute(protocol, connection, async)
+
+        then:
+        getCollectionHelper().count() == 1
+
+        where:
+        async << [false, true]
+    }
 }