Develop (#43)

* Feature/enhanced logging (#39)

* added options for logging incoming record keys and headers
* added simple batch write failure listener
* cleaned up logging messages for headers, and added hashed log messages.

* Updates to the AWS CloudFormation Quickstart. Also, added documentation for using the connector with AWS MSK (from Biju). (#42)

Co-authored-by: Anthony Clavio <[email protected]>

Author: BillFarber

AWS-CloudFormation/CloudFormation-QuickStart.md

@@ -4,7 +4,7 @@ The three servers are the Kafka/Zookeeper server, the MarkLogic server, and a se
 
 _This is not intended to be a description of setting up a production environment._
 
-_All the resource files have been configure and compiled specifically for this example.
+_All the resource files have been configured and compiled specifically for this example.
 In particular, IP addresses are in the config files._
 
 ## Requirements
@@ -31,40 +31,53 @@ Please note: if you are using the default values in the resource files, you will
 
 ## Build the AWS Resources
 1. Navigate to the [AWS CloudFormation](https://console.aws.amazon.com/cloudformation/home?region=us-east-1) page.
-1. Click "Create Stack"
+1. Click "Create Stack" and "With new resources (standard)"
 1. Click the "Upload a template file" radio button
 1. Click "Upload File" and upload "readyToGo.json"
 1. Click "Next"
 1. Give your stack a name, and click "Next"
 1. Click "Next"
 1. Click "Create Stack"
 1. Go to the list of your [AWS Instances](https://console.aws.amazon.com/ec2/v2/home?region=us-east-1#Instances:sort=tag:Name)
-1. Now wait until initialization is complete.
+1. <strong>Now wait until initialization is complete.<strong> (Instance Status Check says "2/2 checks passed")
+
+## View the empty Kafka database in MarkLogic
+
+1. On the list of your AWS instances, click on the instance named "instanceMarkLogicA".
+2. Copy the Public DNS (IPv4)
+3. In a browser, open the page http://\<public DNS>:8000
+   * Enter the username "admin"
+   * Enter the password "admin"
+   * (If this page does not open for you, wait 5 minutes to allow server initialization to complete)
+4. In the QConsole UI, change the *Database* to "Kafka"
+5. Click *Explore* to verify that the database has no documents.
 
 ## Start the Connector
-1. On the list of your AWS Instances, click on the instance named, "instanceKafkaWorkerA".
+1. On the list of your AWS Instances, click on the instance named, "TemplateBased-Kafka-Worker-A".
 1. Copy the Public DNS (IPv4).
 1. Ssh to the TemplateBased-Kafka-Worker-A server.
-`ssh -i kafka.pem bitnami@<Public DNS>`
+```
+ssh -i kafka.pem bitnami@<Public DNS>
+```
 1. Start the connector
-`sudo /opt/bitnami/kafka/bin/connect-standalone.sh /opt/bitnami/kafka/config/marklogic-connect-standalone.properties /opt/bitnami/kafka/config/marklogic-sink.properties`
+```
+sudo /opt/bitnami/kafka/bin/connect-standalone.sh /opt/bitnami/kafka/config/marklogic-connect-standalone.properties /opt/bitnami/kafka/config/marklogic-sink.properties
+```
 
 ## Generate some Messages
 This step uses the JAR file from a small project for producing test messages. It can be found in [my GitHub account](https://github.com/BillFarber/KafkaProducer)
-1. On the list of your AWS Instances, click on the instance named, "instanceKafkaWorkerA".
+1. On the list of your AWS Instances, click on the instance named, "TemplateBased-Kafka-Worker-A".
 1. Copy the Public DNS (IPv4).
 1. Ssh to the TemplateBased-Kafka-Worker-A server.
-`ssh -i kafka.pem bitnami@<Public DNS>`
+```
+ssh -i kafka.pem bitnami@<Public DNS>
+```
 1. Send some messages to the Kafka topic
-`java -jar /home/bitnami/kafka-producer-1.0-SNAPSHOT.jar -c 4 -m 5 -h ip-172-31-48-44.ec2.internal:9092 -t marklogic`
+```
+java -jar /home/bitnami/kafka-producer-1.0-SNAPSHOT.jar -c 4 -m 5 -h ip-172-31-48-44.ec2.internal:9092 -t marklogic
+```
 
 ## View the Messages in MarkLogic
 
-1. On the list of your AWS instances, click on the instance named "instanceMarkLogicA".
-2. Copy the Public DNS (IPv4)
-3. In a browser, open the page http://\<public DNS>:8000
-   * Enter the username "admin"
-   * Enter the password "admin"
-
-4. In the QConsole UI, change the *Database* to "Kafka"
-5. Click *Explore* to view the documents you created from "Kafka-Worker-A"
+1. Go back to the MarkLogic QConsole web page (see previous instructions)
+1. Click *Explore* to view the documents you created from "Kafka-Worker-A"

AWS-CloudFormation/cloudFormationTemplates/readyToGo.json

@@ -111,7 +111,7 @@
           "sudo wget https://cf-templates-3tc878fhthic-us-east-1.s3.amazonaws.com/Kafka-Initialization/kafka-producer-1.0-SNAPSHOT.jar -O /home/bitnami/kafka-producer-1.0-SNAPSHOT.jar\n",
           "sudo chmod 644 /home/bitnami/kafka-producer-1.0-SNAPSHOT.jar\n",
           "sudo mkdir -p /home/bitnami/kafka/config\n",
-          "sudo wget https://cf-templates-3tc878fhthic-us-east-1.s3.amazonaws.com/Kafka-Initialization/kafka-connect-marklogic-0.9.0.jar -O /opt/bitnami/kafka/libs/kafka-connect-marklogic-0.9.0.jar\n",
+          "sudo wget https://cf-templates-3tc878fhthic-us-east-1.s3.amazonaws.com/Kafka-Initialization/kafka-connect-marklogic-1.4.0.jar -O /opt/bitnami/kafka/libs/kafka-connect-marklogic-1.4.0.jar\n",
           "sudo wget https://cf-templates-3tc878fhthic-us-east-1.s3.amazonaws.com/Kafka-Initialization/marklogic-connect-standalone.properties -O /opt/bitnami/kafka/config/marklogic-connect-standalone.properties\n",
           "sudo wget https://cf-templates-3tc878fhthic-us-east-1.s3.amazonaws.com/Kafka-Initialization/marklogic-sink.properties -O /opt/bitnami/kafka/config/marklogic-sink.properties\n"
         ]]}}
@@ -160,6 +160,7 @@
         ],
         "UserData"       : { "Fn::Base64" : { "Fn::Join" : ["", [
           "#!/bin/bash -xe\n",
+          "sleep 180s\n",
           "sudo apt-get update && sudo apt-get -y upgrade\n",
           "sudo apt -y install sysstat\n",
           "sudo wget https://cf-templates-3tc878fhthic-us-east-1.s3.amazonaws.com/Kafka-Initialization/server.properties -O /opt/bitnami/kafka/config/server.properties\n",
@@ -221,16 +222,18 @@
         ],
         "UserData"       : { "Fn::Base64" : { "Fn::Join" : ["", [
           "#!/bin/bash -xe\n",
+          "sleep 180s\n",
           "sudo yum -y update\n",
           "sudo echo \"MARKLOGIC_EC2_HOST=0\" > /etc/marklogic.conf\n",
           "sudo service MarkLogic start\n",
           "sudo yum install sysstat\n",
+          "sleep 60s\n",
           "curl -X POST -d \"\" http://localhost:8001/admin/v1/init\n",
-          "sleep 30s\n",
+          "sleep 60s\n",
           "curl -i -X POST -d \"admin-username=admin&admin-password=admin&realm=public\" http://localhost:8001/admin/v1/instance-admin\n",
-          "sleep 30s\n",
-          "sudo echo \"<rest-api xmlns=\\\"http://marklogic.com/rest-api\\\"><name>Kafka</name><database>Kafka</database><port>8003</port></rest-api>\" > /etc/kafkaServer.conf\n",
-          "curl --anyauth --user admin:admin -X POST -d@'/etc/kafkaServer.conf' -i -H \"Content-type: application/xml\" http://localhost:8002/v1/rest-apis\n"
+          "sleep 60s\n",
+          "sudo echo \"<rest-api xmlns=\\\"http://marklogic.com/rest-api\\\"><name>Kafka</name><database>Kafka</database><port>8003</port></rest-api>\" > /tmp/kafkaServer.conf\n",
+          "curl --anyauth --user admin:admin -X POST -d@'/tmp/kafkaServer.conf' -i -H \"Content-type: application/xml\" http://localhost:8002/v1/rest-apis\n"
         ]]}}
       },
       "Metadata": {
@@ -672,4 +675,4 @@
       }
     }
   }
-}
+}

