Fix reporting _id for Bulk API edge case for pre-2.6 servers

When upserted _id is not returned on custom _id's

JAVA-1263

Author: rozza

src/main/com/mongodb/DBCollectionImpl.java

@@ -734,8 +734,9 @@ BulkWriteResult executeWriteCommandProtocol() {
                 @Override
                 WriteResult executeWriteProtocol(final int i) {
                     ModifyRequest update = updateRequests.get(i);
-                    return update(update.getQuery(), update.getUpdateDocument(), update.isUpsert(), update.isMulti(), writeConcern,
-                                  encoder);
+                    WriteResult writeResult = update(update.getQuery(), update.getUpdateDocument(), update.isUpsert(),
+                            update.isMulti(), writeConcern, encoder);
+                    return addMissingUpserted(update, writeResult);
                 }
 
                 @Override
@@ -759,8 +760,9 @@ BulkWriteResult executeWriteCommandProtocol() {
                 @Override
                 WriteResult executeWriteProtocol(final int i) {
                     ModifyRequest update = replaceRequests.get(i);
-                    return update(update.getQuery(), update.getUpdateDocument(), update.isUpsert(), update.isMulti(), writeConcern,
-                                  encoder);
+                    WriteResult writeResult = update(update.getQuery(), update.getUpdateDocument(), update.isUpsert(),
+                            update.isMulti(), writeConcern, encoder);
+                    return addMissingUpserted(update, writeResult);
                 }
 
                 @Override
@@ -921,6 +923,26 @@ private DBObject getErrorResponseDetails(final DBObject response) {
                 }
                 return details;
             }
+
+            WriteResult addMissingUpserted(final ModifyRequest update, final WriteResult writeResult) {
+                // On pre 2.6 servers upserts with custom _id's would be not be reported so we  check if _id
+                // was in the update query or the find query then massage the writeResult.
+                if (update.isUpsert() && writeConcern.callGetLastError() && !writeResult.isUpdateOfExisting()
+                        && writeResult.getUpsertedId() == null) {
+                    DBObject updateDocument = update.getUpdateDocument();
+                    DBObject query = update.getQuery();
+                    if (updateDocument.containsField("_id")) {
+                        CommandResult commandResult = writeResult.getLastError();
+                        commandResult.put("upserted", updateDocument.get("_id"));
+                        return new WriteResult(commandResult, writeResult.getLastConcern());
+                    } else  if (query.containsField("_id")) {
+                        CommandResult commandResult = writeResult.getLastError();
+                        commandResult.put("upserted", query.get("_id"));
+                        return new WriteResult(commandResult, writeResult.getLastConcern());
+                    }
+                }
+                return writeResult;
+            }
         }
     }
 }

src/test/com/mongodb/BulkWriteOperationSpecification.groovy

@@ -351,6 +351,47 @@ class BulkWriteOperationSpecification extends FunctionalSpecification {
         ordered << [true, false]
     }
 
+    def 'when a custom _id is upserted it should be in the write result'() {
+        given:
+        def operation = initializeBulkOperation(ordered)
+        operation.find(new BasicDBObject('_id', 0)).upsert()
+                .updateOne(new BasicDBObject('$set', new BasicDBObject('a', 0)))
+        operation.find(new BasicDBObject('a', 1)).upsert().replaceOne(new BasicDBObject('_id', 1))
+        operation.find(new BasicDBObject('_id', 2)).upsert().replaceOne(new BasicDBObject('_id', 2))
+
+        when:
+        def result = operation.execute()
+
+        then:
+        result == new AcknowledgedBulkWriteResult(UPDATE, 0, expectedModifiedCount(0), [new BulkWriteUpsert(0, 0),
+                                                                                        new BulkWriteUpsert(1, 1),
+                                                                                        new BulkWriteUpsert(2, 2)])
+        collection.count() == 3
+
+        where:
+        ordered << [true, false]
+    }
+
+    def 'unacknowledged upserts with custom _id should not error'() {
+        given:
+        def operation = initializeBulkOperation(ordered)
+        operation.find(new BasicDBObject('_id', 0)).upsert()
+                .updateOne(new BasicDBObject('$set', new BasicDBObject('a', 0)))
+        operation.find(new BasicDBObject('a', 1)).upsert().replaceOne(new BasicDBObject('_id', 1))
+        operation.find(new BasicDBObject('_id', 2)).upsert().replaceOne(new BasicDBObject('_id', 2))
+
+        when:
+        def result = operation.execute(WriteConcern.UNACKNOWLEDGED)
+        collection.insert(new BasicDBObject('_id', 4))
+
+        then:
+        result == new UnacknowledgedBulkWriteResult()
+        collection.count() == 4
+
+        where:
+        ordered << [true, false]
+    }
+
     def 'when a replacement document is 16MB, the document is still replaced'() {
         given:
         collection.insert(new BasicDBObject('_id', 1))
@@ -813,4 +854,4 @@ class BulkWriteOperationSpecification extends FunctionalSpecification {
     private static Integer expectedModifiedCount(final int expectedCountForServersThatSupportIt) {
         (serverIsAtLeastVersion(2.5)) ? expectedCountForServersThatSupportIt : null
     }
-}
+}