JAVA-1131: With bulk writes in emulation mode, treat GLE failures with either wnote or jnote as write concern errors

Author: jyemin

src/main/com/mongodb/DBCollectionImpl.java

@@ -845,10 +845,16 @@ private BulkWriteResult executeWriteProtocol() {
                         WriteResult writeResult = executeWriteProtocol(i);
                         if (writeConcern.callGetLastError()) {
                             bulkWriteBatchCombiner.addResult(getResult(writeResult), indexMap);
+                            // When a journaled write is requested but journaling is disabled, it's not thrown as an exception.
+                            if (isWriteConcernError(writeResult.getLastError()))  {
+                                bulkWriteBatchCombiner.addWriteConcernErrorResult(getWriteConcernError(writeResult.getLastError()));
+                            }
                         }
                     } catch (WriteConcernException writeException) {
-                        if (isWriteConcernError(writeException))  {
-                            bulkWriteBatchCombiner.addWriteConcernErrorResult(getWriteConcernError(writeException));
+                        if (isWriteConcernError(writeException.getCommandResult()))  {
+                            bulkWriteBatchCombiner.addResult(getResult(new WriteResult(writeException.getCommandResult(), writeConcern)),
+                                                             indexMap);
+                            bulkWriteBatchCombiner.addWriteConcernErrorResult(getWriteConcernError(writeException.getCommandResult()));
                         } else {
                             bulkWriteBatchCombiner.addWriteErrorResult(getBulkWriteError(writeException), indexMap);
                         }
@@ -865,10 +871,6 @@ private int getCount(final WriteResult writeResult) {
                 return getType() == INSERT ? 1 : writeResult.getN();
             }
 
-            private boolean isWriteConcernError(final WriteConcernException writeException) {
-                return writeException.getCommandResult().get("wtimeout") != null;
-            }
-
             List<BulkWriteUpsert> getUpsertedItems(final WriteResult writeResult) {
                 return writeResult.getUpsertedId() == null
                        ? Collections.<BulkWriteUpsert>emptyList()
@@ -882,9 +884,26 @@ private BulkWriteError getBulkWriteError(final WriteConcernException writeExcept
                                           0);
             }
 
-            private WriteConcernError getWriteConcernError(final WriteConcernException writeException) {
-                return new WriteConcernError(writeException.getCode(), writeException.getCommandResult().getString("err"),
-                                             getErrorResponseDetails(writeException.getCommandResult()));
+            // Accommodating GLE representation of write concern errors
+            private boolean isWriteConcernError(final CommandResult commandResult) {
+                return commandResult.get("wtimeout") != null || commandResult.get("wnote") != null || commandResult.get("jnote") != null;
+            }
+
+            private WriteConcernError getWriteConcernError(final CommandResult commandResult) {
+                return new WriteConcernError(commandResult.getCode(), getWriteConcernErrorMessage(commandResult),
+                                             getErrorResponseDetails(commandResult));
+            }
+
+            // GLE uses jnote and wnote as alternative ways or reporting write concern errors
+            private String getWriteConcernErrorMessage(final CommandResult commandResult) {
+                String errorMessage = commandResult.getString("jnote");
+                if (errorMessage == null) {
+                    errorMessage = commandResult.getString("wnote");
+                }
+                if (errorMessage == null) {
+                    errorMessage = commandResult.getString("err");
+                }
+                return errorMessage;
             }
 
             private DBObject getErrorResponseDetails(final DBObject response) {

src/test/com/mongodb/BulkWriteOperationSpecification.groovy

@@ -579,6 +579,65 @@ class BulkWriteOperationSpecification extends FunctionalSpecification {
         ordered << [false]
     }
 
+    def 'when w > 1 write concern is used on a standalone server with write commands, CommandFailureException is thrown'() {
+        assumeTrue(isStandalone() && serverIsAtLeastVersion(2.6))
+
+        given:
+        def operation = collection.initializeUnorderedBulkOperation()
+        operation.insert(new BasicDBObject('_id', 1))
+
+        when:
+        operation.execute(new WriteConcern(2, 1))
+
+        then:
+        thrown(CommandFailureException)
+
+        where:
+        ordered << [true, false]
+    }
+
+    def 'when w > 1 write concern is used on a standalone server without write commands, BulkWriteException is thrown'() {
+        assumeTrue(isStandalone() && !serverIsAtLeastVersion(2.6))
+
+        given:
+        def operation = collection.initializeUnorderedBulkOperation()
+        operation.insert(new BasicDBObject('_id', 1))
+        operation.insert(new BasicDBObject('_id', 2))
+
+        when:
+        operation.execute(new WriteConcern(2, 1))
+
+        then:
+        def e = thrown(BulkWriteException)
+        e.writeResult == new AcknowledgedBulkWriteResult(INSERT, 2, [])
+        e.writeConcernError != null
+        e.writeConcernError.getDetails().containsField('wnote')
+
+        where:
+        ordered << [true, false]
+    }
+
+    def 'when j write concern is used on a server without journaling or write commands, BulkWriteException is thrown'() {
+        assumeTrue(!isSharded() && isServerStartedWithJournalingDisabled() && !serverIsAtLeastVersion(2.6))
+
+        given:
+        def operation = collection.initializeUnorderedBulkOperation()
+        operation.insert(new BasicDBObject('_id', 1))
+        operation.insert(new BasicDBObject('_id', 2))
+
+        when:
+        operation.execute(WriteConcern.JOURNALED)
+
+        then:
+        def e = thrown(BulkWriteException)
+        e.writeResult == new AcknowledgedBulkWriteResult(INSERT, 2, [])
+        e.writeConcernError != null
+        e.writeConcernError.getDetails().containsField('jnote')
+
+        where:
+        ordered << [true, false]
+    }
+
     def 'execute should throw IllegalStateException when already executed'() {
         given:
         def operation = initializeBulkOperation(ordered)