AWS-CloudFormation/s3Resources/kafka-connect-marklogic-0.9.0.jar

@@ -15,16 +15,39 @@
 
 # These are defaults. This file just demonstrates how to override some settings.
 bootstrap.servers=172.31.48.44:9092
+# (Optional) The group id provides a way to logically group connectors to distribute the load across multiple instances.
 group.id=marklogic
 
-# The converters specify the format of data in Kafka and how to translate it into Connect data. Every Connect user will
-# need to configure these based on the format they want their data in when loaded from or stored into Kafka
+# The next two sections are necessary to establish an SSL connection to the Kafka servers.
+# To enable SSL, uncomment and customize the lines that start with "#* " in those two sections.
+
+# SSL connection properties
+# For more information, see https://docs.confluent.io/current/kafka/encryption.html#encryption-ssl-connect
+#   These top-level settings are used by the Connect worker for group coordination and to read and write to the internal
+#   topics which are used to track the cluster's state (e.g. configs and offsets).
+#* security.protocol=SSL
+# You must create a truststore that contains either the server certificate or a trusted CA that signed the server cert.
+# This is how I did it with keytools:
+#   keytool -keystore kafka.truststore.jks -alias caroot -import -file ca-cert -storepass "XXXXX" -keypass "XXXXX" -noprompt
+#* ssl.truststore.location=/secure/path/to/certs/kafka.client.truststore.jks
+#* ssl.truststore.password=truststorePassphrase
+# For now, turn off hostname verification since we're using self-signed certificates
+# This might also be fixable by fixing the "Subject Alternative Name (SAN)", but I'm not a cert expert.
+#* ssl.endpoint.identification.algorithm=
 
