Added the ml.dmsdk.transform and ml.dmsdk.transformParams properties because they must be declared.

Added an example using a server-side transform from the connector.
Added a basic SJS transform and a shell command for installing the transform.

Author: BillFarber

config/marklogic-sink.properties

@@ -68,3 +68,6 @@ ml.document.uriPrefix=/kafka-data/
 
 # Optional - a suffix to append to each URI
 ml.document.uriSuffix=.json
+
+ml.dmsdk.transform=
+ml.dmsdk.transformParams=

examples/dmsdk-transform/installTransform.sh

@@ -0,0 +1 @@
+curl --anyauth --user admin:admin -X PUT -i --data-binary @"./trans-ex.sjs" -H "Content-type: application/vnd.marklogic-javascript" 'http://ec2-54-173-231-36.compute-1.amazonaws.com:8000/LATEST/config/transforms/js-example'

examples/dmsdk-transform/marklogic-sink-with-transform.properties

@@ -0,0 +1,74 @@
+# Kafka-specific properties
+
+name=marklogic-sink
+connector.class=com.marklogic.kafka.connect.sink.MarkLogicSinkConnector
+
+# Should only need one task since it's using a WriteBatcher, which is multi-threaded
+tasks.max=1
+
+# Topics to consume from [comma separated list for multiple topics]
+topics=marklogic
+
+
+# MarkLogic connector-specific properties
+
+# A MarkLogic host to connect to. The connector uses the Data Movement SDK, and thus it will connect to each of the
+# hosts in a cluster.
+ml.connection.host=localhost
+
+# The port of a REST API server to connect to.
+ml.connection.port=8000
+
+# Optional - the name of a database to connect to. If your REST API server has a content database matching that of the
+# one that you want to write documents to, you do not need to set this.
+ml.connection.database=Documents
+
+# 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
+ml.connection.securityContextType=DIGEST
+
+# Set these based on the security context type defined above
+ml.connection.username=admin
+ml.connection.password=admin
+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.
+ml.connection.simpleSsl=false
+
+# 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
+# collection of records. Thus, if the connector never receives at one time more than the value of this property, then
+# the value of this property will have no impact.
+ml.dmsdk.batchSize=100
+
+# Sets the number of threads used by the Data Movement SDK for parallelizing writes to MarkLogic. Similar to the batch
+# size property above, this may never come into play depending on how many records the connector receives at once.
+ml.dmsdk.threadCount=8
+
+# Optional - a comma-separated list of collections that each document should be written to
+ml.document.collections=kafka-data
+
+# Optional - specify the format of each document; either JSON, XML, BINARY, TEXT, or UNKNOWN
+ml.document.format=JSON
+
+# Optional - specify a mime type for each document; typically the format property above will be used instead of this
+ml.document.mimeType=
+
+# Optional - a comma-separated list of roles and capabilities that define the permissions for each document written to MarkLogic
+ml.document.permissions=rest-reader,read,rest-writer,update
+
+# Optional - a prefix to prepend to each URI; the URI itself is a UUID
+ml.document.uriPrefix=/kafka-data/
+
+# Optional - a suffix to append to each URI
+ml.document.uriSuffix=.json
+
+# Optional - specify a MarkLogic server-side transform and required parameters
+ml.dmsdk.transform=js-example
+ml.dmsdk.transformParams=

examples/dmsdk-transform/trans-ex.sjs

@@ -0,0 +1,17 @@
+function insertTimestamp(context, params, content)
+{
+  if (context.inputType.search('json') >= 0) {
+    var result = content.toObject();
+    if (context.acceptTypes) {                 /* read */
+      result.readTimestamp = fn.currentDateTime();
+    } else {                                   /* write */
+      result.writeTimestamp = fn.currentDateTime();
+    }
+    return result;
+  } else {
+    /* Pass thru for non-JSON documents */
+    return content;
+  }
+};
+
+exports.transform = insertTimestamp;