Merge pull request #154 from marklogic/feature/dhf-confluent

DHF integration in sink connector now works with Confluent

Author: rjrudin

CONTRIBUTING.md

@@ -80,10 +80,11 @@ does not already exist) to point to where you extracted the Confluent Platform d
 
 Then build and copy the connector to the Confluent Platform directory that you configured above:
 
-    ./gradlew copyConnectorToConfluent
+    ./gradlew clean copyConnectorToConfluent
 
 Note that any time you modify the MarkLogic Kafka connector code, you'll need to repeat the 
-`./gradlew copyConnectorToConfluent` step.
+`./gradlew clean copyConnectorToConfluent` step. Note that `clean` is included to ensure that in case you've changed 
+any connector dependencies, old dependencies will not be included in the connector archive.
 
 Next, start Confluent:
 

README.md

@@ -560,11 +560,6 @@ ingestion step depends on reading data from a filesystem. You can however run an
 transform as described above. Please see the DHF documentation for information on how to configure the DHF REST transform
 for running an ingestion step. 
 
-**Warning** - as of the MarkLogic Kafka 1.7.0 release, this feature may not work on certain versions of Confluent 
-Platform due to dependency conflicts. Testing has verified that this feature works on the latest Apache Kafka 2.8.x
-release and the latest Apache Kafka 3.3.x release. 
-
-
 ### Writing data via custom code (Bulk Data Services)
 
 MarkLogic's [Bulk Data Services](https://github.com/marklogic/java-client-api/wiki/Bulk-Data-Services) feature is 

build.gradle

@@ -66,15 +66,14 @@ dependencies {
     exclude module: "logback-classic"
   }
 
-  // 1.2.1 will be published soon, and then we'll switch to that
-  testImplementation 'com.marklogic:marklogic-junit5:1.2.1'
+  testImplementation 'com.marklogic:marklogic-junit5:1.3.0'
 
   testImplementation "org.apache.kafka:connect-api:${kafkaVersion}"
   testImplementation "org.apache.kafka:connect-json:${kafkaVersion}"
   testImplementation 'net.mguenther.kafka:kafka-junit:3.2.2'
 
   // Forcing logback to be used for test logging
-  testImplementation "ch.qos.logback:logback-classic:1.2.11"
+  testImplementation "ch.qos.logback:logback-classic:1.3.5"
   testImplementation "org.slf4j:jcl-over-slf4j:1.7.36"
 
   documentation files('LICENSE.txt')
@@ -184,7 +183,17 @@ task connectorArchive_CopyDocumentationToBuildDirectory(type: Copy, group: confl
 task connectorArchive_CopyDependenciesToBuildDirectory(type: Copy, group: confluentArchiveGroup, dependsOn: jar) {
   description = "Copy the dependency jars into the lib folder"
   from jar
-  from configurations.runtimeClasspath.findAll { it.name.endsWith('jar') }
+  // Confluent already includes the Jackson dependencies that this connector depends on. If the connector includes any
+  // itself, and the DHF integration is used with the sink connector, then the following error will occur when DHF
+  // tries to connect to the Manage API of MarkLogic:
+  // java.lang.ClassCastException: com.fasterxml.jackson.datatype.jdk8.Jdk8Module cannot be cast to com.fasterxml.jackson.databind.Module
+  //	at org.springframework.http.converter.json.Jackson2ObjectMapperBuilder.registerWellKnownModulesIfAvailable(Jackson2ObjectMapperBuilder.java:849)
+  // stackoverflow indicates this may be due to multiple copies of Jackson being on the classpath, as Jdk8Module
+  // otherwise should be castable to Module.
+  // Testing has verified that excluding all "jackson-" jars still results in the connector working properly with
+  // Confluent 7.3.1. This has no impact on using the connector with plain Apache Kafka which does not rely on
+  // constructing this connector archive.
+  from configurations.runtimeClasspath.findAll { it.name.endsWith('jar') && !it.name.startsWith("jackson-")}
   into "${baseArchiveBuildDir}/${baseArchiveName}/lib"
 }
 

config/marklogic-sink.properties

@@ -13,7 +13,7 @@ tasks.max=1
 topics=marklogic
 
 
-# Properties defined by the MarkLogic Kafka connector
+# Properties defined by the MarkLogic Kafka connector; change these as needed to fit your environment.
 
 # Required; a MarkLogic host to connect to. By default, the connector uses the Data Movement SDK, and thus it will
 # connect to each of the hosts in a cluster.

config/marklogic-source.properties

@@ -7,23 +7,23 @@ name=marklogic-source
 connector.class=com.marklogic.kafka.connect.source.MarkLogicSourceConnector
 
 
-# Properties defined by the MarkLogic Kafka connector
+# Properties defined by the MarkLogic Kafka connector; change these as needed to fit your environment.
 
 # Required; a MarkLogic host to connect to. By default, the connector uses the Data Movement SDK, and thus it will
 # connect to each of the hosts in a cluster.
 ml.connection.host=localhost
 
 # Required; the port of a REST API app server to connect to; if using Bulk Data Services, can be a plain HTTP app server
-ml.connection.port=8000
+ml.connection.port=8018
 
 # Required; the authentication scheme used by the server defined by ml.connection.port; either 'DIGEST', 'BASIC', 'CERTIFICATE', 'KERBEROS', or 'NONE'
 ml.connection.securityContextType=DIGEST
 
 # MarkLogic username for 'DIGEST' and 'BASIC' authentication
-ml.connection.username=
+ml.connection.username=kafka-test-user
 
 # MarkLogic password for 'DIGEST' and 'BASIC' authentication
-ml.connection.password=
+ml.connection.password=kafkatest
 
 # Path to PKCS12 file for 'CERTIFICATE' authentication
 # ml.connection.certFile=