Added support for MarkLogic Cloud auth

Fairly simple matter of just adding two config options and passing them along to the existing class that collects inputs for constructing a client.

Author: rjrudin

README.md

@@ -105,6 +105,16 @@ Both digest and basic authentication require the following properties to be conf
 - `ml.connection.username` = the name of the MarkLogic user to authenticate as
 - `ml.connection.password` = the password of the MarkLogic user
 
+### Configuring MarkLogic Cloud authentication
+
+Cloud authentication requires the following properties to be configured:
+
+- `ml.connection.basePath` = the base path in your MarkLogic Cloud instance that points to the REST API server you 
+  wish to connect to
+- `ml.connection.cloudApiKey` = the API key for authenticating with your MarkLogic Cloud instance
+
+You should also set `ml.connection.port` to 443 for connecting to MarkLogic Cloud.
+
 ### Configuring certificate authentication
 
 Certificate authentication requires the following properties to be configured:

build.gradle

@@ -20,6 +20,16 @@ repositories {
 
   // For testing
   mavenLocal()
+  maven {
+    url "https://nexus.marklogic.com/repository/maven-snapshots/"
+  }
+}
+
+// Do not cache changing modules
+configurations.all {
+  resolutionStrategy {
+    cacheChangingModulesFor 0, 'seconds'
+  }
 }
 
 configurations {
@@ -37,11 +47,12 @@ dependencies {
   compileOnly "org.apache.kafka:connect-runtime:${kafkaVersion}"
   compileOnly "org.slf4j:slf4j-api:1.7.36"
 
-  implementation "com.marklogic:marklogic-client-api:6.0.0"
-  implementation "com.marklogic:ml-javaclient-util:4.4.0"
+  implementation ('com.marklogic:ml-javaclient-util:4.5-SNAPSHOT') {
+    changing = true
+  }
 
   // Force DHF to use the latest version of ml-app-deployer, which minimizes security vulnerabilities
-  implementation "com.marklogic:ml-app-deployer:4.4.0"
+  implementation "com.marklogic:ml-app-deployer:4.5-SNAPSHOT"
 
   implementation "com.fasterxml.jackson.dataformat:jackson-dataformat-csv:2.14.1"
 

src/main/java/com/marklogic/kafka/connect/DefaultDatabaseClientConfigBuilder.java

@@ -40,6 +40,8 @@ public DatabaseClientConfig buildDatabaseClientConfig(Map<String, Object> parsed
 
         clientConfig.setHost((String) parsedConfig.get(MarkLogicConfig.CONNECTION_HOST));
         clientConfig.setPort((Integer) parsedConfig.get(MarkLogicConfig.CONNECTION_PORT));
+        clientConfig.setBasePath((String) parsedConfig.get(MarkLogicConfig.CONNECTION_BASE_PATH));
+
         String securityContextType = ((String) parsedConfig.get(MarkLogicConfig.CONNECTION_SECURITY_CONTEXT_TYPE)).toUpperCase();
         clientConfig.setSecurityContextType(SecurityContextType.valueOf(securityContextType));
 
@@ -75,6 +77,8 @@ public DatabaseClientConfig buildDatabaseClientConfig(Map<String, Object> parsed
             configureSimpleSsl(clientConfig);
         }
 
+        clientConfig.setCloudApiKey((String)parsedConfig.get(MarkLogicConfig.CONNECTION_CLOUD_API_KEY));
+
         return clientConfig;
     }
 

src/main/java/com/marklogic/kafka/connect/MarkLogicConfig.java

@@ -31,6 +31,7 @@
 public class MarkLogicConfig extends AbstractConfig {
     public static final String CONNECTION_HOST = "ml.connection.host";
     public static final String CONNECTION_PORT = "ml.connection.port";
+    public static final String CONNECTION_BASE_PATH = "ml.connection.basePath";
     public static final String CONNECTION_DATABASE = "ml.connection.database";
     public static final String CONNECTION_SECURITY_CONTEXT_TYPE = "ml.connection.securityContextType";
     public static final String CONNECTION_USERNAME = "ml.connection.username";
@@ -40,6 +41,7 @@ public class MarkLogicConfig extends AbstractConfig {
     public static final String CONNECTION_CERT_FILE = "ml.connection.certFile";
     public static final String CONNECTION_CERT_PASSWORD = "ml.connection.certPassword";
     public static final String CONNECTION_EXTERNAL_NAME = "ml.connection.externalName";
+    public static final String CONNECTION_CLOUD_API_KEY = "ml.connection.cloudApiKey";
     public static final String ENABLE_CUSTOM_SSL = "ml.connection.enableCustomSsl";
     public static final String TLS_VERSION = "ml.connection.customSsl.tlsVersion";
     public static final String SSL_HOST_VERIFIER = "ml.connection.customSsl.hostNameVerifier";
@@ -58,6 +60,9 @@ public static void addDefinitions(ConfigDef configDef) {
         .define(CONNECTION_PORT, Type.INT, ConfigDef.NO_DEFAULT_VALUE, ConfigDef.Range.atLeast(0), Importance.HIGH,
             "Required; the port of a REST API app server to connect to; if using Bulk Data Services, can be a plain HTTP app server",
             GROUP, -1, ConfigDef.Width.MEDIUM, "Port")
+        .define(CONNECTION_BASE_PATH, Type.STRING, null, Importance.MEDIUM,
+            "Base path for all calls to MarkLogic; typically used when a reverse proxy is in front of MarkLogic",
+            GROUP, -1, ConfigDef.Width.MEDIUM, "Base Path")
         .define(CONNECTION_SECURITY_CONTEXT_TYPE, Type.STRING, "DIGEST", CONNECTION_SECURITY_CONTEXT_TYPE_RV, Importance.HIGH,
             "Required; the authentication scheme used by the server defined by ml.connection.port; either 'DIGEST', 'BASIC', 'CERTIFICATE', 'KERBEROS', or 'NONE'",
             GROUP, -1, ConfigDef.Width.MEDIUM, "Security Context Type", CONNECTION_SECURITY_CONTEXT_TYPE_RV)
@@ -79,6 +84,9 @@ public static void addDefinitions(ConfigDef configDef) {
         .define(CONNECTION_EXTERNAL_NAME, Type.STRING, null, Importance.MEDIUM,
             "External name for 'KERBEROS' authentication",
             GROUP, -1, ConfigDef.Width.MEDIUM, "Kerberos External Name")
+        .define(CONNECTION_CLOUD_API_KEY, Type.STRING, null, Importance.MEDIUM,
+            "API key for connecting to MarkLogic Cloud. Should set port to 443 when connecting to MarkLogic Cloud.",
+            GROUP, -1, ConfigDef.Width.MEDIUM, "Cloud API Key")
         .define(CONNECTION_TYPE, Type.STRING, "", CONNECTION_TYPE_RV, Importance.MEDIUM,
             "Set to 'GATEWAY' when the host identified by ml.connection.host is a load balancer. See https://docs.marklogic.com/guide/java/data-movement#id_26583 for more information.",
             GROUP, -1, ConfigDef.Width.MEDIUM, "Connection Type", CONNECTION_TYPE_RV)

src/test/java/com/marklogic/kafka/connect/BuildDatabaseClientConfigTest.java

@@ -83,6 +83,18 @@ void kerberosAuthentication() {
         assertEquals("some-name", clientConfig.getExternalName());
     }
 
+    @Test
+    void cloudAuthentication() {
+        config.put(MarkLogicSinkConfig.CONNECTION_SECURITY_CONTEXT_TYPE, "cloud");
+        config.put(MarkLogicSinkConfig.CONNECTION_CLOUD_API_KEY, "my-key");
+        config.put(MarkLogicSinkConfig.CONNECTION_BASE_PATH, "/my/path");
+
+        DatabaseClientConfig clientConfig = builder.buildDatabaseClientConfig(config);
+        assertEquals(SecurityContextType.CLOUD, clientConfig.getSecurityContextType());
+        assertEquals("my-key", clientConfig.getCloudApiKey());
+        assertEquals("/my/path", clientConfig.getBasePath());
+    }
+
     @Test
     void digestAuthenticationAndSimpleSsl() {
         config.put(MarkLogicSinkConfig.CONNECTION_SECURITY_CONTEXT_TYPE, "digest");