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

Issue:
Add a config property to append a string to the user agent.
**## Problem Statement**
To determine whether connections and records are coming from Confluent Cloud, we will need to append a string to the user agent.

**## Proposed Solution**
It could be a config property called "provider" with null by default.


## Next Steps

- [ ] Team consensus to proceed
- [ ] Schedule Design Session
- [ ] Complete Design Review


Pull request description:
adding new providerName config
## Type of PR

- [ ] Documentation changes
- [x] Code changes
- [x] Test changes
- [ ] CI-CD changes
- [ ] GitHub Template changes

## Purpose of PR
adding new provider config to allow us to differentiate between ConfluentCloud managed, and self managed. 

## Issues Closed or Referenced
- Closes #376 


Patch:
diff --git a/src/main/java/com/azure/cosmos/kafka/connect/CosmosDBConfig.java b/src/main/java/com/azure/cosmos/kafka/connect/CosmosDBConfig.java
index a083e11..8745f87 100644
--- a/src/main/java/com/azure/cosmos/kafka/connect/CosmosDBConfig.java
+++ b/src/main/java/com/azure/cosmos/kafka/connect/CosmosDBConfig.java
@@ -1,5 +1,6 @@
 package com.azure.cosmos.kafka.connect;
 
+import org.apache.commons.lang3.ObjectUtils.Null;
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigDef.Importance;
@@ -32,12 +33,16 @@ public class CosmosDBConfig extends AbstractConfig {
         + "For example: topic1#con1,topic2#con2.";
     private static final String COSMOS_CONTAINER_TOPIC_MAP_DISPLAY = "Topic-Container map";
 
+    public static final String  COSMOS_PROVIDER_NAME_CONF = "connect.cosmos.provider.name";
+    private static final String COSMOS_PROVIDER_NAME_DEFAULT = null;
+
     public static final int COSMOS_DATABASE_GROUP_ORDER = 2;
     public static final String COSMOS_CLIENT_USER_AGENT_SUFFIX = "APN/1.0 Microsoft/1.0 KafkaConnect/";
-
+    
     private String connEndpoint;
     private String connKey;
     private String databaseName;
+    private String providerName;
     private TopicContainerMap topicContainerMap = TopicContainerMap.empty();
 
     public CosmosDBConfig(ConfigDef config, Map<String, String> parsedConfig) {
@@ -47,6 +52,7 @@ public class CosmosDBConfig extends AbstractConfig {
         connKey = this.getPassword(COSMOS_CONN_KEY_CONF).value();
         databaseName = this.getString(COSMOS_DATABASE_NAME_CONF);
         topicContainerMap = TopicContainerMap.deserialize(this.getString(COSMOS_CONTAINER_TOPIC_MAP_CONF));
+        providerName = this.getString(COSMOS_PROVIDER_NAME_CONF);
     }
 
     public CosmosDBConfig(Map<String, String> parsedConfig) {
@@ -89,6 +95,12 @@ public class CosmosDBConfig extends AbstractConfig {
                 connectionGroupOrder++,
                 Width.LONG,
                 COSMOS_CONN_KEY_DISPLAY
+            )
+            .defineInternal(
+                COSMOS_PROVIDER_NAME_CONF, 
+                Type.STRING, 
+                COSMOS_PROVIDER_NAME_DEFAULT,
+                Importance.LOW        
             );
     }
 
@@ -142,6 +154,9 @@ public class CosmosDBConfig extends AbstractConfig {
         return this.topicContainerMap;
     }
 
+    public String getProviderName() {
+        return this.providerName;
+    }
 }
 
 


Test patch:
diff --git a/src/test/java/com/azure/cosmos/kafka/connect/CosmosConfigTest.java b/src/test/java/com/azure/cosmos/kafka/connect/CosmosConfigTest.java
new file mode 100644
index 0000000..203239f
--- /dev/null
+++ b/src/test/java/com/azure/cosmos/kafka/connect/CosmosConfigTest.java
@@ -0,0 +1,47 @@
+package com.azure.cosmos.kafka.connect;
+
+import org.apache.kafka.common.config.ConfigException;
+import org.junit.Test;
+
+import java.util.HashMap;
+
+import static org.junit.Assert.*;
+
+public class CosmosConfigTest {
+    private static final String COSMOS_URL = "https://<cosmosinstance-name>.documents.azure.com:443/";
+
+    public static HashMap<String, String> setupConfigsWithProvider() {
+        HashMap<String, String> configs = new HashMap<>();
+        configs.put(CosmosDBConfig.COSMOS_CONN_ENDPOINT_CONF, COSMOS_URL);
+        configs.put(CosmosDBConfig.COSMOS_CONN_KEY_CONF, "mykey");
+        configs.put(CosmosDBConfig.COSMOS_DATABASE_NAME_CONF, "mydb");
+        configs.put(CosmosDBConfig.COSMOS_CONTAINER_TOPIC_MAP_CONF, "mytopic5#mycontainer6");
+        configs.put(CosmosDBConfig.COSMOS_PROVIDER_NAME_CONF, "myprovider");
+
+        return configs;
+    }
+
+    public static HashMap<String, String> setupConfigs() {
+        HashMap<String, String> configs = new HashMap<>();
+        configs.put(CosmosDBConfig.COSMOS_CONN_ENDPOINT_CONF, COSMOS_URL);
+        configs.put(CosmosDBConfig.COSMOS_CONN_KEY_CONF, "mykey");
+        configs.put(CosmosDBConfig.COSMOS_DATABASE_NAME_CONF, "mydb");
+        configs.put(CosmosDBConfig.COSMOS_CONTAINER_TOPIC_MAP_CONF, "mytopic5#mycontainer6");
+        
+        return configs;
+    }
+
+    @Test
+    public void shouldHaveDefaultValues() {
+        // Adding required Configuration with no default value.
+        CosmosDBConfig config = new CosmosDBConfig(setupConfigs());
+        assertNull("Provider Name should be null unless set", config.getProviderName());
+    }
+    
+    @Test
+    public void shouldAcceptValidConfig() {
+        // Adding required Configuration with no default value.
+        CosmosDBConfig config = new CosmosDBConfig(setupConfigsWithProvider());
+        assertEquals("myprovider", config.getProviderName());
+    }    
+}
diff --git a/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConnectorTest.java b/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConnectorTest.java
index 36ddfe7..aee24f9 100644
--- a/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConnectorTest.java
+++ b/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConnectorTest.java
@@ -42,6 +42,7 @@ public class CosmosDBSinkConnectorTest {
         CosmosDBSinkConfig.COSMOS_CONN_ENDPOINT_CONF, "https://endpoint:port/",
         CosmosDBSinkConfig.COSMOS_CONN_KEY_CONF, "superSecretPassword",
         CosmosDBSinkConfig.COSMOS_DATABASE_NAME_CONF, "superAwesomeDatabase",
+        CosmosDBSinkConfig.COSMOS_PROVIDER_NAME_CONF, "superAwesomeProvider",
         CosmosDBSinkConfig.COSMOS_CONTAINER_TOPIC_MAP_CONF, "topic#container"
     ));
     Map<String, List<String>> errorMessages = config.configValues().stream()
@@ -61,6 +62,7 @@ public class CosmosDBSinkConnectorTest {
         CosmosDBSinkConfig.COSMOS_CONN_ENDPOINT_CONF, "https://endpoint:port/",
         CosmosDBSinkConfig.COSMOS_CONN_KEY_CONF, "superSecretPassword",
         CosmosDBSinkConfig.COSMOS_DATABASE_NAME_CONF, "superAwesomeDatabase",
+        CosmosDBSinkConfig.COSMOS_PROVIDER_NAME_CONF, "superAwesomeProvider",
         CosmosDBSinkConfig.COSMOS_CONTAINER_TOPIC_MAP_CONF, "topic#container"
     ));
     for (ConfigValue value : config.configValues()) {