-# Set the converters to the JsonConverter for JSON
-#key.converter=org.apache.kafka.connect.json.JsonConverter
-#value.converter=org.apache.kafka.connect.json.JsonConverter
+# Yes, both of these sections are required.
+#   Connect workers manage the producers used by source connectors and the consumers used by sink connectors.
+#   So, for the connectors to leverage security, you also have to override the default producer/consumer
+#   configuration that the worker uses.
+#* consumer.bootstrap.servers=localhost:9093
+#* consumer.security.protocol=SSL
+#* consumer.ssl.truststore.location=/secure/path/to/certs/kafka.client.truststore.jks
+#* consumer.ssl.truststore.password=truststorePassphrase
+#* consumer.ssl.endpoint.identification.algorithm=
 
-# Set the converters to the StringConverter for XML
+
+# The converters specify the format of data in Kafka and how to translate it into Connect data. Every Connect user will
+# need to configure these based on the format they want their data in when loaded from or stored into Kafka
 key.converter=org.apache.kafka.connect.storage.StringConverter
 value.converter=org.apache.kafka.connect.storage.StringConverter
 

AWS-CloudFormation/s3Resources/kafka-connect-marklogic-1.4.0.jar

@@ -23,7 +23,8 @@ ml.connection.port=8003
 # one that you want to write documents to, you do not need to set this.
 ml.connection.database=Kafka
 
-# Optional - set to "gateway" when using a load balancer, else leave blank. See https://docs.marklogic.com/guide/java/data-movement#id_26583 for more information.
+# Optional - set to "gateway" when using a load balancer, else leave blank.
+# See https://docs.marklogic.com/guide/java/data-movement#id_26583 for more information.
 ml.connection.type=
 
 # Either DIGEST, BASIC, CERTIFICATE, KERBEROS, or NONE
@@ -36,10 +37,15 @@ ml.connection.certFile=
 ml.connection.certPassword=
 ml.connection.externalName=
 
