Bumping dependencies and using shadowJar

Also modified ConvertSinkRecordTest to use Jackson instead of gson so we can drop the gson dependency.

Author: rjrudin

README.md

@@ -17,7 +17,14 @@ This is a connector for subscribing to Kafka queues and pushing messages to Mark
 1. Copy the jar to the <kafkaHome>/libs on the remote server.
 1. Copy the two properties (config/marklogic-connect-distributed.properties config/marklogic-sink.properties) to <kafkaHome>/config on the remote server.
 
-See https://kafka.apache.org/quickstart for instructions on starting up Zookeeper and Kafka.
+See https://kafka.apache.org/quickstart for instructions on starting up Zookeeper and Kafka, which as of August 2022 
+will instruct you to run the following commands (in separate terminal windows, both from the Kafka home directory):
+
+    bin/zookeeper-server-start.sh config/zookeeper.properties
+
+and: 
+
+    bin/kafka-server-start.sh config/server.properties
 
 To start the Kafka connector in standalone mode (from the Kafka home directory):
 

build.gradle

@@ -1,29 +1,35 @@
 plugins {
   id 'java'
-  id 'net.saliman.properties' version '1.4.6'
+  id 'net.saliman.properties' version '1.5.2'
+  id 'com.github.johnrengelman.shadow' version '7.1.2'
   id "com.github.jk1.dependency-license-report" version "1.3"
 }
 
-sourceCompatibility = 1.8
+java {
+  sourceCompatibility = 1.8
+  targetCompatibility = 1.8
+}
 
 repositories {
   mavenCentral()
-  jcenter()
 }
 
 configurations {
   documentation
   assets
 }
 
