Fixed unordered bulk write operation to keep sending batches to the server even after an error in the current batch.

  JAVA-1541

Author: jyemin

src/main/com/mongodb/DBCollectionImpl.java

@@ -128,13 +128,15 @@ public List<Cursor> parallelScan(final ParallelScanOptions options) {
 
     @Override
     BulkWriteResult executeBulkWriteOperation(final boolean ordered, final List<WriteRequest> writeRequests,
-                                              final WriteConcern writeConcern, DBEncoder encoder) {
+                                              WriteConcern writeConcern, DBEncoder encoder) {
         isTrue("no operations", !writeRequests.isEmpty());
 
         if (writeConcern == null) {
             throw new IllegalArgumentException("Write concern can not be null");
         }
 
+        writeConcern = writeConcern.continueOnError(!ordered);
+
         if (encoder == null) {
             encoder = DefaultDBEncoder.FACTORY.create();
         }

src/test/com/mongodb/BulkWriteOperationSpecification.groovy

@@ -781,6 +781,7 @@ class BulkWriteOperationSpecification extends FunctionalSpecification {
         ordered << [true, false]
     }
 
+
     // just need to check one case here, since the others are checked above
     def 'should throw IllegalStateException when already executed with write concern'() {
         given:
@@ -826,6 +827,44 @@ class BulkWriteOperationSpecification extends FunctionalSpecification {
         ordered << [true, false]
     }
 
+    def 'should continue to execute batches after a failure if writes are unordered'() {
+        given:
+        collection.insert(new BasicDBObject('_id', 0))
+        collection.insert(new BasicDBObject('_id', 1000))
+
+        when:
+        def operation = initializeBulkOperation(false)
+        for (int i = 0; i < 2000; i++) {
+            operation.insert(new BasicDBObject('_id', i))
+        }
+        operation.execute(WriteConcern.ACKNOWLEDGED)
+
+        then:
+        def ex = thrown(BulkWriteException)
+        ex.writeErrors.size() == 2
+        ex.getWriteResult().getInsertedCount() == 1998
+        collection.count() == 2000
+    }
+
+    def 'should stop executing batches after a failure if writes are ordered'() {
+        given:
+        collection.insert(new BasicDBObject('_id', 500))
+        collection.insert(new BasicDBObject('_id', 1500))
+
+        when:
+        def operation = initializeBulkOperation(true)
+        for (int i = 0; i < 2000; i++) {
+            operation.insert(new BasicDBObject('_id', i))
+        }
+        operation.execute(WriteConcern.ACKNOWLEDGED)
+
+        then:
+        def ex = thrown(BulkWriteException)
+        ex.writeErrors.size() == 1
+        ex.getWriteResult().getInsertedCount() == 500
+        collection.count() == 502
+    }
+
     private static void addWritesToOperation(BulkWriteOperation operation) {
         operation.find(new BasicDBObject('_id', 1)).updateOne(new BasicDBObject('$set', new BasicDBObject('x', 2)))
         operation.find(new BasicDBObject('_id', 2)).updateOne(new BasicDBObject('$set', new BasicDBObject('x', 3)))