Repository: microsoft/kafka-connect-cosmosdb
Pull request #446 (2022-02-09 01:00:47)

Reuse the same cosmosdb endpoint/key validation in source connector
## Type of PR

- [ ] Documentation changes
- [x] Code changes
- [x] Test changes
- [ ] CI-CD changes
- [ ] GitHub Template changes

## Purpose of PR
`CosmosDB` endpoint and master key validation logic is the same for both *SINK* connector and *SOURCE* connector, the shared code is moved into `CosmosDBConfig`.

## Observability + Testing
- What changes or considerations did you make in relation to observability?
  misconfigured `CosmosDB` endpoint or master key for *SOURCE* connector will fail validation.
- Did you add testing to account for any new or changed work?
  Yes


## 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/src/main/java/com/azure/cosmos/kafka/connect/CosmosDBConfig.java b/src/main/java/com/azure/cosmos/kafka/connect/CosmosDBConfig.java
index 64e7d18..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,30 +18,51 @@ 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
 public class CosmosDBConfig extends AbstractConfig {
-    public static final String COSMOS_CONN_ENDPOINT_CONF = "connect.cosmos.connection.endpoint";
-    public static final String COSMOS_CONN_KEY_CONF = "connect.cosmos.master.key";
-    public static final String COSMOS_DATABASE_NAME_CONF = "connect.cosmos.databasename";
-    public static final String COSMOS_CONTAINER_TOPIC_MAP_CONF = "connect.cosmos.containers.topicmap";
-    public static final String COSMOS_PROVIDER_NAME_CONF = "connect.cosmos.provider.name";
-    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/";
-    public static final String TOLERANCE_ON_ERROR_CONFIG = "errors.tolerance";
-    public static final String TOLERANCE_ON_ERROR_DOC = "Tolerance level. 'none' for failing on error. 'all' for log and continue";
     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";             
     private static final String COSMOS_CONN_KEY_DOC = "Cosmos connection master (primary) key.";
     private static final String COSMOS_CONN_KEY_DISPLAY = "Cosmos Connection Key";
+
+    public static final String COSMOS_DATABASE_NAME_CONF = "connect.cosmos.databasename";
     private static final String COSMOS_DATABASE_NAME_DOC = "Cosmos target database to write records into.";
     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_DISPLAY = "Topic-Container map";
     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";
+                    + "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;
@@ -158,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/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");
+  }
 }