[camel] branch main updated: Ability to pass sharded key to upsert query in mongo component (#7666)

[davsclaus]

This is an automated email from the ASF dual-hosted git repository.

davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git


The following commit(s) were added to refs/heads/main by this push:
     new 8acc8d73838 Ability to pass sharded key to upsert query in mongo 
component (#7666)
8acc8d73838 is described below

commit 8acc8d738386959b0dd8888efba49752aa80897d
Author: Kanstantsin I <vault.sch...@gmail.com>
AuthorDate: Tue May 24 10:51:21 2022 +0300

    Ability to pass sharded key to upsert query in mongo component (#7666)
    
    Co-authored-by: Kanstantsin Ihnatsenka <kignate...@tula.co>
---
 .../camel/component/mongodb/MongoDbProducer.java   |  6 ++++-
 .../mongodb/integration/MongoDbOperationsIT.java   | 27 ++++++++++++++++++++++
 2 files changed, 32 insertions(+), 1 deletion(-)

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 f95805d81e5..e491d086dac 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
@@ -631,7 +631,11 @@ public class MongoDbProducer extends DefaultProducer {
                     result = dbCol.replaceOne(Filters.where("false"), saveObj, 
options);
                     exchange.getIn().setHeader(OID, 
result.getUpsertedId().asObjectId().getValue());
                 } else {
-                    result = dbCol.replaceOne(eq(MONGO_ID, 
saveObj.get(MONGO_ID)), saveObj, options);
+                    Bson mongoIdQuery = eq(MONGO_ID, saveObj.get(MONGO_ID));
+                    //You can pass sharded key query via CRITERIA header to 
allow update sharded collection
+                    Bson query = exchange.getIn().getHeader(CRITERIA, 
Bson.class);
+                    query = query != null ? Filters.and(query, mongoIdQuery) : 
mongoIdQuery;
+                    result = dbCol.replaceOne(query, saveObj, options);
                     exchange.getIn().setHeader(OID, saveObj.get(MONGO_ID));
                 }
                 return result;
diff --git 
a/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/integration/MongoDbOperationsIT.java
 
b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/integration/MongoDbOperationsIT.java
index d5d77825cd2..5c0583bc853 100644
--- 
a/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/integration/MongoDbOperationsIT.java
+++ 
b/components/camel-mongodb/src/test/java/org/apache/camel/component/mongodb/integration/MongoDbOperationsIT.java
@@ -43,6 +43,7 @@ import static com.mongodb.client.model.Filters.or;
 import static com.mongodb.client.model.Updates.combine;
 import static com.mongodb.client.model.Updates.currentTimestamp;
 import static com.mongodb.client.model.Updates.set;
+import static org.apache.camel.component.mongodb.MongoDbConstants.CRITERIA;
 import static org.apache.camel.component.mongodb.MongoDbConstants.MONGO_ID;
 import static org.apache.camel.test.junit5.TestSupport.assertListSize;
 import static org.junit.jupiter.api.Assertions.assertEquals;
@@ -128,6 +129,32 @@ public class MongoDbOperationsIT extends 
AbstractMongoDbITSupport {
 
     }
 
+    @Test
+    public void testSaveWithShardedKey() {
+        // Prepare test
+        assertEquals(0, testCollection.countDocuments());
+        Object[] req = new Object[] {
+                new Document(MONGO_ID, "testSave1").append("scientist", 
"Einstein").append("country", "Germany").toJson(),
+                new Document(MONGO_ID, "testSave2").append("scientist", 
"Copernicus").append("country", "Poland").toJson() };
+        Object result = template.requestBody("direct:insert", req);
+        assertTrue(result instanceof List);
+        assertEquals(2, testCollection.countDocuments(), "Number of records 
persisted must be 2");
+
+        // Testing the save logic
+        Document record1 = testCollection.find(eq(MONGO_ID, 
"testSave1")).first();
+        assertEquals("Einstein", record1.get("scientist"), "Scientist field of 
'testSave1' must equal 'Einstein'");
+        record1.put("scientist", "Kepler");
+
+        //Pass sharded collection key as CRITERIA to prevent 
"MongoWriteException: Failed to target upsert by query :: could not extract 
exact shard key"
+        result = template.requestBodyAndHeader("direct:save", record1, 
CRITERIA, eq("country","Germany"));
+        assertTrue(result instanceof UpdateResult);
+
+        record1 = testCollection.find(eq(MONGO_ID, "testSave1")).first();
+        assertEquals("Kepler", record1.get("scientist"),
+                "Scientist field of 'testSave1' must equal 'Kepler' after save 
operation");
+
+    }
+
     @Test
     public void testSaveWithoutId() {
         // Prepare test