Merge pull request #64 from marklogic-community/feature/58-password

Changing password fields to be of type PASSWORD

Author: rjrudin

CONTRIBUTING.md

@@ -105,18 +105,31 @@ You can also manually configure an instance of the MarkLogic Kafka connector as
 In the list of connectors in Control Center, the connector will initially have a status of "Failed" while it starts up.
 After it starts successfully, it will have a status of "Running". 
 
+## Debugging the MarkLogic Kafka connector
+
+The main mechanism for debugging an instance of the MarkLogic Kafka connector is by examining logs from the 
+connector. You can access those, along with logging from Kafka Connect and all other connectors, by running the 
+following:
+
+    confluent local services connect log -f
+
+See [the log command docs](https://docs.confluent.io/confluent-cli/current/command-reference/local/services/connect/confluent_local_services_connect_log.html)
+for more information.
+
 ## Destroying and setting up the Confluent Platform instance
 
 While developing and testing the MarkLogic Kafka connector, it is common that the "local" instance of Confluent 
 Platform will become unstable and no longer work. The [Confluent local docs](https://docs.confluent.io/confluent-cli/current/command-reference/local/confluent_local_current.html) 
 make reference to this - "The data that are produced are transient and are intended to be temporary". 
 
 It is thus advisable that after you copy a new instance of the MarkLogic Kafka connector into Confluent Platform (i.e. 
-by running `./gradlew copyConnectorToConfluent`), you should destroy your local Confluent Platform instance:
+by running `./gradlew copyConnectorToConfluent`), you should destroy your local Confluent Platform instance (this 
+will usually finish in around 15s):
 
     ./gradlew destroyLocalConfluent
 
-After doing that, you can quickly automate starting Confluent Platform and loading the two connectors via the following:
+After doing that, you can quickly automate starting Confluent Platform and loading the two connectors via the 
+following (this will usually finish in around 1m):
 
     ./gradlew setupLocalConfluent
 
@@ -125,6 +138,11 @@ Remember that if you've modified the connector code, you'll first need to run `.
 Doing the above will provide the most reliable way to get a new and working instance of Confluent Platform with the 
 MarkLogic Kafka connector installed.
 
+For brevity, you may prefer this (Gradle will figure out the tasks as long as only one task starts with "destroy" 
+and one task starts with "setup"):
+
+    ./gradlew destroy setup
+
 You may have luck with simply doing `confluent local services stop`, `./gradlew copyConnectorToConfluent`, and 
 `confluent local services start`, but this has so far not worked reliably - i.e. one of the Confluent Platform 
 services (sometimes Schema Registry, sometimes Control Center) usually stops working. 

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

@@ -4,8 +4,10 @@
 import com.marklogic.client.DatabaseClientFactory;
 import com.marklogic.client.ext.DatabaseClientConfig;
 import com.marklogic.client.ext.SecurityContextType;
+import com.marklogic.client.ext.helper.LoggingObject;
 import com.marklogic.client.ext.modulesloader.ssl.SimpleX509TrustManager;
 import com.marklogic.kafka.connect.sink.MarkLogicSinkConfig;
+import org.apache.kafka.common.config.types.Password;
 
 import javax.net.ssl.*;
 import java.io.FileInputStream;
@@ -16,13 +18,16 @@
 import java.security.NoSuchAlgorithmException;
 import java.util.Map;
 
-public class DefaultDatabaseClientConfigBuilder implements DatabaseClientConfigBuilder {
+public class DefaultDatabaseClientConfigBuilder extends LoggingObject implements DatabaseClientConfigBuilder {
 
     @Override
     public DatabaseClientConfig buildDatabaseClientConfig(Map<String, Object> parsedConfig) {
         DatabaseClientConfig clientConfig = new DatabaseClientConfig();
         clientConfig.setCertFile((String) parsedConfig.get(MarkLogicSinkConfig.CONNECTION_CERT_FILE));
-        clientConfig.setCertPassword((String) parsedConfig.get(MarkLogicSinkConfig.CONNECTION_CERT_PASSWORD));
+        Password certPassword = (Password) parsedConfig.get(MarkLogicSinkConfig.CONNECTION_CERT_PASSWORD);
+        if (certPassword != null) {
+            clientConfig.setCertPassword(certPassword.value());
+        }
         clientConfig.setTrustManager(new SimpleX509TrustManager());
         clientConfig = configureHostNameVerifier(clientConfig, parsedConfig);
         String securityContextType = ((String) parsedConfig.get(MarkLogicSinkConfig.CONNECTION_SECURITY_CONTEXT_TYPE)).toUpperCase();
@@ -37,7 +42,10 @@ public DatabaseClientConfig buildDatabaseClientConfig(Map<String, Object> parsed
         }
         clientConfig.setExternalName((String) parsedConfig.get(MarkLogicSinkConfig.CONNECTION_EXTERNAL_NAME));
         clientConfig.setHost((String) parsedConfig.get(MarkLogicSinkConfig.CONNECTION_HOST));
-        clientConfig.setPassword((String) parsedConfig.get(MarkLogicSinkConfig.CONNECTION_PASSWORD));
+        Password password = (Password) parsedConfig.get(MarkLogicSinkConfig.CONNECTION_PASSWORD);
+        if (password != null) {
+            clientConfig.setPassword(password.value());
+        }
         clientConfig.setPort((Integer) parsedConfig.get(MarkLogicSinkConfig.CONNECTION_PORT));
         Boolean customSsl = (Boolean) parsedConfig.get(MarkLogicSinkConfig.SSL);
         if (customSsl != null && customSsl) {

src/main/java/com/marklogic/kafka/connect/sink/MarkLogicSinkConfig.java

@@ -58,12 +58,13 @@ public class MarkLogicSinkConfig extends AbstractConfig {
         .define(CONNECTION_PORT, Type.INT, Importance.HIGH, "The REST app server port to connect to")
         .define(CONNECTION_DATABASE, Type.STRING, "", Importance.LOW, "Database to connect, if different from the one associated with the port")
         .define(CONNECTION_SECURITY_CONTEXT_TYPE, Type.STRING, "NONE", Importance.HIGH, "Type of MarkLogic security context to create - either digest, basic, kerberos, certificate, or none")
-        .define(CONNECTION_USERNAME, Type.STRING, Importance.HIGH, "Name of MarkLogic user to authenticate as")
-        .define(CONNECTION_PASSWORD, Type.STRING, Importance.HIGH, "Password for the MarkLogic user")
+        .define(CONNECTION_USERNAME, Type.STRING, null, Importance.HIGH, "Name of MarkLogic user to " +
+            "authenticate as")
+        .define(CONNECTION_PASSWORD, Type.PASSWORD, null, Importance.HIGH, "Password for the MarkLogic user")
         .define(CONNECTION_TYPE, Type.STRING, "DIRECT", Importance.LOW, "Connection type; DIRECT or GATEWAY")
         .define(CONNECTION_SIMPLE_SSL, Type.BOOLEAN, false, Importance.LOW, "Set to true to use a trust-everything SSL connection")
         .define(CONNECTION_CERT_FILE, Type.STRING, "", Importance.LOW, "Path to a certificate file")
-        .define(CONNECTION_CERT_PASSWORD, Type.STRING, "", Importance.LOW, "Password for the certificate file")
+        .define(CONNECTION_CERT_PASSWORD, Type.PASSWORD, null, Importance.LOW, "Password for the certificate file")
         .define(CONNECTION_EXTERNAL_NAME, Type.STRING, "", Importance.LOW, "External name for Kerberos authentication")
         .define(DATAHUB_FLOW_NAME, Type.STRING, null, Importance.MEDIUM, "Name of a Data Hub flow to run")
         .define(DATAHUB_FLOW_STEPS, Type.STRING, null, Importance.MEDIUM, "Comma-delimited names of steps to run")

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

@@ -6,6 +6,7 @@
 import com.marklogic.client.ext.SecurityContextType;
 import com.marklogic.kafka.connect.sink.MarkLogicSinkConfig;
 import org.apache.kafka.common.config.ConfigException;
+import org.apache.kafka.common.config.types.Password;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -34,7 +35,7 @@ public void setup() {
     public void basicAuthentication() {
         config.put(MarkLogicSinkConfig.CONNECTION_SECURITY_CONTEXT_TYPE, "basic");
         config.put(MarkLogicSinkConfig.CONNECTION_USERNAME, "some-user");
-        config.put(MarkLogicSinkConfig.CONNECTION_PASSWORD, "some-password");
+        config.put(MarkLogicSinkConfig.CONNECTION_PASSWORD, new Password("some-password"));
 
         DatabaseClientConfig clientConfig = builder.buildDatabaseClientConfig(config);
         assertEquals("some-host", clientConfig.getHost());
@@ -49,7 +50,7 @@ public void basicAuthentication() {
     public void certificateAuthentication() {
         config.put(MarkLogicSinkConfig.CONNECTION_SECURITY_CONTEXT_TYPE, "certificate");
         config.put(MarkLogicSinkConfig.CONNECTION_CERT_FILE, "/path/to/file");
-        config.put(MarkLogicSinkConfig.CONNECTION_CERT_PASSWORD, "cert-password");
+        config.put(MarkLogicSinkConfig.CONNECTION_CERT_PASSWORD, new Password("cert-password"));
 
         DatabaseClientConfig clientConfig = builder.buildDatabaseClientConfig(config);
         assertEquals(SecurityContextType.CERTIFICATE, clientConfig.getSecurityContextType());
@@ -102,7 +103,7 @@ public void basicAuthenticationAndMutualSSL() {
         config.put(MarkLogicSinkConfig.SSL_HOST_VERIFIER, "STRICT");
         config.put(MarkLogicSinkConfig.SSL_MUTUAL_AUTH, "true");
         config.put(MarkLogicSinkConfig.CONNECTION_CERT_FILE, absolutePath);
-        config.put(MarkLogicSinkConfig.CONNECTION_CERT_PASSWORD, "abc");
+        config.put(MarkLogicSinkConfig.CONNECTION_CERT_PASSWORD, new Password("abc"));
 
         DatabaseClientConfig clientConfig = builder.buildDatabaseClientConfig(config);
         assertEquals(SecurityContextType.BASIC, clientConfig.getSecurityContextType());
@@ -122,7 +123,7 @@ public void basicAuthenticationAndMutualSSLWithInvalidHost() {
         config.put(MarkLogicSinkConfig.SSL_HOST_VERIFIER, "SOMETHING");
         config.put(MarkLogicSinkConfig.SSL_MUTUAL_AUTH, "true");
         config.put(MarkLogicSinkConfig.CONNECTION_CERT_FILE, absolutePath);
-        config.put(MarkLogicSinkConfig.CONNECTION_CERT_PASSWORD, "abc");
+        config.put(MarkLogicSinkConfig.CONNECTION_CERT_PASSWORD, new Password("abc"));
 
         DatabaseClientConfig clientConfig = builder.buildDatabaseClientConfig(config);
         assertEquals(SecurityContextType.BASIC, clientConfig.getSecurityContextType());
@@ -165,8 +166,6 @@ public void testMissingRequired() {
         Map<String, Object> allRequiredValuesConfig = new HashMap<>();
         allRequiredValuesConfig.put(MarkLogicSinkConfig.CONNECTION_HOST, "");
         allRequiredValuesConfig.put(MarkLogicSinkConfig.CONNECTION_PORT, 8000);
-        allRequiredValuesConfig.put(MarkLogicSinkConfig.CONNECTION_USERNAME, "");
-        allRequiredValuesConfig.put(MarkLogicSinkConfig.CONNECTION_PASSWORD, "");
         MarkLogicSinkConfig.CONFIG_DEF.parse(allRequiredValuesConfig);
 
         Set<String> keys = allRequiredValuesConfig.keySet();

src/test/resources/confluent/marklogic-purchases-connector.json

@@ -12,6 +12,8 @@
     "ml.connection.password": "admin",
     "ml.connection.securityContextType": "DIGEST",
     "ml.document.format": "JSON",
-    "ml.document.uriSuffix": ".json"
+    "ml.document.uriPrefix": "/purchase/",
+    "ml.document.uriSuffix": ".json",
+    "ml.document.collections": "purchases,kafka-data"
   }
-}
+}