Repository: microsoft/kafka-connect-cosmosdb (java)

Issue:
Sink only has option for CreateItem, Upsert should be considered
**## Problem Statement**
Working on a real scenarios, sometimes we need to option to have upinsert (including partial when available at Cosmos) , or we're going to have errors

**## Proposed Solution**
Because UpsertItem has an overhead associated with it, in terms of RU, and performance (b/c the backend has to first check if the doc exists, and if so, update it, else create it) if we offer Upsert it should be in addition, not instead of, CreateItem. 

We will expose a new Sink connector configuration option
_**cosmosdb.sink.useUpsert**_ boolean, required: no, default:false

If useUpsert is enabled then the Sink task will perform an Upsert instead of Create
If userUpsert is false (default) then the Sink task will execute a Create

Errors to be handled - 
- Create when item already exists
- Upsert / Create when no id present in document


## Next Steps
- [x] Team consensus to proceed 
- [x] Complete Design
- [x] Design Reviewed


Pull request:
Exposed upsert configuration for Sink Connector and added user-agent to cosmosdb java client


## Type of PR

- [ ] Documentation changes
- [x] Code changes
- [x] Test changes
- [ ] CI-CD changes
- [ ] GitHub Template changes

## Purpose of PR
- Exposed use upsert config to Sink Connector Property to support both Insert and Upsert Operations
- Set default value to false for upsert config
- Set User-Agent for CosmosDB Java client