-# Set to "true" for a "simple" SSL strategy that uses the JVM's default SslContext and X509TrustManager and a
-# "trust everything" HostnameVerifier. Further customization of an SSL connection via properties is not supported. If
-# you need to do so, consider using the source code for this connector as a starting point.
+# Set "ml.connection.simpleSsl" to "true" for a "simple" SSL strategy that uses the JVM's default SslContext and
+# X509TrustManager and a "trust everything" HostnameVerifier. Further customization of an SSL connection via properties
+# is not supported. If you need to do so, consider using the source code for this connector as a starting point.
 ml.connection.simpleSsl=false
+# You must also ensure that the server cert or the signing CA cert is imported in the JVMs cacerts file.
+# These commands may be used to get the server cert and to import it into your cacerts file.
+# Don't forget to customize the commands for your particular case.
+#   openssl x509 -in <(openssl s_client -connect <server>:8004 -prexit 2>/dev/null) -out ~/example.crt
+#   sudo keytool -importcert -file ~/example.crt -alias <server> -keystore /path/to/java/lib/security/cacerts -storepass <storepass-password>
 
 # Sets the number of documents to be written in a batch to MarkLogic. This may not have any impact depending on the
 # connector receives data from Kafka, as the connector calls flushAsync on the DMSDK WriteBatcher after processing every
@@ -54,6 +60,9 @@ ml.dmsdk.threadCount=8
 # Optional - a comma-separated list of collections that each document should be written to
 ml.document.collections=kafka-data
 
+# Optional - set this to true so that the name of the topic that the connector reads from is added as a collection to each document inserted by the connector
+ml.document.addTopicToCollections=false
+
 # Optional - specify the format of each document; either JSON, XML, BINARY, TEXT, or UNKNOWN
 ml.document.format=JSON
 
@@ -68,3 +77,30 @@ ml.document.uriPrefix=/kafka-data/
 
 # Optional - a suffix to append to each URI
 ml.document.uriSuffix=.json
+
+# Optional - name of a REST transform to use when writing documents
+# For Data Hub, can use mlRunIngest
+ml.dmsdk.transform=
+
+# Optional - delimited set of transform names and values
+# Data Hub example = flow-name,ingestion_mapping_mastering-flow,step,1
+ml.dmsdk.transformParams=
+
+# Optional - delimiter for transform parameter names and values
+ml.dmsdk.transformParamsDelimiter=,
+
+# Properties for running a Data Hub flow
+# Using examples/dh-5-example in the DH project, could use the following config:
+# ml.datahub.flow.name=ingestion_mapping_mastering-flow
+# ml.datahub.flow.steps=2,3,4
+ml.datahub.flow.name=
+ml.datahub.flow.steps=
+# Whether or not the response data from running a flow should be logged at the info level
+ml.datahub.flow.logResponse=true
+
+ml.id.strategy=
+ml.id.strategy.paths=
+ml.connection.enableCustomSsl=false
+ml.connection.customSsl.tlsVersion=
+ml.connection.customSsl.hostNameVerifier=
+ml.connection.customSsl.mutualAuth=false

AWS-CloudFormation/s3Resources/marklogic-connect-standalone.properties

@@ -97,3 +97,7 @@ ml.datahub.flow.name=
 ml.datahub.flow.steps=
 # Whether or not the response data from running a flow should be logged at the info level
 ml.datahub.flow.logResponse=true
+
+# Logging configurations
+ml.log.record.key=false
+ml.log.record.headers=false

AWS-CloudFormation/s3Resources/marklogic-sink.properties

