Repository: microsoft/kafka-connect-cosmosdb
Pull request #416 (2021-11-22 20:31:08)

Version bump-1.2.0 dev to main
1. Version bump-1.2.0 dev to main
2. CosmosDB version increase 4.15.0 -> 4.21.1


Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
new file mode 100644
index 0000000..9655c3d
--- /dev/null
+++ b/CHANGELOG.md
@@ -0,0 +1,52 @@
+## Release History
+
+### 1.2.0 (2021-11-22)
+#### New Features
+* Updated `azure-cosmos` to v4.21.1
+
+### 1.1.0 (2021-07-08)
+#### New Features
+* TODO
+#### Key Bug Fixes
+* TODO
+
+### 1.0.10-beta (2021-07-06)
+#### New Features
+* TODO
+#### Key Bug Fixes
+* TODO
+
+
+### 1.0.9-beta (2021-03-26)
+#### New Features
+* TODO
+#### Key Bug Fixes
+* TODO
+
+
+### 1.0.7-beta (2021-03-16)
+#### New Features
+* TODO
+#### Key Bug Fixes
+* TODO
+
+
+### 1.0.5-beta (2021-03-01)
+#### New Features
+* TODO
+#### Key Bug Fixes
+* TODO
+
+
+### 1.0.4-beta (2021-02-24)
+#### New Features
+* Updates to the mainfest.json for release to refresh metadata on Confluent Hub
+
+
+### 1.0.3-beta (2021-02-16)
+#### New Features
+* This is the first public preview release for both Sink and Source Connectors which is also published on Confluent Hub.
+* Enhanced Integration Tests.
+* Added support for standalone mode.
+* Updated documentation and minor fixes.
+* Checkstyle Implementation.
diff --git a/pom.xml b/pom.xml
index 8a99b4b..62cc1c4 100644
--- a/pom.xml
+++ b/pom.xml
@@ -7,7 +7,7 @@
 
     <groupId>com.azure.cosmos.kafka</groupId>
     <artifactId>kafka-connect-cosmos</artifactId>
-    <version>1.1.0</version>
+    <version>1.2.0</version>
 
     <name> kafka-connect-cosmos</name>
     <url>https://github.com/microsoft/kafka-connect-cosmosdb</url>
@@ -48,7 +48,7 @@
         <dependency>
             <groupId>com.azure</groupId>
             <artifactId>azure-cosmos</artifactId>
-            <version>4.15.0</version>
+            <version>4.21.1</version>
         </dependency>
         <dependency>
             <groupId>com.jayway.jsonpath</groupId>
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()) {