## Review notes
![image](https://user-images.githubusercontent.com/71272427/102287593-63b09980-3f00-11eb-93f1-4b1a80625d9d.png)

## Issues Closed or Referenced

- Closes #147 #167 


Patch:
diff --git a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/SettingDefaults.java b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/SettingDefaults.java
index 0bd2587..d5a6e8b 100644
--- a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/SettingDefaults.java
+++ b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/SettingDefaults.java
@@ -8,4 +8,5 @@ public class SettingDefaults {
     public static final Long TASK_BUFFER_SIZE=10000L;
     public static final Long TASK_BATCH_SIZE = 100L;
     public static final Long TASK_POLL_INTERVAL = 1000L;
+    public static final String COSMOS_CLIENT_USER_AGENT_SUFFIX = "APN/1.0 Microsoft/1.0 KafkaConnect/";
 }
diff --git a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/CosmosDBSinkTask.java b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/CosmosDBSinkTask.java
index fec33f2..5ff93e9 100644
--- a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/CosmosDBSinkTask.java
+++ b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/CosmosDBSinkTask.java
@@ -1,5 +1,6 @@
 package com.microsoft.azure.cosmosdb.kafka.connect.sink;
 
+import com.microsoft.azure.cosmosdb.kafka.connect.SettingDefaults;
 import com.azure.cosmos.CosmosClient;
 import com.azure.cosmos.CosmosClientBuilder;
 import com.azure.cosmos.CosmosContainer;
@@ -37,6 +38,7 @@ public class CosmosDBSinkTask extends SinkTask {
         this.client = new CosmosClientBuilder()
                 .endpoint(settings.getEndpoint())
                 .key(settings.getKey())
+                .userAgentSuffix(SettingDefaults.COSMOS_CLIENT_USER_AGENT_SUFFIX+version())
                 .buildClient();
 
         client.createDatabaseIfNotExists(settings.getDatabaseName());
@@ -68,7 +70,12 @@ public class CosmosDBSinkTask extends SinkTask {
                     logger.debug("Value.toString(): " + (record.value() != null ? record.value().toString() : "<null>"));
                 }
                 try {
-                    container.createItem(record.value());
+                    if(!this.settings.getUseUpsert()) {
+                        container.createItem(record.value());
+                    }
+                    else {
+                        container.upsertItem(record.value());
+                    }
                 } catch (BadRequestException bre) {
                     throw new CosmosDBWriteException(record, bre);
                 }
diff --git a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/SinkSettingDefaults.java b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/SinkSettingDefaults.java
new file mode 100644
index 0000000..57110d9
--- /dev/null
+++ b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/SinkSettingDefaults.java
@@ -0,0 +1,5 @@
+package com.microsoft.azure.cosmosdb.kafka.connect.sink;
+
+public class SinkSettingDefaults {
+    public static final Boolean USE_UPSERT = false;
+}
diff --git a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/SinkSettings.java b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/SinkSettings.java
index 5dace75..1e177a3 100644
--- a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/SinkSettings.java
+++ b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/SinkSettings.java
@@ -1,6 +1,7 @@
 package com.microsoft.azure.cosmosdb.kafka.connect.sink;
 
 import com.microsoft.azure.cosmosdb.kafka.connect.Setting;
+import com.microsoft.azure.cosmosdb.kafka.connect.BooleanSetting;
 import com.microsoft.azure.cosmosdb.kafka.connect.Settings;
 import org.apache.commons.collections4.ListUtils;
 import org.apache.kafka.connect.sink.SinkTask;
@@ -13,6 +14,7 @@ import java.util.List;
  */
 public class SinkSettings extends Settings {
     private String postProcessor;
+    private Boolean useUpsert;
     private final List<Setting> sinkSettings = Arrays.asList(
             //Add all sink settings here:
             new Setting(SinkTask.TOPICS_CONFIG, "List of topics to consume, separated by commas.", "Topics", s -> {
@@ -22,11 +24,32 @@ public class SinkSettings extends Settings {
                             "Under the hood, the regex is compiled to a <code>java.util.regex.Pattern</code>. " +
                             "Only one of " + SinkTask.TOPICS_CONFIG + " or " + SinkTask.TOPICS_REGEX_CONFIG + " should be specified.",
                     "Topics RegEx", s -> {
-            }, () -> "")
+            }, () -> ""),
+            new BooleanSetting(Settings.PREFIX + ".sink.useUpsert", "Behaviour of operation to Cosmos DB. " +
+                    "'false' is the default value and signals that all operations to Cosmos DB are Insert; 'true' changes the behaviour to use Upsert operation.",
+                    "Use Upsert", SinkSettingDefaults.USE_UPSERT, this::setUseUpsert, this::getUseUpsert)
     );
 
     @Override
     protected List<Setting> getAllSettings() {
         return ListUtils.union(super.getAllSettings(), sinkSettings);
     }
+
+    /**
+     * Returns boolean value to use upsert operation
+     *
+     * @return useUpsert
+     */
+    public Boolean getUseUpsert() {
+        return this.useUpsert;
+    }
+
+    /**
+     * Sets boolean value to use upsert operation
+     *
+     * @param useUpsert
+     */
+    public void setUseUpsert(Boolean useUpsert) {
+        this.useUpsert = useUpsert;
+    }
 }
diff --git a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/CosmosDBSourceTask.java b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/CosmosDBSourceTask.java
index c130e51..f30284d 100644
--- a/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/CosmosDBSourceTask.java
+++ b/src/main/java/com/microsoft/azure/cosmosdb/kafka/connect/source/CosmosDBSourceTask.java
@@ -1,5 +1,6 @@
 package com.microsoft.azure.cosmosdb.kafka.connect.source;
 
+import com.microsoft.azure.cosmosdb.kafka.connect.SettingDefaults;
 import com.azure.cosmos.*;
 import com.azure.cosmos.models.*;
 import com.fasterxml.jackson.databind.JsonNode;
@@ -147,6 +148,7 @@ public class CosmosDBSourceTask extends SourceTask {
                 .key(this.settings.getKey())
                 .consistencyLevel(ConsistencyLevel.SESSION)
                 .contentResponseOnWriteEnabled(true)
+                .userAgentSuffix(SettingDefaults.COSMOS_CLIENT_USER_AGENT_SUFFIX+version())
                 .buildAsyncClient();
     }
 


Test patch:
diff --git a/src/test/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/CosmosDBSinkTaskTest.java b/src/test/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/CosmosDBSinkTaskTest.java
index c3533b8..b997f21 100644
--- a/src/test/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/CosmosDBSinkTaskTest.java
+++ b/src/test/java/com/microsoft/azure/cosmosdb/kafka/connect/sink/CosmosDBSinkTaskTest.java
@@ -30,6 +30,7 @@ public class CosmosDBSinkTaskTest {
     private final String topicName = "testtopic";
     private final String containerName = "container666";
     private final String databaseName = "fakeDatabase312";
+    private final Boolean upsertFalse = false;
     private CosmosDBSinkTask testTask;
     private CosmosClient mockCosmosClient;
     private CosmosContainer mockContainer;
@@ -42,6 +43,7 @@ public class CosmosDBSinkTaskTest {
         SinkSettings settings = new SinkSettings();
         settings.setTopicContainerMap(TopicContainerMap.deserialize(topicName + "#" + containerName));
         settings.setDatabaseName(databaseName);
+        settings.setUseUpsert(upsertFalse);
         FieldUtils.writeField(testTask, "settings", settings, true);
 
         //Mock the Cosmos SDK