@@ -46,7 +46,11 @@ public class MarkLogicSinkConfig extends AbstractConfig {
 	public static final String TLS_VERSION = "ml.connection.customSsl.tlsVersion";
 	public static final String SSL_HOST_VERIFIER = "ml.connection.customSsl.hostNameVerifier";
 	public static final String SSL_MUTUAL_AUTH = "ml.connection.customSsl.mutualAuth";
-	public static final String ID_STRATEGY = "ml.id.strategy";
+
+	public static final String LOGGING_RECORD_KEY = "ml.log.record.key";
+	public static final String LOGGING_RECORD_HEADERS = "ml.log.record.headers";
+
+  public static final String ID_STRATEGY = "ml.id.strategy";
 	public static final String ID_STRATEGY_PATH = "ml.id.strategy.paths";
 
 	public static ConfigDef CONFIG_DEF = new ConfigDef()
@@ -80,7 +84,11 @@ public class MarkLogicSinkConfig extends AbstractConfig {
 		.define(TLS_VERSION, Type.STRING, Importance.LOW, "Version of TLS to connect to MarkLogic SSL enabled App server. Ex. TLSv1.2")
 		.define(SSL_HOST_VERIFIER, Type.STRING, Importance.LOW, "The strictness of Host Verifier - ANY, COMMON, STRICT")
 		.define(SSL_MUTUAL_AUTH, Type.BOOLEAN, Importance.LOW, "Mutual Authentication for Basic or Digest : true or false")
-		.define(ID_STRATEGY, Type.STRING, Importance.LOW, "The ID Strategy for URI.")
+
+    .define(LOGGING_RECORD_KEY, Type.BOOLEAN, false, Importance.LOW, "Log incoming record keys")
+		.define(LOGGING_RECORD_HEADERS, Type.BOOLEAN, false, Importance.LOW, "Log incoming record headers")
+
+    .define(ID_STRATEGY, Type.STRING, Importance.LOW, "The ID Strategy for URI.")
 		.define(ID_STRATEGY_PATH, Type.STRING, Importance.LOW, "The JSON path for ID Strategy")
 		;
 

config/marklogic-sink.properties

@@ -6,16 +6,16 @@
 import com.marklogic.client.document.ServerTransform;
 import com.marklogic.client.ext.DatabaseClientConfig;
 import com.marklogic.client.ext.DefaultConfiguredDatabaseClientFactory;
+import com.marklogic.client.io.DocumentMetadataHandle;
+import com.marklogic.client.io.Format;
+import com.marklogic.client.io.StringHandle;
 import com.marklogic.kafka.connect.DefaultDatabaseClientConfigBuilder;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.apache.kafka.connect.sink.SinkTask;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Arrays;
-import java.util.Collection;
-import java.util.List;
-import java.util.Map;
+import java.util.*;
 import java.io.IOException;
 
 /**
@@ -30,11 +30,16 @@ public class MarkLogicSinkTask extends SinkTask {
 	private DataMovementManager dataMovementManager;
 	private WriteBatcher writeBatcher;
 	private SinkRecordConverter sinkRecordConverter;
+	private boolean logKeys = false;
+	private boolean logHeaders = false;
 
 	@Override
 	public void start(final Map<String, String> config) {
 		logger.info("Starting");
 
+		logKeys = Boolean.parseBoolean(config.get(MarkLogicSinkConfig.LOGGING_RECORD_KEY));
+		logHeaders = Boolean.parseBoolean(config.get(MarkLogicSinkConfig.LOGGING_RECORD_HEADERS));
+
 		sinkRecordConverter = new DefaultSinkRecordConverter(config);
 
 		DatabaseClientConfig databaseClientConfig = new DefaultDatabaseClientConfigBuilder().buildDatabaseClientConfig(config);
@@ -50,6 +55,12 @@ public void start(final Map<String, String> config) {
 			writeBatcher.withTransform(transform);
 		}
 
+		writeBatcher.onBatchFailure((writeBatch, throwable) -> {
+			int batchSize = writeBatch.getItems().length;
+			logger.error("#error failed to write {} records", batchSize);
+			logger.error("#error batch failure:", throwable);
+		});
+
 		final String flowName = config.get(MarkLogicSinkConfig.DATAHUB_FLOW_NAME);
 		if (flowName != null && flowName.trim().length() > 0) {
 			writeBatcher.onBatchSuccess(buildSuccessListener(flowName, config, databaseClientConfig));
@@ -147,10 +158,23 @@ public void put(final Collection<SinkRecord> records) {
 			return;
 		}
 
+		final List<String> headers = new ArrayList<>();
+
 		records.forEach(record -> {
+			
 			if (record == null) {
 				logger.warn("Skipping null record object.");
 			} else {
+				if (logKeys) {
+					logger.info("#record key {}", record.key());
+				}
+				if (logHeaders) {
+					headers.clear();
+					record.headers().forEach(header -> {
+						headers.add(String.format("%s:%s", header.key(), header.value().toString()));
+					});
+					logger.info("#record headers: {}", headers);
+				}
 				if (logger.isDebugEnabled()) {
 					logger.debug("Processing record value {} in topic {}", record.value(), record.topic());
 				}