[
https://issues.apache.org/jira/browse/CAMEL-12302?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=16385815#comment-16385815
]
ASF GitHub Bot commented on CAMEL-12302:
----------------------------------------
davsclaus closed pull request #2244: CAMEL-12302 : MongoDB - add bulk writes
operation
URL: https://github.com/apache/camel/pull/2244
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/components/camel-mongodb/src/main/docs/mongodb-component.adoc
b/components/camel-mongodb/src/main/docs/mongodb-component.adoc
index dbbff25a4ff..ce882822cd8 100644
--- a/components/camel-mongodb/src/main/docs/mongodb-component.adoc
+++ b/components/camel-mongodb/src/main/docs/mongodb-component.adoc
@@ -521,6 +521,43 @@ A header with key `CamelMongoDbRecordsAffected` is returned
containing the number of records deleted (copied from
`WriteResult.getN()`).
+
+==== Bulk Write Operations
+
+===== bulkWrite
+
+*Available as of Camel 2.21*
+
+Performs write operations in bulk with controls for order of execution.
+Requires a `List<WriteModel<DBObject>>` as the IN message body containing
commands for insert, update, and delete operations.
+
+The following example will insert a new scientist "Pierre Curie", update
record with id "5" by setting the value of the "scientist" field to
+"Marie Curie" and delete record with id "3" :
+
+[source,java]
+------------------------------------------------------------------------------------------------------------------
+// route:
from("direct:bulkWrite").to("mongodb:myDb?database=science&collection=notableScientists&operation=bulkWrite");
+List<WriteModel<DBObject>> bulkOperations = Arrays.asList(
+ new InsertOneModel<>(new BasicDBObject("scientist", "Pierre
Curie")),
+ new UpdateOneModel<>(new BasicDBObject("_id", "5"),
+ new BasicDBObject("$set", new
BasicDBObject("scientist", "Marie Curie"))),
+ new DeleteOneModel<>(new BasicDBObject("_id", "3")));
+
+BulkWriteResult result = template.requestBody("direct:bulkWrite",
bulkOperations, BulkWriteResult.class);
+------------------------------------------------------------------------------------------------------------------
+
+By default, operations are executed in order and interrupted on the first
write error without processing any remaining write operations in the list.
+To instruct MongoDB to continue to process remaining write operations in the
list, set the `CamelMongoDbBulkOrdered` IN message header to `false`.
+Unordered operations are executed in parallel and this behavior is not
guaranteed.
+
+[width="100%",cols="10%,10%,10%,70%",options="header",]
+|=======================================================================
+|Header key |Quick constant |Description (extracted from MongoDB API doc)
|Expected type
+
+|`CamelMongoDbBulkOrdered` |`MongoDbConstants.BULK_ORDERED` | Perform an
ordered or unordered operation execution. Defaults to true. |boolean/Boolean
+|=======================================================================
+
+
==== Other operations
===== aggregate
@@ -530,7 +567,6 @@ containing the number of records deleted (copied from
Perform a aggregation with the given pipeline contained in the
body. *Aggregations could be long and heavy operations. Use with care.*
-
[source,java]
----------------------------------------------------------------------------------------------------------------------------------------------------------------------
diff --git
a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbComponent.java
b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbComponent.java
index e2e8fee5171..dee4340c6c0 100644
---
a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbComponent.java
+++
b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbComponent.java
@@ -33,7 +33,8 @@
public static final Set<MongoDbOperation> WRITE_OPERATIONS =
new
HashSet<MongoDbOperation>(Arrays.asList(MongoDbOperation.insert,
MongoDbOperation.save,
- MongoDbOperation.update, MongoDbOperation.remove));
+ MongoDbOperation.update, MongoDbOperation.remove,
MongoDbOperation.bulkWrite));
+
private static final Logger LOG =
LoggerFactory.getLogger(MongoDbComponent.class);
public MongoDbComponent() {
diff --git
a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbConstants.java
b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbConstants.java
index 88cf6724ad3..90c599eb955 100644
---
a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbConstants.java
+++
b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbConstants.java
@@ -39,6 +39,7 @@
public static final String WRITERESULT = "CamelMongoWriteResult";
public static final String OID = "CamelMongoOid";
public static final String DISTINCT_QUERY_FIELD =
"CamelMongoDbDistinctQueryField";
+ public static final String BULK_ORDERED = "CamelMongoDbBulkOrdered";
private MongoDbConstants() {
}
diff --git
a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbOperation.java
b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbOperation.java
index 00228cd80e3..4af38ff2f77 100644
---
a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbOperation.java
+++
b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbOperation.java
@@ -34,6 +34,9 @@
// delete operations
remove,
+ //Bulk operations
+ bulkWrite,
+
// aggregate
aggregate,
diff --git
a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbProducer.java
b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbProducer.java
index f0192d938cf..157718a1865 100644
---
a/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbProducer.java
+++
b/components/camel-mongodb/src/main/java/org/apache/camel/component/mongodb/MongoDbProducer.java
@@ -27,12 +27,15 @@
import com.mongodb.BasicDBList;
import com.mongodb.BasicDBObject;
import com.mongodb.DBObject;
+import com.mongodb.bulk.BulkWriteResult;
import com.mongodb.client.AggregateIterable;
import com.mongodb.client.DistinctIterable;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
+import com.mongodb.client.model.BulkWriteOptions;
import com.mongodb.client.model.UpdateOptions;
+import com.mongodb.client.model.WriteModel;
import com.mongodb.client.result.DeleteResult;
import com.mongodb.client.result.UpdateResult;
@@ -69,6 +72,7 @@
bind(MongoDbOperation.remove, createDoRemove());
bind(MongoDbOperation.save, createDoSave());
bind(MongoDbOperation.update, createDoUpdate());
+ bind(MongoDbOperation.bulkWrite, createDoBulkWrite());
}
public MongoDbProducer(MongoDbEndpoint endpoint) {
@@ -585,4 +589,24 @@ private Object getMultiInsertBody(Exchange exchange) {
};
}
+ private Function<Exchange, Object> createDoBulkWrite() {
+ return exchange -> {
+ try {
+ MongoCollection<BasicDBObject> dbCol =
calculateCollection(exchange);
+
+ Boolean ordered =
exchange.getIn().getHeader(MongoDbConstants.BULK_ORDERED, Boolean.TRUE,
Boolean.class);
+ BulkWriteOptions options = new
BulkWriteOptions().ordered(ordered);
+
+ @SuppressWarnings("unchecked")
+ List<WriteModel<BasicDBObject>> requests =
exchange.getIn().getMandatoryBody((Class<List<WriteModel<BasicDBObject>>>)(Class<?>)List.class);
+
+ BulkWriteResult result = dbCol.bulkWrite(requests, options);
+ return result;
+
+ } catch (InvalidPayloadException e) {
+ throw new CamelMongoDbException("Invalid payload for bulk
write", e);
+ }
+ };
+ }
+
}
diff --git
a/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/MongoDbBulkWriteOperationTest.java
b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/MongoDbBulkWriteOperationTest.java
new file mode 100644
index 00000000000..6076d669a85
--- /dev/null
+++
b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/MongoDbBulkWriteOperationTest.java
@@ -0,0 +1,123 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mongodb;
+
+import java.util.Arrays;
+import java.util.List;
+
+import com.mongodb.BasicDBObject;
+import com.mongodb.DBObject;
+import com.mongodb.bulk.BulkWriteResult;
+import com.mongodb.client.model.DeleteManyModel;
+import com.mongodb.client.model.DeleteOneModel;
+import com.mongodb.client.model.InsertOneModel;
+import com.mongodb.client.model.ReplaceOneModel;
+import com.mongodb.client.model.UpdateManyModel;
+import com.mongodb.client.model.UpdateOneModel;
+import com.mongodb.client.model.WriteModel;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.Test;
+
+public class MongoDbBulkWriteOperationTest extends AbstractMongoDbTest {
+
+ @Test
+ public void testBulkWrite() throws Exception {
+ // Test that the collection has 0 documents in it
+ assertEquals(0, testCollection.count());
+ pumpDataIntoTestCollection();
+ List<WriteModel<DBObject>> bulkOperations = Arrays
+ .asList(new InsertOneModel<>(new BasicDBObject("scientist",
"Pierre Curie")),
+ new UpdateOneModel<>(new BasicDBObject("_id", "2"),
+ new BasicDBObject("$set", new
BasicDBObject("scientist", "Charles Darwin"))),
+ new UpdateManyModel<>(new BasicDBObject("scientist",
"Curie"),
+ new BasicDBObject("$set", new
BasicDBObject("scientist", "Marie Curie"))),
+ new ReplaceOneModel<>(new BasicDBObject("_id", "1"), new
BasicDBObject("scientist", "Albert Einstein")),
+ new DeleteOneModel<>(new BasicDBObject("_id", "3")),
+ new DeleteManyModel<>(new BasicDBObject("scientist",
"Bohr")));
+
+ BulkWriteResult result = template.requestBody("direct:bulkWrite",
bulkOperations, BulkWriteResult.class);
+
+ assertNotNull(result);
+ // 1 insert
+ assertEquals("Records inserted should be 2 : ", 1,
result.getInsertedCount());
+ // 1 updateOne + 100 updateMany + 1 replaceOne
+ assertEquals("Records matched should be 102 : ", 102,
result.getMatchedCount());
+ assertEquals("Records modified should be 102 : ", 102,
result.getModifiedCount());
+ // 1 deleteOne + 100 deleteMany
+ assertEquals("Records deleted should be 101 : ", 101,
result.getDeletedCount());
+ }
+
+ @Test
+ public void testOrderedBulkWriteWithError() throws Exception {
+ // Test that the collection has 0 documents in it
+ assertEquals(0, testCollection.count());
+ pumpDataIntoTestCollection();
+
+ List<WriteModel<DBObject>> bulkOperations = Arrays
+ .asList(new InsertOneModel<>(new BasicDBObject("scientist",
"Pierre Curie")),
+ // this insert failed and bulk stop
+ new InsertOneModel<>(new BasicDBObject("_id", "1")),
+ new InsertOneModel<>(new BasicDBObject("scientist",
"Descartes")),
+ new UpdateOneModel<>(new BasicDBObject("_id", "5"), new
BasicDBObject("$set", new BasicDBObject("scientist", "Marie Curie"))),
+ new DeleteOneModel<>(new BasicDBObject("_id", "2")));
+
+ try {
+ template.requestBody("direct:bulkWrite", bulkOperations,
BulkWriteResult.class);
+ fail("Bulk operation should throw Exception");
+ } catch (CamelExecutionException e) {
+ extractAndAssertCamelMongoDbException(e, "duplicate key error");
+ // count = 1000 records + 1 inserted
+ assertEquals(1001, testCollection.count());
+ }
+ }
+
+ @Test
+ public void testUnorderedBulkWriteWithError() throws Exception {
+ // Test that the collection has 0 documents in it
+ assertEquals(0, testCollection.count());
+ pumpDataIntoTestCollection();
+
+ List<WriteModel<DBObject>> bulkOperations = Arrays
+ .asList(new InsertOneModel<>(new BasicDBObject("scientist",
"Pierre Curie")),
+ // this insert failed and bulk continue
+ new InsertOneModel<>(new BasicDBObject("_id", "1")),
+ new InsertOneModel<>(new BasicDBObject("scientist",
"Descartes")),
+ new UpdateOneModel<>(new BasicDBObject("_id", "5"), new
BasicDBObject("$set", new BasicDBObject("scientist", "Marie Curie"))),
+ new DeleteOneModel<>(new BasicDBObject("_id", "2")));
+ try {
+ template.requestBody("direct:unorderedBulkWrite", bulkOperations,
BulkWriteResult.class);
+ fail("Bulk operation should throw Exception");
+ } catch (CamelExecutionException e) {
+ extractAndAssertCamelMongoDbException(e, "duplicate key error");
+ // count = 1000 + 2 inserted + 1 deleted
+ assertEquals(1001, testCollection.count());
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() {
+
from("direct:bulkWrite").to("mongodb:myDb?database={{mongodb.testDb}}&collection={{mongodb.testCollection}}&operation=bulkWrite");
+
from("direct:unorderedBulkWrite").setHeader(MongoDbConstants.BULK_ORDERED).constant(false)
+
.to("mongodb:myDb?database={{mongodb.testDb}}&collection={{mongodb.testCollection}}&operation=bulkWrite");
+ }
+ };
+ }
+}
diff --git a/components/camel-mongodb3/src/main/docs/mongodb3-component.adoc
b/components/camel-mongodb3/src/main/docs/mongodb3-component.adoc
index aef5bf03403..191dbdb933c 100644
--- a/components/camel-mongodb3/src/main/docs/mongodb3-component.adoc
+++ b/components/camel-mongodb3/src/main/docs/mongodb3-component.adoc
@@ -569,6 +569,42 @@ A header with key `CamelMongoDbRecordsAffected` is returned
containing the number of records deleted (copied from
`WriteResult.getN()`).
+==== Bulk Write Operations
+
+===== bulkWrite
+
+*Available as of Camel 2.21*
+
+Performs write operations in bulk with controls for order of execution.
+Requires a `List<WriteModel<Document>>` as the IN message body containing
commands for insert, update, and delete operations.
+
+The following example will insert a new scientist "Pierre Curie", update
record with id "5" by setting the value of the "scientist" field to
+"Marie Curie" and delete record with id "3" :
+
+[source,java]
+------------------------------------------------------------------------------------------------------------------
+// route:
from("direct:bulkWrite").to("mongodb:myDb?database=science&collection=notableScientists&operation=bulkWrite");
+List<WriteModel<Document>> bulkOperations = Arrays.asList(
+ new InsertOneModel<>(new Document("scientist", "Pierre Curie")),
+ new UpdateOneModel<>(new Document("_id", "5"),
+ new Document("$set", new
Document("scientist", "Marie Curie"))),
+ new DeleteOneModel<>(new Document("_id", "3")));
+
+BulkWriteResult result = template.requestBody("direct:bulkWrite",
bulkOperations, BulkWriteResult.class);
+------------------------------------------------------------------------------------------------------------------
+
+By default, operations are executed in order and interrupted on the first
write error without processing any remaining write operations in the list.
+To instruct MongoDB to continue to process remaining write operations in the
list, set the `CamelMongoDbBulkOrdered` IN message header to `false`.
+Unordered operations are executed in parallel and this behavior is not
guaranteed.
+
+[width="100%",cols="10%,10%,10%,70%",options="header",]
+|=======================================================================
+|Header key |Quick constant |Description (extracted from MongoDB API doc)
|Expected type
+
+|`CamelMongoDbBulkOrdered` |`MongoDbConstants.BULK_ORDERED` | Perform an
ordered or unordered operation execution. Defaults to true. |boolean/Boolean
+|=======================================================================
+
+
==== Other operations
===== aggregate
diff --git
a/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbConstants.java
b/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbConstants.java
index 7196a9897c3..30f204f51b4 100644
---
a/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbConstants.java
+++
b/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbConstants.java
@@ -40,6 +40,7 @@
public static final String WRITERESULT = "CamelMongoWriteResult";
public static final String OID = "CamelMongoOid";
public static final String DISTINCT_QUERY_FIELD =
"CamelMongoDbDistinctQueryField";
+ public static final String BULK_ORDERED = "CamelMongoDbBulkOrdered";
public static final String MONGO_ID = "_id"; // default id field
diff --git
a/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbOperation.java
b/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbOperation.java
index 6ddc7d3a9cf..6f1b127a9b6 100644
---
a/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbOperation.java
+++
b/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbOperation.java
@@ -33,7 +33,10 @@
// delete operations
remove,
-
+
+ //Bulk operations
+ bulkWrite,
+
// aggregate
aggregate,
diff --git
a/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbProducer.java
b/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbProducer.java
index 17cd024a8a4..9213849486c 100644
---
a/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbProducer.java
+++
b/components/camel-mongodb3/src/main/java/org/apache/camel/component/mongodb3/MongoDbProducer.java
@@ -24,13 +24,16 @@
import java.util.function.Function;
import java.util.stream.Collectors;
+import com.mongodb.bulk.BulkWriteResult;
import com.mongodb.client.AggregateIterable;
import com.mongodb.client.DistinctIterable;
import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoCollection;
import com.mongodb.client.MongoDatabase;
+import com.mongodb.client.model.BulkWriteOptions;
import com.mongodb.client.model.Filters;
import com.mongodb.client.model.UpdateOptions;
+import com.mongodb.client.model.WriteModel;
import com.mongodb.client.result.DeleteResult;
import com.mongodb.client.result.UpdateResult;
@@ -77,6 +80,7 @@
{
bind(MongoDbOperation.aggregate, createDoAggregate());
+ bind(MongoDbOperation.bulkWrite, createDoBulkWrite());
bind(MongoDbOperation.command, createDoCommand());
bind(MongoDbOperation.count, createDoCount());
bind(MongoDbOperation.findDistinct, createDoDistinct());
@@ -608,4 +612,24 @@ private void processAndTransferResult(Object result,
Exchange exchange, MongoDbO
}
};
}
+
+ private Function<Exchange, Object> createDoBulkWrite() {
+ return exchange -> {
+ try {
+ MongoCollection<Document> dbCol =
calculateCollection(exchange);
+
+ Boolean ordered =
exchange.getIn().getHeader(MongoDbConstants.BULK_ORDERED, Boolean.TRUE,
Boolean.class);
+ BulkWriteOptions options = new
BulkWriteOptions().ordered(ordered);
+
+ @SuppressWarnings("unchecked")
+ List<WriteModel<Document>> requests =
(List<WriteModel<Document>>)
exchange.getIn().getMandatoryBody((Class<List<WriteModel<Document>>>)Class.class.cast(List.class));
+
+ BulkWriteResult result = dbCol.bulkWrite(requests, options);
+ return result;
+
+ } catch (InvalidPayloadException e) {
+ throw new CamelMongoDbException("Invalid payload for bulk
write", e);
+ }
+ };
+ }
}
diff --git
a/components/camel-mongodb3/src/test/java/org/apache/camel/component/mongodb3/MongoDbBulkWriteOperationTest.java
b/components/camel-mongodb3/src/test/java/org/apache/camel/component/mongodb3/MongoDbBulkWriteOperationTest.java
new file mode 100644
index 00000000000..34b53bc5c1f
--- /dev/null
+++
b/components/camel-mongodb3/src/test/java/org/apache/camel/component/mongodb3/MongoDbBulkWriteOperationTest.java
@@ -0,0 +1,122 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.mongodb3;
+
+import java.util.Arrays;
+import java.util.List;
+
+import com.mongodb.bulk.BulkWriteResult;
+import com.mongodb.client.model.DeleteManyModel;
+import com.mongodb.client.model.DeleteOneModel;
+import com.mongodb.client.model.InsertOneModel;
+import com.mongodb.client.model.ReplaceOneModel;
+import com.mongodb.client.model.UpdateManyModel;
+import com.mongodb.client.model.UpdateOneModel;
+import com.mongodb.client.model.WriteModel;
+
+import org.apache.camel.CamelExecutionException;
+import org.apache.camel.builder.RouteBuilder;
+import org.bson.Document;
+import org.junit.Test;
+
+public class MongoDbBulkWriteOperationTest extends AbstractMongoDbTest {
+
+ @Test
+ public void testBulkWrite() throws Exception {
+ // Test that the collection has 0 documents in it
+ assertEquals(0, testCollection.count());
+ pumpDataIntoTestCollection();
+ List<WriteModel<Document>> bulkOperations = Arrays
+ .asList(new InsertOneModel<>(new Document("scientist", "Pierre
Curie")),
+ new UpdateOneModel<>(new Document("_id", "2"),
+ new Document("$set", new
Document("scientist", "Charles Darwin"))),
+ new UpdateManyModel<>(new Document("scientist", "Curie"),
+ new Document("$set", new Document("scientist",
"Marie Curie"))),
+ new ReplaceOneModel<>(new Document("_id", "1"), new
Document("scientist", "Albert Einstein")),
+ new DeleteOneModel<>(new Document("_id", "3")),
+ new DeleteManyModel<>(new Document("scientist", "Bohr")));
+
+ BulkWriteResult result = template.requestBody("direct:bulkWrite",
bulkOperations, BulkWriteResult.class);
+
+ assertNotNull(result);
+ // 1 insert
+ assertEquals("Records inserted should be 2 : ", 1,
result.getInsertedCount());
+ // 1 updateOne + 100 updateMany + 1 replaceOne
+ assertEquals("Records matched should be 102 : ", 102,
result.getMatchedCount());
+ assertEquals("Records modified should be 102 : ", 102,
result.getModifiedCount());
+ // 1 deleteOne + 100 deleteMany
+ assertEquals("Records deleted should be 101 : ", 101,
result.getDeletedCount());
+ }
+
+ @Test
+ public void testOrderedBulkWriteWithError() throws Exception {
+ // Test that the collection has 0 documents in it
+ assertEquals(0, testCollection.count());
+ pumpDataIntoTestCollection();
+
+ List<WriteModel<Document>> bulkOperations = Arrays
+ .asList(new InsertOneModel<>(new Document("scientist", "Pierre
Curie")),
+ // this insert failed and bulk stop
+ new InsertOneModel<>(new Document("_id", "1")),
+ new InsertOneModel<>(new Document("scientist",
"Descartes")),
+ new UpdateOneModel<>(new Document("_id", "5"), new
Document("$set", new Document("scientist", "Marie Curie"))),
+ new DeleteOneModel<>(new Document("_id", "2")));
+
+ try {
+ template.requestBody("direct:bulkWrite", bulkOperations,
BulkWriteResult.class);
+ fail("Bulk operation should throw Exception");
+ } catch (CamelExecutionException e) {
+ extractAndAssertCamelMongoDbException(e, "duplicate key error");
+ // count = 1000 records + 1 inserted
+ assertEquals(1001, testCollection.count());
+ }
+ }
+
+ @Test
+ public void testUnorderedBulkWriteWithError() throws Exception {
+ // Test that the collection has 0 documents in it
+ assertEquals(0, testCollection.count());
+ pumpDataIntoTestCollection();
+
+ List<WriteModel<Document>> bulkOperations = Arrays
+ .asList(new InsertOneModel<>(new Document("scientist", "Pierre
Curie")),
+ // this insert failed and bulk continue
+ new InsertOneModel<>(new Document("_id", "1")),
+ new InsertOneModel<>(new Document("scientist",
"Descartes")),
+ new UpdateOneModel<>(new Document("_id", "5"), new
Document("$set", new Document("scientist", "Marie Curie"))),
+ new DeleteOneModel<>(new Document("_id", "2")));
+ try {
+ template.requestBody("direct:unorderedBulkWrite", bulkOperations,
BulkWriteResult.class);
+ fail("Bulk operation should throw Exception");
+ } catch (CamelExecutionException e) {
+ extractAndAssertCamelMongoDbException(e, "duplicate key error");
+ // count = 1000 + 2 inserted + 1 deleted
+ assertEquals(1001, testCollection.count());
+ }
+ }
+
+ @Override
+ protected RouteBuilder createRouteBuilder() throws Exception {
+ return new RouteBuilder() {
+ public void configure() {
+
from("direct:bulkWrite").to("mongodb3:myDb?database={{mongodb.testDb}}&collection={{mongodb.testCollection}}&operation=bulkWrite");
+
from("direct:unorderedBulkWrite").setHeader(MongoDbConstants.BULK_ORDERED).constant(false)
+
.to("mongodb3:myDb?database={{mongodb.testDb}}&collection={{mongodb.testCollection}}&operation=bulkWrite");
+ }
+ };
+ }
+}
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
> camel-mongodb : Support for bulk writes operation
> -------------------------------------------------
>
> Key: CAMEL-12302
> URL: https://issues.apache.org/jira/browse/CAMEL-12302
> Project: Camel
> Issue Type: New Feature
> Components: camel-mongodb, camel-mongodb3
> Reporter: Farès Hassak
> Assignee: Andrea Cosentino
> Priority: Major
> Fix For: 2.21.0
>
>
> Hello,
> To improve write performance, we need support for bulk operations :
> [https://mongodb.github.io/mongo-java-driver/3.6/driver/tutorials/bulk-writes/]
> [https://docs.mongodb.com/manual/core/bulk-write-operations/]
> I will add new bulkWrite operation that expect a list of WriteModel Object in
> body.
> Example :
> {code:java}
> // route:
> from("direct:bulkWrite").to("mongodb:myDb?database=science&collection=notableScientists&operation=bulkWrite");
> List<WriteModel<DBObject>> bulkOperations = Arrays.asList(
> new InsertOneModel<>(new BasicDBObject("scientist",
> "Descartes")),
> new InsertOneModel<>(new BasicDBObject("scientist", "Pierre
> Curie")),
> new UpdateOneModel<>(new BasicDBObject("_id", "5"), new
> BasicDBObject("$set", new BasicDBObject("scientist", "Marie Curie"))),
> new DeleteOneModel<>(new BasicDBObject("_id", "2")));
> BulkWriteResult result = template.requestBody("direct:bulkWrite",
> bulkOperations, BulkWriteResult.class);
> {code}
>
> An header with key CamelMongoDbBulkOrdered will be added (true is the default
> value).
> {code:java}
> BulkWriteResult result = template.requestBodyAndHeader("direct:bulkWrite",
> bulkOperations, MongoDbConstants.BULK_ORDERED, false, BulkWriteResult.class)
> {code}
> Later, i will add support for String :
> {code:java}
> // route:
> from("direct:bulk").to("mongodb:myDb?database=science&collection=notableScientists&operation=bulkWrite");
> template.requestBody("direct:bulkWrite", "[
> { \"insertOne\" : { \"document\" : { \"scientist\" : \"Descartes\" }}},
> { \"insertOne\" : { \"document\" : { \"scientist\" : \"Pierre Curie\" }}},
> { \"updateOne\" : { \"filter\" : { \"_id\" : \"5\" }, \"update\" : {
> \"$set\" : { \"scientist\" : \"Marie Curie\" }}}},
> { \"deleteOne\" : { \"filter\" : { \"_id\" : 2} }} ]");
> {code}
> I will do 2 PR.
> Farès
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