+ext {
+  kafkaVersion = "2.8.1"
+}
+
 dependencies {
-  compileOnly "org.apache.kafka:connect-api:2.5.0"
-  compileOnly "org.apache.kafka:connect-json:2.5.0"
+  compileOnly "org.apache.kafka:connect-api:${kafkaVersion}"
+  compileOnly "org.apache.kafka:connect-json:${kafkaVersion}"
 
-  compile "com.marklogic:marklogic-client-api:5.3.0"
-  compile("com.marklogic:marklogic-data-hub:5.2.4") {
-    // Prefer the version above
-    exclude module: "marklogic-client-api"
+  implementation "com.marklogic:marklogic-client-api:5.5.3"
+
+  implementation("com.marklogic:marklogic-data-hub:5.7.2") {
     // Excluding these because there's no need for them
     exclude module: "spring-boot-autoconfigure"
     exclude module: "spring-integration-http"
@@ -34,18 +40,14 @@ dependencies {
     exclude module: "logback-classic"
   }
 
-  testCompile "org.junit.jupiter:junit-jupiter-api:5.3.0"
-  testCompile "org.apache.kafka:connect-api:2.5.0"
-  testCompile "org.apache.kafka:connect-json:2.5.0"
-  testCompile "com.google.code.gson:gson:2.8.6"
-
-  // Needed by Gradle 4.6+ - see https://www.petrikainulainen.net/programming/testing/junit-5-tutorial-running-unit-tests-with-gradle/
-  testRuntime "org.junit.jupiter:junit-jupiter-engine:5.3.0"
+  testImplementation 'org.junit.jupiter:junit-jupiter:5.9.0'
+  testImplementation "org.apache.kafka:connect-api:${kafkaVersion}"
+  testImplementation "org.apache.kafka:connect-json:${kafkaVersion}"
 
   // Forcing logback to be used for test logging
-  testRuntime "ch.qos.logback:logback-classic:1.1.8"
-  testRuntime group: "org.slf4j", name: "jcl-over-slf4j", version: "1.7.22"
-  testRuntime group: "org.slf4j", name: "slf4j-api", version: "1.7.22"
+  testImplementation "ch.qos.logback:logback-classic:1.2.11"
+  testImplementation "org.slf4j:jcl-over-slf4j:1.7.36"
+  testImplementation "org.slf4j:slf4j-api:1.7.36"
 
   documentation files('LICENSE.txt')
   documentation files('NOTICE.txt')
@@ -60,21 +62,20 @@ test {
   useJUnitPlatform()
 }
 
-// Customize the Java plugin's jar task to produce a "fat" jar with all dependencies included
-jar {
-  manifest {
-    attributes 'Implementation-Title': 'Kafka-Connect-MarkLogic',
-      'Implementation-Version': version
-  }
-  from { configurations.compile.collect { it.isDirectory() ? it : zipTree(it) } }
+shadowJar {
+  // Exclude DHF source files
+  exclude "hub-internal-artifacts/**"
+  exclude "hub-internal-config/**"
+  exclude "ml-config/**"
+  exclude "ml-modules*/**"
+  exclude "scaffolding/**"
 }
 
-task copyJarToKafka(type: Copy) {
+task copyJarToKafka(type: Copy, dependsOn: shadowJar) {
   description = "Used for local development and testing; copies the jar to your local Kafka install"
   from "build/libs"
   into "${kafkaHome}/libs"
 }
-copyJarToKafka.mustRunAfter(jar)
 
 task copyPropertyFilesToKafka(type: Copy) {
   description = "Used for local development and testing; copies the properties files to your local Kafka install"
@@ -84,7 +85,7 @@ task copyPropertyFilesToKafka(type: Copy) {
 
 task deploy {
   description = "Used for local development and testing; builds the jar and copies it and the properties files to your local Kafka install"
-  dependsOn = ["jar", "copyJarToKafka", "copyPropertyFilesToKafka"]
+  dependsOn = ["copyJarToKafka", "copyPropertyFilesToKafka"]
 }
 
 // Tasks for building the archive required for submitting to the Confluence Connector Hub
@@ -134,7 +135,7 @@ task connectorArchive_CopyDependenciesToBuildDirectory(type: Copy) {
   dependsOn = [jar]
 
   from jar
-  from configurations.compile
+  from configurations.runtimeClasspath.findAll { it.name.endsWith('jar') }
   into "${baseArchiveBuildDir}/${baseArchiveName}/lib"
 }
 

gradle.properties

@@ -1,5 +1,5 @@
 group=com.marklogic
-version=1.6.1
+version=1.7.0-SNAPSHOT
 
 # For the Confluent Connector Archive
 componentOwner=marklogic

gradle/wrapper/gradle-wrapper.properties

@@ -1,5 +1,5 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-4.10-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-7.5.1-bin.zip
 zipStoreBase=GRADLE_USER_HOME
 zipStorePath=wrapper/dists

src/test/java/com/marklogic/kafka/connect/sink/ConvertSinkRecordTest.java

@@ -1,7 +1,7 @@
 package com.marklogic.kafka.connect.sink;
 
-import com.google.gson.JsonObject;
-import com.google.gson.JsonParser;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.marklogic.client.document.DocumentWriteOperation;
 import com.marklogic.client.io.BytesHandle;
 import com.marklogic.client.io.DocumentMetadataHandle;
@@ -20,12 +20,11 @@
  */
 public class ConvertSinkRecordTest {
 
-    DefaultSinkRecordConverter converter;
-    MarkLogicSinkTask markLogicSinkTask = new MarkLogicSinkTask();
-    private JsonObject doc1, doc2, doc3;
+    private DefaultSinkRecordConverter converter;
+    private final MarkLogicSinkTask markLogicSinkTask = new MarkLogicSinkTask();
 
     @Test
-    public void allPropertiesSet() throws IOException {
+    public void allPropertiesSet() {
         Map<String, Object> config = new HashMap<>();
         config.put("ml.document.collections", "one,two");
         config.put("ml.document.format", "json");
@@ -57,10 +56,9 @@ public void allPropertiesSet() throws IOException {
     }
 
     @Test
-    public void noPropertiesSet() throws IOException {
+    public void noPropertiesSet() {
         Map<String, Object> kafkaConfig = new HashMap<String, Object>();
         converter = new DefaultSinkRecordConverter(kafkaConfig);
-        converter.getDocumentWriteOperationBuilder();
 
         DocumentWriteOperation op = converter.convert(newSinkRecord("doesn't matter"));
 
@@ -73,11 +71,10 @@ public void noPropertiesSet() throws IOException {
     }
 
     @Test
-    public void UriWithUUIDStrategy() throws IOException {
+    public void uriWithUUIDStrategy() {
         Map<String, Object> kafkaConfig = new HashMap<String, Object>();
         kafkaConfig.put("ml.id.strategy", "UUID");
         converter = new DefaultSinkRecordConverter(kafkaConfig);
-        converter.getDocumentWriteOperationBuilder();
 
         DocumentWriteOperation op = converter.convert(newSinkRecord("doesn't matter"));
 
@@ -90,10 +87,9 @@ public void UriWithUUIDStrategy() throws IOException {
     }
 
     @Test
-    public void UriWithDefaultStrategy() throws IOException {
+    public void uriWithDefaultStrategy() {
         Map<String, Object> kafkaConfig = new HashMap<String, Object>();
         converter = new DefaultSinkRecordConverter(kafkaConfig);
-        converter.getDocumentWriteOperationBuilder();
 
         DocumentWriteOperation op = converter.convert(newSinkRecord("doesn't matter"));
 
@@ -106,11 +102,10 @@ public void UriWithDefaultStrategy() throws IOException {
     }
 
     @Test
-    public void UriWithKafkaMetaData() throws IOException {
+    public void uriWithKafkaMetaData() {
         Map<String, Object> kafkaConfig = new HashMap<String, Object>();
         kafkaConfig.put("ml.id.strategy", "KAFKA_META_WITH_SLASH");
         converter = new DefaultSinkRecordConverter(kafkaConfig);
-        converter.getDocumentWriteOperationBuilder();
 
         DocumentWriteOperation op = converter.convert(newSinkRecord("doesn't matter"));
 
@@ -122,14 +117,12 @@ public void UriWithKafkaMetaData() throws IOException {
     }
 
     @Test
-    public void UriWithJsonPath() throws IOException {
-        JsonParser parser = new JsonParser();
-        doc1 = parser.parse("{\"f1\":\"100\"}").getAsJsonObject();
+    public void uriWithJsonPath() throws IOException {
+        JsonNode doc1 = new ObjectMapper().readTree("{\"f1\":\"100\"}");
         Map<String, Object> kafkaConfig = new HashMap<String, Object>();
         kafkaConfig.put("ml.id.strategy", "JSONPATH");
         kafkaConfig.put("ml.id.strategy.paths", "/f1");
         converter = new DefaultSinkRecordConverter(kafkaConfig);
-        converter.getDocumentWriteOperationBuilder();
 
         DocumentWriteOperation op = converter.convert(newSinkRecord(doc1));
 
@@ -142,14 +135,12 @@ public void UriWithJsonPath() throws IOException {
     }
 
     @Test
-    public void UriWithHashedJsonPaths() throws IOException {
-        JsonParser parser = new JsonParser();
-        doc1 = parser.parse("{\"f1\":\"100\",\"f2\":\"200\"}").getAsJsonObject();
+    public void uriWithHashedJsonPaths() throws IOException {
+        JsonNode doc1 = new ObjectMapper().readTree("{\"f1\":\"100\",\"f2\":\"200\"}");
         Map<String, Object> kafkaConfig = new HashMap<String, Object>();
         kafkaConfig.put("ml.id.strategy", "HASH");
         kafkaConfig.put("ml.id.strategy.paths", "/f1,/f2");
         converter = new DefaultSinkRecordConverter(kafkaConfig);
-        converter.getDocumentWriteOperationBuilder();
 
         DocumentWriteOperation op = converter.convert(newSinkRecord(doc1));
 
@@ -162,11 +153,10 @@ public void UriWithHashedJsonPaths() throws IOException {
     }
 
     @Test
-    public void UriWithHashedKafkaMeta() throws IOException {
+    public void uriWithHashedKafkaMeta() {
         Map<String, Object> kafkaConfig = new HashMap<String, Object>();
         kafkaConfig.put("ml.id.strategy", "KAFKA_META_HASHED");
         converter = new DefaultSinkRecordConverter(kafkaConfig);
-        converter.getDocumentWriteOperationBuilder();
 
         DocumentWriteOperation op = converter.convert(newSinkRecord("doesn't matter"));
 
@@ -179,7 +169,7 @@ public void UriWithHashedKafkaMeta() throws IOException {
     }
 
     @Test
-    public void binaryContent() throws IOException {
+    public void binaryContent() {
         converter = new DefaultSinkRecordConverter(new HashMap<>());
 
         DocumentWriteOperation op = converter.convert(newSinkRecord("hello world".getBytes()));