Repository: microsoft/kafka-connect-cosmosdb
Pull request #449 (2022-02-09 22:50:34)

Merging latest dev to main
## Type of PR

- [ ] Documentation changes
- [ ] Code changes
- [ ] Test changes
- [ ] CI-CD changes
- [ ] GitHub Template changes

## Purpose of PR

## Observability + Testing
- What changes or considerations did you make in relation to observability?

- Did you add testing to account for any new or changed work?


## Review notes

## Issues Closed or Referenced

- Closes #<issue number> (this will automatically close the issue when the PR closes)
- References #<issue number> (this references the issue but does not close with PR)



Patch:
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 787c265..cf3cc5b 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -1,5 +1,12 @@
 ## Release History
 
+### 1.3.0 (2022-02-09)
+#### New Features
+* Added error tolerance config into CosmosDBSink - [PR 443](https://github.com/microsoft/kafka-connect-cosmosdb/pull/443)
+* Added endpoint/key config validation - [PR 446](https://github.com/microsoft/kafka-connect-cosmosdb/pull/446)
+* Updated `kafka-connect` version to 2.8.1.
+* Updated `azure-cosmos` version to 4.25.0.
+
 ### 1.2.5 (2022-02-01)
 #### Key Bug Fixes
 * Fixed NullPointerException issue with nullable structs. See this [PR 439](https://github.com/microsoft/kafka-connect-cosmosdb/pull/439).
diff --git a/README.md b/README.md
index f07238d..5e4c75e 100644
--- a/README.md
+++ b/README.md
@@ -6,7 +6,7 @@
 
 ## Introduction
 
-This project is pre-production. File any issues / feature requests / questions etc. you may have in the [Issues](https://github.com/microsoft/kafka-connect-cosmosdb/issues) for this repo.
+File any issues / feature requests / questions etc. you may have in the [Issues](https://github.com/microsoft/kafka-connect-cosmosdb/issues) for this repo.
 
 This project provides connectors for [Kafka Connect](http://kafka.apache.org/documentation.html#connect) to read from and write data to [Azure Cosmos DB](https://azure.microsoft.com/en-us/services/cosmos-db/).
 
@@ -148,6 +148,9 @@ For more information on the performance tests run for the Sink and Source connec
 
 Refer to the [Performance Environment Setup](./src/perf/README.md) for exact steps on deploying the performance test environment for the Connectors.
 
+## Dead Letter Queue
+We introduced the standard dead level queue from Kafka. For more info see: https://www.confluent.io/blog/kafka-connect-deep-dive-error-handling-dead-letter-queues
+
 ## Resources
 
 - [Kafka Connect](http://kafka.apache.org/documentation.html#connect)
diff --git a/pom.xml b/pom.xml
index c8f271f..de163dd 100644
--- a/pom.xml
+++ b/pom.xml
@@ -7,7 +7,7 @@
 
     <groupId>com.azure.cosmos.kafka</groupId>
     <artifactId>kafka-connect-cosmos</artifactId>
-    <version>1.2.5</version>
+    <version>1.3.0</version>
 
     <name> kafka-connect-cosmos</name>
     <url>https://github.com/microsoft/kafka-connect-cosmosdb</url>
@@ -16,7 +16,7 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <maven.compiler.source>11</maven.compiler.source>
         <maven.compiler.target>11</maven.compiler.target>
-        <kafka.version>2.7.0</kafka.version>
+        <kafka.version>2.8.1</kafka.version>
         <log4j.version>2.17.1</log4j.version>
         <slf4j.version>1.7.30</slf4j.version>
         <confluent-log4j.version>1.2.17-cp2</confluent-log4j.version>
@@ -48,7 +48,7 @@
         <dependency>
             <groupId>com.azure</groupId>
             <artifactId>azure-cosmos</artifactId>
-            <version>4.21.1</version>
+            <version>4.25.0</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 8745f87..d788946 100644
--- a/src/main/java/com/azure/cosmos/kafka/connect/CosmosDBConfig.java
+++ b/src/main/java/com/azure/cosmos/kafka/connect/CosmosDBConfig.java
@@ -1,6 +1,14 @@
 package com.azure.cosmos.kafka.connect;
 
-import org.apache.commons.lang3.ObjectUtils.Null;
+import static com.azure.cosmos.kafka.connect.CosmosDBConfig.CosmosClientBuilder.createClient;
+
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.kafka.connect.sink.CosmosDBSinkConfig;
+import java.net.InetAddress;
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.net.UnknownHostException;
+import java.util.regex.Pattern;
 import org.apache.kafka.common.config.AbstractConfig;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigDef.Importance;
@@ -10,16 +18,27 @@ import org.apache.kafka.common.config.ConfigDef.Validator;
 import org.apache.kafka.common.config.ConfigDef.Width;
 
 import java.util.Map;
+import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.ConfigValue;
 
-@SuppressWarnings ({"squid:S1854", "squid:S2160"})  // suppress unneeded int *groupOrder variables, equals method
+@SuppressWarnings({"squid:S1854", "squid:S2160"})  // suppress unneeded int *groupOrder variables, equals method
 public class CosmosDBConfig extends AbstractConfig {
     private static final Validator NON_EMPTY_STRING = new NonEmptyString();
     
+    private static final String VALID_ENDPOINT_SCHEME = "https";
+    private static final int VALID_ENDPOINT_PORT = 443;
+    private static final Pattern VALID_ENDPOINT_COSMOS_INSTANCE_PATTERN = Pattern.compile("[a-z0-9\\-]{3,44}");
+   
+    public static final String TOLERANCE_ON_ERROR_CONFIG = "errors.tolerance";
+    public static final String TOLERANCE_ON_ERROR_DOC = 
+        "Error tolerance level.\n"
+            + "'none' for fail on error. 'all' for log and continue";
+       
     public static final String COSMOS_CONN_ENDPOINT_CONF = "connect.cosmos.connection.endpoint";
     private static final String COSMOS_CONN_ENDPOINT_DOC = "Cosmos endpoint URL.";
     private static final String COSMOS_CONN_ENDPOINT_DISPLAY = "Cosmos Endpoint";
-
-    public static final String COSMOS_CONN_KEY_CONF = "connect.cosmos.master.key";
+    
+    public static final String COSMOS_CONN_KEY_CONF = "connect.cosmos.master.key";             
     private static final String COSMOS_CONN_KEY_DOC = "Cosmos connection master (primary) key.";
     private static final String COSMOS_CONN_KEY_DISPLAY = "Cosmos Connection Key";
 
@@ -28,17 +47,22 @@ public class CosmosDBConfig extends AbstractConfig {
     private static final String COSMOS_DATABASE_NAME_DISPLAY = "Cosmos Database name";
 
     public static final String COSMOS_CONTAINER_TOPIC_MAP_CONF = "connect.cosmos.containers.topicmap";
-    private static final String COSMOS_CONTAINER_TOPIC_MAP_DOC = 
-        "A comma delimited list of Kafka topics mapped to Cosmos containers.\n" 
-        + "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_CONTAINER_TOPIC_MAP_DOC =
+            "A comma delimited list of Kafka topics mapped to Cosmos containers.\n"
+                    + "For example: topic1#con1,topic2#con2.";    
+        
+    public static final String COSMOS_PROVIDER_NAME_CONF = "connect.cosmos.provider.name";
     private static final String COSMOS_PROVIDER_NAME_DEFAULT = null;
 
+  private static final String INVALID_TOPIC_MAP_FORMAT =
+        "Invalid entry for topic-container map. The topic-container map should be a comma-delimited "
+            + "list of Kafka topic to Cosmos containers. Each mapping should be a pair of Kafka "
+            + "topic and Cosmos container separated by '#'. For example: topic1#con1,topic2#con2.";
+
     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;
@@ -61,7 +85,7 @@ public class CosmosDBConfig extends AbstractConfig {
 
     public static ConfigDef getConfig() {
         ConfigDef result = new ConfigDef();
-        
+
         defineConnectionConfigs(result);
         defineDatabaseConfigs(result);
 
@@ -71,71 +95,77 @@ public class CosmosDBConfig extends AbstractConfig {
     private static void defineConnectionConfigs(ConfigDef result) {
         final String connectionGroupName = "Connection";
         int connectionGroupOrder = 0;
-        
+
         result
-            .define(
-                COSMOS_CONN_ENDPOINT_CONF,
-                Type.STRING,
-                ConfigDef.NO_DEFAULT_VALUE,
-                NON_EMPTY_STRING,
-                Importance.HIGH,
-                COSMOS_CONN_ENDPOINT_DOC,
-                connectionGroupName,
-                connectionGroupOrder++,
-                Width.LONG,
-                COSMOS_CONN_ENDPOINT_DISPLAY
-            )
-            .define(
-                COSMOS_CONN_KEY_CONF,
-                Type.PASSWORD,
-                ConfigDef.NO_DEFAULT_VALUE,
-                Importance.HIGH,
-                COSMOS_CONN_KEY_DOC,
-                connectionGroupName,
-                connectionGroupOrder++,
-                Width.LONG,
-                COSMOS_CONN_KEY_DISPLAY
-            )
-            .defineInternal(
-                COSMOS_PROVIDER_NAME_CONF, 
-                Type.STRING, 
-                COSMOS_PROVIDER_NAME_DEFAULT,
-                Importance.LOW        
-            );
+                .define(
+                        COSMOS_CONN_ENDPOINT_CONF,
+                        Type.STRING,
+                        ConfigDef.NO_DEFAULT_VALUE,
+                        NON_EMPTY_STRING,
+                        Importance.HIGH,
+                        COSMOS_CONN_ENDPOINT_DOC,
+                        connectionGroupName,
+                        connectionGroupOrder++,
+                        Width.LONG,
+                        COSMOS_CONN_ENDPOINT_DISPLAY
+                )
+                .define(
+                        COSMOS_CONN_KEY_CONF,
+                        Type.PASSWORD,
+                        ConfigDef.NO_DEFAULT_VALUE,
+                        Importance.HIGH,
+                        COSMOS_CONN_KEY_DOC,
+                        connectionGroupName,
+                        connectionGroupOrder++,
+                        Width.LONG,
+                        COSMOS_CONN_KEY_DISPLAY
+                ).define(
+                        TOLERANCE_ON_ERROR_CONFIG,
+                        Type.STRING,
+                        "none",
+                        Importance.MEDIUM,
+                        TOLERANCE_ON_ERROR_DOC
+                )
+                .defineInternal(
+                        COSMOS_PROVIDER_NAME_CONF,
+                        Type.STRING,
+                        COSMOS_PROVIDER_NAME_DEFAULT,
+                        Importance.LOW
+                );
     }
 
     private static void defineDatabaseConfigs(ConfigDef result) {
         final String databaseGroupName = "Database";
         int databaseGroupOrder = 0;
-        
+
         // When adding new config defines below, update COSMOS_DATABASE_GROUP_ORDER
         // This way, the source/sink configs will resume order from the right position.
 
         result
-            .define(
-                COSMOS_DATABASE_NAME_CONF,
-                Type.STRING,
-                ConfigDef.NO_DEFAULT_VALUE,
-                NON_EMPTY_STRING,
-                Importance.HIGH,
-                COSMOS_DATABASE_NAME_DOC,
-                databaseGroupName,
-                databaseGroupOrder++,
-                Width.MEDIUM,
-                COSMOS_DATABASE_NAME_DISPLAY
-            )
-            .define(
-                COSMOS_CONTAINER_TOPIC_MAP_CONF,
-                Type.STRING,
-                ConfigDef.NO_DEFAULT_VALUE,
-                NON_EMPTY_STRING,
-                Importance.HIGH,
-                COSMOS_CONTAINER_TOPIC_MAP_DOC,
-                databaseGroupName,
-                databaseGroupOrder++,
-                Width.MEDIUM,
-                COSMOS_CONTAINER_TOPIC_MAP_DISPLAY
-            );
+                .define(
+                        COSMOS_DATABASE_NAME_CONF,
+                        Type.STRING,
+                        ConfigDef.NO_DEFAULT_VALUE,
+                        NON_EMPTY_STRING,
+                        Importance.HIGH,
+                        COSMOS_DATABASE_NAME_DOC,
+                        databaseGroupName,
+                        databaseGroupOrder++,
+                        Width.MEDIUM,
+                        COSMOS_DATABASE_NAME_DISPLAY
+                )
+                .define(
+                        COSMOS_CONTAINER_TOPIC_MAP_CONF,
+                        Type.STRING,
+                        ConfigDef.NO_DEFAULT_VALUE,
+                        NON_EMPTY_STRING,
+                        Importance.HIGH,
+                        COSMOS_CONTAINER_TOPIC_MAP_DOC,
+                        databaseGroupName,
+                        databaseGroupOrder++,
+                        Width.MEDIUM,
+                        COSMOS_CONTAINER_TOPIC_MAP_DISPLAY
+                );
     }
 
     public String getConnEndpoint() {
@@ -157,6 +187,79 @@ public class CosmosDBConfig extends AbstractConfig {
     public String getProviderName() {
         return this.providerName;
     }
-}
 
+    // VisibleForTesting
+    public static void validateEndpoint(String endpoint) throws URISyntaxException, UnknownHostException {
+        URI uri = new URI(endpoint);
+        if (!VALID_ENDPOINT_SCHEME.equalsIgnoreCase(uri.getScheme())) {
+            throw new ConfigException("Endpoint must have scheme: " + VALID_ENDPOINT_SCHEME);
+        }
+        if (uri.getPort() != -1 && VALID_ENDPOINT_PORT != uri.getPort()) {
+            throw new ConfigException("Endpoint must have port: " + VALID_ENDPOINT_PORT);
+        }
+        if (uri.getPath() != null && !uri.getPath().isEmpty() && !uri.getPath().equalsIgnoreCase("/")) {
+            throw new ConfigException("Endpoint must not contain path: " + uri.getPath());
+        }
+        if (uri.getQuery() != null) {
+            throw new ConfigException("Endpoint must not contain query component: " + uri.getQuery());
+        }
+        if (uri.getFragment() != null) {
+            throw new ConfigException("Endpoint must not contain fragment: " + uri.getFragment());
+        }
+        String host = uri.getHost();
+        String cosmosInstance = host.split("\\.")[0];
+        if (!VALID_ENDPOINT_COSMOS_INSTANCE_PATTERN.matcher(cosmosInstance).matches()) {
+            throw new ConfigException("Invalid cosmos instance: " + cosmosInstance);
+        }
+        InetAddress ia = InetAddress.getByName(host);
+        if (ia.isLoopbackAddress() || ia.isLoopbackAddress() || ia.isSiteLocalAddress()) {
+            throw new ConfigException("Invalid host: " + host);
+        }
+    }
+
+    public static void validateConnection(Map<String, String> connectorConfigs, Map<String, ConfigValue> configValues) {
+        String endpoint = connectorConfigs.get(CosmosDBSinkConfig.COSMOS_CONN_ENDPOINT_CONF);
+        try {
+            validateEndpoint(endpoint);
+        } catch (Exception e) {
+            configValues.get(CosmosDBSinkConfig.COSMOS_CONN_ENDPOINT_CONF)
+                .addErrorMessage("Invalid endpoint: " + e.getMessage());
+        }
+
+        String key = connectorConfigs.get(CosmosDBSinkConfig.COSMOS_CONN_KEY_CONF);
+        try {
+            createClient(endpoint, key);
+        } catch (Exception e) {
+            configValues.get(CosmosDBSinkConfig.COSMOS_CONN_ENDPOINT_CONF)
+                .addErrorMessage("Could not connect to endpoint with error: " + e.getMessage());
+            configValues.get(CosmosDBSinkConfig.COSMOS_CONN_KEY_CONF)
+                .addErrorMessage("Could not connect to endpoint with error: " + e.getMessage());
+        }
+    }
+
+    public static void validateTopicMap(Map<String, String> connectorConfigs,
+        Map<String, ConfigValue> configValues) {
+
+        String topicMap = connectorConfigs.get(CosmosDBSinkConfig.COSMOS_CONTAINER_TOPIC_MAP_CONF);
+        try {
+            TopicContainerMap.deserialize(topicMap);
+        } catch (Exception e) {
+            configValues.get(CosmosDBSinkConfig.COSMOS_CONTAINER_TOPIC_MAP_CONF)
+                .addErrorMessage(INVALID_TOPIC_MAP_FORMAT);
+        }
+    }
 
+    /* separate class for mocking static method in tests */
+    public static class CosmosClientBuilder {
+        public static void createClient(String endpoint, String key) {
+            try (CosmosClient unused = new com.azure.cosmos.CosmosClientBuilder()
+                .endpoint(endpoint)
+                .key(key)
+                .userAgentSuffix(COSMOS_CLIENT_USER_AGENT_SUFFIX
+                    + CosmosDBConfig.class.getPackage().getImplementationVersion())
+                .buildClient()) {
+                // Just try to create the client to validate connectivity to the endpoint with key.
+            }
+        }
+    }
+}
diff --git a/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConnector.java b/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConnector.java
index 2bc4ee2..e0ca9ef 100644
--- a/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConnector.java
+++ b/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkConnector.java
@@ -1,15 +1,14 @@
 package com.azure.cosmos.kafka.connect.sink;
 
+import static com.azure.cosmos.kafka.connect.CosmosDBConfig.validateConnection;
+import static com.azure.cosmos.kafka.connect.CosmosDBConfig.validateTopicMap;
+
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
-import com.azure.cosmos.CosmosClient;
-import com.azure.cosmos.CosmosClientBuilder;
-import com.azure.cosmos.kafka.connect.CosmosDBConfig;
-import com.azure.cosmos.kafka.connect.TopicContainerMap;
 import org.apache.kafka.common.config.Config;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigValue;
@@ -22,12 +21,6 @@ import org.slf4j.LoggerFactory;
  * A Sink connector that publishes topic messages to CosmosDB.
  */
 public class CosmosDBSinkConnector extends SinkConnector {
-
-    private static final String INVALID_TOPIC_MAP_FORMAT =
-        "Invalid entry for topic-container map. The topic-container map should be a comma-delimited "
-            + "list of Kafka topic to Cosmos containers. Each mapping should be a pair of Kafka "
-            + "topic and Cosmos container separated by '#'. For example: topic1#con1,topic2#con2.";
-
     private static final Logger logger = LoggerFactory.getLogger(CosmosDBSinkConnector.class);
     private Map<String, String> configProps;
 
@@ -82,41 +75,4 @@ public class CosmosDBSinkConnector extends SinkConnector {
 
         return config;
     }
-
-    private void validateConnection(Map<String, String> connectorConfigs, Map<String, ConfigValue> configValues) {
-        String endpoint = connectorConfigs.get(CosmosDBSinkConfig.COSMOS_CONN_ENDPOINT_CONF);
-        String key = connectorConfigs.get(CosmosDBSinkConfig.COSMOS_CONN_KEY_CONF);
-        try {
-            createClient(endpoint, key);
-        } catch (Exception e) {
-            configValues.get(CosmosDBSinkConfig.COSMOS_CONN_ENDPOINT_CONF)
-                .addErrorMessage("Could not connect to endpoint with error: " + e.getMessage());
-            configValues.get(CosmosDBSinkConfig.COSMOS_CONN_KEY_CONF)
-                .addErrorMessage("Could not connect to endpoint with error: " + e.getMessage());
-        }
-    }
-
-    private void validateTopicMap(Map<String, String> connectorConfigs,
-        Map<String, ConfigValue> configValues) {
-
-        String topicMap = connectorConfigs.get(CosmosDBSinkConfig.COSMOS_CONTAINER_TOPIC_MAP_CONF);
-        try {
-            TopicContainerMap.deserialize(topicMap);
-        } catch (Exception e) {
-            configValues.get(CosmosDBSinkConfig.COSMOS_CONTAINER_TOPIC_MAP_CONF)
-                .addErrorMessage(INVALID_TOPIC_MAP_FORMAT);
-        }
-    }
-
-    // visible for testing
-    void createClient(String endpoint, String key) {
-        try (CosmosClient unused = new CosmosClientBuilder()
-            .endpoint(endpoint)
-            .key(key)
-            .userAgentSuffix(CosmosDBConfig.COSMOS_CLIENT_USER_AGENT_SUFFIX + version())
-            .buildClient()) {
-            // Just try to create the client to validate connectivity to the endpoint with key.
-        }
-    }
 }
-
diff --git a/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTask.java b/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTask.java
index 83c7015..468a941 100644
--- a/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTask.java
+++ b/src/main/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTask.java
@@ -1,24 +1,29 @@
 package com.azure.cosmos.kafka.connect.sink;
 
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
 import com.azure.cosmos.CosmosClient;
 import com.azure.cosmos.CosmosClientBuilder;
 import com.azure.cosmos.CosmosContainer;
+import com.azure.cosmos.CosmosException;
 import com.azure.cosmos.implementation.BadRequestException;
 import com.azure.cosmos.kafka.connect.CosmosDBConfig;
 import com.azure.cosmos.kafka.connect.sink.id.strategy.AbstractIdStrategyConfig;
 import com.azure.cosmos.kafka.connect.sink.id.strategy.IdStrategy;
 
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
-import java.util.stream.Collectors;
-
 import org.apache.commons.collections4.CollectionUtils;
 import org.apache.kafka.connect.data.Struct;
+import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.apache.kafka.connect.sink.SinkTask;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+
+import static com.azure.cosmos.kafka.connect.CosmosDBConfig.TOLERANCE_ON_ERROR_CONFIG;
 
 /**
  * Implements the Kafka Task for the CosmosDB Sink Connector
@@ -38,6 +43,7 @@ public class CosmosDBSinkTask extends SinkTask {
         logger.trace("Sink task started.");
         this.config = new CosmosDBSinkConfig(map);
 
+
         this.client = new CosmosClientBuilder()
                 .endpoint(config.getConnEndpoint())
                 .key(config.getConnKey())
@@ -66,6 +72,9 @@ public class CosmosDBSinkTask extends SinkTask {
             String containerName = entry.getKey();
             CosmosContainer container = client.getDatabase(config.getDatabaseName()).getContainer(containerName);
             for (SinkRecord record : entry.getValue()) {
+                if (record.key() != null) {
+                    MDC.put(String.format("CosmosDbSink-%s", containerName), record.key().toString());
+                }
                 logger.debug("Writing record, value type: {}", record.value().getClass().getName());
                 logger.debug("Key Schema: {}", record.keySchema());
                 logger.debug("Value schema: {}", record.valueSchema());
@@ -83,13 +92,37 @@ public class CosmosDBSinkTask extends SinkTask {
 
                 try {
                     addItemToContainer(container, recordValue);
-                } catch (BadRequestException bre) {
-                    throw new CosmosDBWriteException(record, bre);
+                } catch (CosmosException | ConnectException bre) {
+                    sendToDlqIfConfigured(record, bre);
+                } finally {
+                    MDC.clear();
                 }
             }
         }
     }
 
+    /**
+     * Sends data to a dead letter queue
+     *
+     * @param record the kafka record that contains error
+     */
+    private void sendToDlqIfConfigured(SinkRecord record, RuntimeException exception) {
+
+        if (context != null && context.errantRecordReporter() != null) {
+            context.errantRecordReporter().report(record, exception);
+            if (!config.getString(TOLERANCE_ON_ERROR_CONFIG).equalsIgnoreCase("all")) {
+                throw exception;
+            }
+        } else {
+            if (config.getString(TOLERANCE_ON_ERROR_CONFIG).equalsIgnoreCase("all")) {
+                logger.error("Could not upload record to CosmosDb, but tolerance is set to all. Value: {}."
+                        + " Error: {}", record.toString(), exception.getMessage());
+            } else {
+                throw new CosmosDBWriteException(record, exception);
+            }
+        }
+    }
+
     private void maybeInsertId(Object recordValue, SinkRecord sinkRecord) {
         if (!(recordValue instanceof Map)) {
             return;
diff --git a/src/main/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceConnector.java b/src/main/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceConnector.java
index fa41ea9..abee04e 100644
--- a/src/main/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceConnector.java
+++ b/src/main/java/com/azure/cosmos/kafka/connect/source/CosmosDBSourceConnector.java
@@ -1,8 +1,15 @@
 package com.azure.cosmos.kafka.connect.source;
 
+import static com.azure.cosmos.kafka.connect.CosmosDBConfig.validateConnection;
+import static com.azure.cosmos.kafka.connect.CosmosDBConfig.validateTopicMap;
+
+import java.util.function.Function;
+import java.util.stream.Collectors;
 import org.apache.commons.lang3.RandomUtils;
+import org.apache.kafka.common.config.Config;
 import org.apache.kafka.common.config.ConfigDef;
 import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.ConfigValue;
 import org.apache.kafka.connect.connector.Task;
 import org.apache.kafka.connect.errors.ConnectException;
 import org.apache.kafka.connect.source.SourceConnector;
@@ -75,4 +82,20 @@ public class CosmosDBSourceConnector extends SourceConnector {
     public String version() {
         return this.getClass().getPackage().getImplementationVersion();
     }
+
+    @Override
+    public Config validate(Map<String, String> connectorConfigs) {
+        Config config = super.validate(connectorConfigs);
+        if (config.configValues().stream().anyMatch(cv -> !cv.errorMessages().isEmpty())) {
+            return config;
+        }
+
+        Map<String, ConfigValue> configValues = config.configValues().stream().collect(
+            Collectors.toMap(ConfigValue::name, Function.identity()));
+
+        validateConnection(connectorConfigs, configValues);
+        validateTopicMap(connectorConfigs, configValues);
+
+        return config;
+    }
 }


Test patch:
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 aee24f9..be05f01 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
@@ -1,21 +1,25 @@
 package com.azure.cosmos.kafka.connect.sink;
 
+import com.azure.cosmos.kafka.connect.CosmosDBConfig;
+import com.azure.cosmos.kafka.connect.CosmosDBConfig.CosmosClientBuilder;
 import com.google.common.collect.ImmutableMap;
 import org.apache.kafka.common.config.Config;
+import org.apache.kafka.common.config.ConfigException;
 import org.apache.kafka.common.config.ConfigValue;
 import org.junit.Test;
 
 import java.util.List;
 import java.util.Map;
 import java.util.stream.Collectors;
+import org.mockito.MockedStatic;
 
 import static org.hamcrest.CoreMatchers.not;
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.collection.IsEmptyCollection.empty;
+import static org.junit.Assert.assertThrows;
+import static org.mockito.AdditionalAnswers.answerVoid;
 import static org.mockito.ArgumentMatchers.anyString;
-import static org.mockito.Mockito.doNothing;
-import static org.mockito.Mockito.doThrow;
-import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.mockStatic;
 
 public class CosmosDBSinkConnectorTest {
 
@@ -33,56 +37,72 @@ public class CosmosDBSinkConnectorTest {
 
   @Test
   public void testValidateCannotConnectToCosmos() {
-    CosmosDBSinkConnector connector = spy(CosmosDBSinkConnector.class);
-    doThrow(new IllegalArgumentException())
-        .when(connector)
-        .createClient(anyString(), anyString());
+    CosmosDBSinkConnector connector = new CosmosDBSinkConnector();
 
-    Config config = connector.validate(ImmutableMap.of(
-        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()
-        .collect(Collectors.toMap(ConfigValue::name, ConfigValue::errorMessages));
-    assertThat(errorMessages.get(CosmosDBSinkConfig.COSMOS_CONN_ENDPOINT_CONF), not(empty()));
-    assertThat(errorMessages.get(CosmosDBSinkConfig.COSMOS_CONN_KEY_CONF), not(empty()));
+    try (MockedStatic<CosmosClientBuilder> cosmosDBClient
+        = mockStatic(CosmosClientBuilder.class)) {
+
+      cosmosDBClient
+          .when(() -> CosmosClientBuilder.createClient(anyString(), anyString()))
+          .thenThrow(IllegalArgumentException.class);
+
+      Config config = connector.validate(ImmutableMap.of(
+          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()
+          .collect(Collectors.toMap(ConfigValue::name, ConfigValue::errorMessages));
+      assertThat(errorMessages.get(CosmosDBSinkConfig.COSMOS_CONN_ENDPOINT_CONF), not(empty()));
+      assertThat(errorMessages.get(CosmosDBSinkConfig.COSMOS_CONN_KEY_CONF), not(empty()));
+    }
   }
 
   @Test
   public void testValidateHappyPath() {
-    CosmosDBSinkConnector connector = spy(CosmosDBSinkConnector.class);
-    doNothing()
-        .when(connector)
-        .createClient(anyString(), anyString());
+    CosmosDBSinkConnector connector = new CosmosDBSinkConnector();
 
-    Config config = connector.validate(ImmutableMap.of(
-        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()) {
-      assertThat("Expecting empty error message for config " + value.name(),
-          value.errorMessages(), empty());
+    try (MockedStatic<CosmosClientBuilder> cosmosDBClient
+        = mockStatic(CosmosClientBuilder.class)) {
+      cosmosDBClient
+          .when(() -> CosmosClientBuilder.createClient(anyString(), anyString()))
+          .then(answerVoid((s1, s2) -> {
+          }));
+
+      Config config = connector.validate(ImmutableMap.of(
+          CosmosDBSinkConfig.COSMOS_CONN_ENDPOINT_CONF,
+          "https://cosmos-instance.documents.azure.com:443/",
+          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()) {
+        assertThat("Expecting empty error message for config " + value.name(),
+            value.errorMessages(), empty());
+      }
     }
   }
 
   @Test
   public void testValidateTopicMapValidFormat() {
-    CosmosDBSinkConnector connector = spy(CosmosDBSinkConnector.class);
-    doNothing()
-        .when(connector)
-        .createClient(anyString(), anyString());
-
-    invalidTopicMapString(connector, "topicOnly");
-    invalidTopicMapString(connector, "#containerOnly");
-    invalidTopicMapString(connector, ",,,,,");
-    invalidTopicMapString(connector, "###");
-    invalidTopicMapString(connector, "partially#correct,but,not#entirely");
+    try (MockedStatic<CosmosClientBuilder> cosmosDBConfig
+        = mockStatic(CosmosClientBuilder.class)) {
+
+      cosmosDBConfig
+          .when(() -> CosmosClientBuilder.createClient(anyString(), anyString()))
+          .then(answerVoid((s1, s2) -> {}));
+
+      CosmosDBSinkConnector connector = new CosmosDBSinkConnector();
+
+      invalidTopicMapString(connector, "topicOnly");
+      invalidTopicMapString(connector, "#containerOnly");
+      invalidTopicMapString(connector, ",,,,,");
+      invalidTopicMapString(connector, "###");
+      invalidTopicMapString(connector, "partially#correct,but,not#entirely");
+    }
   }
 
   private void invalidTopicMapString(CosmosDBSinkConnector connector, String topicMapConfig) {
@@ -96,4 +116,42 @@ public class CosmosDBSinkConnectorTest {
         .collect(Collectors.toMap(ConfigValue::name, ConfigValue::errorMessages));
     assertThat(errorMessages.get(CosmosDBSinkConfig.COSMOS_CONTAINER_TOPIC_MAP_CONF), not(empty()));
   }
+
+  @Test
+  public void testValidateEndpoint() throws Exception {
+    assertThrows(ConfigException.class,
+        () -> {
+          CosmosDBConfig.validateEndpoint("http://not.valid.schema");
+    });
+    assertThrows(ConfigException.class,
+        () -> {CosmosDBConfig.validateEndpoint("https://not.valid.port:1024");
+    });
+    assertThrows(ConfigException.class,
+        () -> {CosmosDBConfig.validateEndpoint("https://not.valid.path:443/not/valid/path");
+    });
+    assertThrows(ConfigException.class,
+        () -> {CosmosDBConfig.validateEndpoint("https://not.valid.query:443/?query=not-valid");
+    });
+    assertThrows(ConfigException.class,
+        () -> {CosmosDBConfig.validateEndpoint("https://not.valid.query:443/#fragement-not-valid");
+    });
+    assertThrows(ConfigException.class,
+        () -> {CosmosDBConfig.validateEndpoint("https://INSTANCE.documents.azure.com:443/");
+    });
+    assertThrows(ConfigException.class,
+        () -> {CosmosDBConfig.validateEndpoint("https://1.documents.azure.com:443/");
+    });
+    assertThrows(ConfigException.class,
+        () -> {CosmosDBConfig.validateEndpoint("https://longlonglonglonglonglonglonglonglonglonglonglonglonglonginstance.documents.azure.com:443/");
+    });
+    assertThrows(ConfigException.class,
+        () -> {CosmosDBConfig.validateEndpoint("https://localhost:443/");
+    });
+    assertThrows(ConfigException.class,
+        () -> {CosmosDBConfig.validateEndpoint("https://[::1]:443/");
+    });
+
+    CosmosDBConfig.validateEndpoint("https://cosmos-instance.documents.azure.com:443/");
+    CosmosDBConfig.validateEndpoint("https://cosmos-instance.documents.azure.com:443");
+  }
 }
diff --git a/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTaskTest.java b/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTaskTest.java
index 1ed209b..a6d1293 100644
--- a/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTaskTest.java
+++ b/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTaskTest.java
@@ -6,6 +6,7 @@ import com.azure.cosmos.CosmosContainer;
 import com.azure.cosmos.CosmosDatabase;
 import com.azure.cosmos.implementation.BadRequestException;
 import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import org.apache.commons.lang3.reflect.FieldUtils;
diff --git a/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTaskTestNotFails.java b/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTaskTestNotFails.java
new file mode 100644
index 0000000..70286b3
--- /dev/null
+++ b/src/test/java/com/azure/cosmos/kafka/connect/sink/CosmosDBSinkTaskTestNotFails.java
@@ -0,0 +1,92 @@
+package com.azure.cosmos.kafka.connect.sink;
+
+
+import com.azure.cosmos.CosmosClient;
+import com.azure.cosmos.CosmosContainer;
+import com.azure.cosmos.CosmosDatabase;
+import com.azure.cosmos.implementation.BadRequestException;
+import com.fasterxml.jackson.core.JsonParseException;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import org.apache.commons.lang3.reflect.FieldUtils;
+import org.apache.kafka.connect.data.ConnectSchema;
+import org.apache.kafka.connect.data.Schema;
+import org.apache.kafka.connect.errors.ConnectException;
+import org.apache.kafka.connect.sink.ErrantRecordReporter;
+import org.apache.kafka.connect.sink.SinkRecord;
+import org.apache.kafka.connect.sink.SinkTaskContext;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
+import org.mockito.Mockito;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import static junit.framework.TestCase.assertNotNull;
+import static junit.framework.TestCase.fail;
+import static org.mockito.Mockito.*;
+
+public class CosmosDBSinkTaskTestNotFails {
+    private final String topicName = "testtopic";
+    private final String containerName = "container666";
+    private final String databaseName = "fakeDatabase312";
+    private CosmosDBSinkTask testTask;
+    private CosmosClient mockCosmosClient;
+    private CosmosContainer mockContainer;
+    private SinkTaskContext mockContext = Mockito.mock(SinkTaskContext.class);
+    private ErrantRecordReporter mockErrantReporter = Mockito.mock(ErrantRecordReporter.class);
+
+    @Before
+    public void setup() throws IllegalAccessException {
+        testTask = new CosmosDBSinkTask();
+
+        //Configure settings
+        Map<String, String> settingAssignment = CosmosDBSinkConfigTest.setupConfigs();
+        settingAssignment.put(CosmosDBSinkConfig.COSMOS_CONTAINER_TOPIC_MAP_CONF, topicName + "#" + containerName);
+        settingAssignment.put(CosmosDBSinkConfig.COSMOS_DATABASE_NAME_CONF, databaseName);
+        settingAssignment.put(CosmosDBSinkConfig.TOLERANCE_ON_ERROR_CONFIG, "all");
+        CosmosDBSinkConfig config = new CosmosDBSinkConfig(settingAssignment);
+        FieldUtils.writeField(testTask, "config", config, true);
+
+        //Mock the Cosmos SDK
+        mockCosmosClient = Mockito.mock(CosmosClient.class);
+        CosmosDatabase mockDatabase = Mockito.mock(CosmosDatabase.class);
+        when(mockCosmosClient.getDatabase(anyString())).thenReturn(mockDatabase);
+        mockContainer = Mockito.mock(CosmosContainer.class);
+        when(mockDatabase.getContainer(any())).thenReturn(mockContainer);
+        when(mockContext.errantRecordReporter()).thenReturn(mockErrantReporter);
+
+        FieldUtils.writeField(testTask, "client", mockCosmosClient, true);
+
+    }
+    @After()
+    public void resetContext() throws IllegalAccessException {
+        FieldUtils.writeField(testTask,  "context", null, true);
+    }
+
+
+    @Test
+    public void testPutMapThatFailsDoesNotStopTask() throws JsonProcessingException, IllegalAccessException {
+
+        Schema stringSchema = new ConnectSchema(Schema.Type.STRING);
+        Schema mapSchema = new ConnectSchema(Schema.Type.MAP);
+        when(mockContainer.upsertItem(any())).thenThrow(new BadRequestException("Something"));
+        SinkRecord record = new SinkRecord(topicName, 1, stringSchema, "nokey", mapSchema, "{", 0L);
+        testTask.put(List.of(record));
+    }
+    @Test
+    public void testPutMapThatFailsDoesNotStopTaskWithdlq() throws JsonProcessingException, IllegalAccessException {
+        FieldUtils.writeField(testTask,  "context", mockContext, true);
+        Schema stringSchema = new ConnectSchema(Schema.Type.STRING);
+        Schema mapSchema = new ConnectSchema(Schema.Type.MAP);
+        when(mockContainer.upsertItem(any())).thenThrow(new BadRequestException("Something"));
+        SinkRecord record = new SinkRecord(topicName, 1, stringSchema, "nokey", mapSchema, "{", 0L);
+        testTask.put(List.of(record));
+        verify(mockContext.errantRecordReporter(), times(1)).report(any(), any());
+    }
+}
+