Merge branch 'develop'

* Removing some erroneous comments in the example connect-standalone properties file.
* Changes for additional Security options
* Updated build.gradle to new kafka connect and datahub
* Updated documentation for v1.2.2
* Updated Test Cases.
* Some forgotten cleanup left over from building the Confluent archive.
* Support for AVRO, JSON Schema, ProtoBuf messages.
* Also support of ID strategies - UUID, JSONPATH,HASH, KAFKA_META_WITH_SLASH, KAFKA_META_HASHED
m>

Co-authored-by: bgeorge <[email protected]>

Author: BillFarber

.gitignore

@@ -4,7 +4,6 @@
 build
 out
 gradle-local.properties
-testConfig
 
 bin
 .vscode  

CHANGELOG.md

@@ -1,5 +1,14 @@
 # Change Log
 
+## [1.4.0](https://github.com/marklogic-community/kafka-marklogic-connector/releases/tag/1.4.0) (2020-10-12)
+   [Full Changelog](https://github.com/marklogic-community/kafka-marklogic-connector/compare/1.4.0...master)
+
+- Support for JSON Schema, ProtoBuf messages
+- Support of ID strategies: UUID, JSONPATH,HASH, KAFKA_META_WITH_SLASH, KAFKA_META_HASHED
+- Changes for additional Security options
+- Updated build.gradle to new kafka connect and datahub
+
+
 ## [1.3.0](https://github.com/marklogic-community/kafka-marklogic-connector/releases/tag/1.3.0) (2020-08-18)
    [Full Changelog](https://github.com/marklogic-community/kafka-marklogic-connector/compare/1.3.0...master)
 

MarkLogic_Kafka_Connector_v1.2.2.pdf

@@ -20,6 +20,7 @@ dependencies {
     compileOnly "org.apache.kafka:connect-api:2.5.0"
     compileOnly "org.apache.kafka:connect-json:2.5.0"
 
+
     compile ("com.marklogic:marklogic-data-hub:5.2.2") {
         // Excluding these because there's no need for them
         exclude module: "spring-boot-autoconfigure"
@@ -34,6 +35,7 @@ dependencies {
     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"

MarkLogic_Kafka_Connector_v1.2.3.pdf

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

build.gradle

@@ -1,10 +1,14 @@
 package com.marklogic.client.ext.document;
 
+import java.io.IOException;
+
 import com.marklogic.client.document.DocumentWriteOperation;
+import com.marklogic.client.id.strategy.IdStrategy;
 import com.marklogic.client.ext.util.DefaultDocumentPermissionsParser;
 import com.marklogic.client.impl.DocumentWriteOperationImpl;
 import com.marklogic.client.io.DocumentMetadataHandle;
 import com.marklogic.client.io.marker.AbstractWriteHandle;
+import com.marklogic.client.document.RecordContent;
 
 public class DocumentWriteOperationBuilder {
 
@@ -13,34 +17,31 @@ public class DocumentWriteOperationBuilder {
 	private String uriSuffix;
 	private String collections;
 	private String permissions;
-
-	private ContentIdExtractor contentIdExtractor = new DefaultContentIdExtractor();
-
-	public DocumentWriteOperation build(AbstractWriteHandle content, DocumentMetadataHandle metadata ) {
+	
+	public DocumentWriteOperation build(RecordContent record) throws IOException {
+		
+		AbstractWriteHandle content = record.getContent();
+		DocumentMetadataHandle metadata = record.getAdditionalMetadata();
+		String uri = record.getId();
+		
 		if (content == null) {
 			throw new NullPointerException("'content' must not be null");
 		}
-
 		if (hasText(collections)) {
 			metadata.getCollections().addAll(collections.trim().split(","));
 		}
 		if (hasText(permissions)) {
 			new DefaultDocumentPermissionsParser().parsePermissions(permissions.trim(), metadata.getPermissions());
 		}
 
-		String uri = buildUri(content);
-		return build(operationType, uri, metadata, content);
-	}
-
-	protected String buildUri(AbstractWriteHandle content) {
-		String uri = contentIdExtractor.extractId(content);
 		if (hasText(uriPrefix)) {
 			uri = uriPrefix + uri;
 		}
 		if (hasText(uriSuffix)) {
 			uri += uriSuffix;
 		}
-		return uri;
+
+		return build(operationType, uri, metadata, content);
 	}
 
 	/**
@@ -86,8 +87,4 @@ public DocumentWriteOperationBuilder withOperationType(DocumentWriteOperation.Op
 		return this;
 	}
 
-	public DocumentWriteOperationBuilder withContentIdExtractor(ContentIdExtractor contentIdExtractor) {
-		this.contentIdExtractor = contentIdExtractor;
-		return this;
-	}
 }

gradle.properties

@@ -0,0 +1,36 @@
+package com.marklogic.client.document;
+
+import com.marklogic.client.io.marker.AbstractWriteHandle;
+import com.marklogic.client.io.DocumentMetadataHandle;
+
+public class RecordContent {
+
+	AbstractWriteHandle content;
+	DocumentMetadataHandle additionalMetadata;
+	String id; 
+	
+	public AbstractWriteHandle getContent() {
+		return content;
+	}
+	
+	public void setContent(AbstractWriteHandle content) {
+		this.content = content;
+	}	
+	
+	public DocumentMetadataHandle getAdditionalMetadata() {
+		return additionalMetadata;
+	}
+	
+	public void setAdditionalMetadata(DocumentMetadataHandle additionalMetadata) {
+		this.additionalMetadata = additionalMetadata;
+	}	
+	
+	public String getId() {
+		return id;
+	}
+	public void setId(String id) {
+		this.id = id;
+	}	
+
+}
+

src/main/java/com/marklogic/client/ext/document/ContentIdExtractor.java

@@ -0,0 +1,15 @@
+package com.marklogic.client.id.strategy;
+
+import java.util.UUID;
+
+import com.marklogic.client.io.marker.AbstractWriteHandle;
+
+
+public class DefaultStrategy implements IdStrategy{
+	
+	@Override
+	public String generateId(AbstractWriteHandle content, String topic, Integer partition, long offset) {
+		return UUID.randomUUID().toString();
+	}
+    
+} 

src/main/java/com/marklogic/client/ext/document/DefaultContentIdExtractor.java

@@ -0,0 +1,57 @@
+package com.marklogic.client.id.strategy;
+
+import java.io.IOException;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.UUID;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.marklogic.client.io.marker.AbstractWriteHandle;
+
+
+public class HashedJSONPathsStrategy implements IdStrategy{
+	
+	private static final Logger logger = LoggerFactory.getLogger(HashedJSONPathsStrategy.class);
+	
+	private String [] paths;
+	
+	public HashedJSONPathsStrategy(String [] paths) {
+		this.paths = paths;
+	}
+	
+	@Override
+	public String generateId(AbstractWriteHandle content, String topic, Integer partition, long offset) {
+		ObjectMapper om = new ObjectMapper();
+		String valueString = "";
+		try {
+			MessageDigest md = MessageDigest.getInstance("MD5");
+			JsonNode node = om.readTree(content.toString());
+			for (int i=0; i<paths.length; i++) {
+				valueString = valueString + node.at(paths[i].trim()).asText();
+			}
+			String id = bytesToHex(md.digest(valueString.getBytes()));
+			return id;
+		}
+		catch (IOException e) {
+			logger.warn("IOException. Not creating MD5 URI, instead generating UUID");
+			return UUID.randomUUID().toString();
+		} 
+		catch (NoSuchAlgorithmException e) {
+			logger.warn("NoSuchAlgorithmException. Not creating MD5 URI, instead generating UUID");
+			return UUID.randomUUID().toString();
+		}
+	}
+
+	private static String bytesToHex(byte[] bytes) {
+    StringBuilder sb = new StringBuilder();
+    for (byte b : bytes) {
+        sb.append(String.format("%02x", b));
+    }
+    return sb.toString();
+	}
+    
+} 

src/main/java/com/marklogic/client/ext/document/DocumentWriteOperationBuilder.java

@@ -0,0 +1,37 @@
+package com.marklogic.client.id.strategy;
+
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.UUID;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.marklogic.client.io.marker.AbstractWriteHandle;
+
+public class HashedKafkaMetaStrategy implements IdStrategy {
+	
+	private static final Logger logger = LoggerFactory.getLogger(HashedKafkaMetaStrategy.class);
+
+	@Override
+	public String generateId(AbstractWriteHandle content,String topic, Integer partition, long offset) {
+		String id = "";
+		try {
+			MessageDigest md = MessageDigest.getInstance("MD5");
+			String tmp = topic + partition.toString() + String.valueOf(offset);
+			id = bytesToHex(md.digest(tmp.getBytes()));
+			return id;
+		} catch (NoSuchAlgorithmException e) {
+			logger.warn("NoSuchAlgorithmException. Not creating MD5 URI, instead generating UUID");
+			return UUID.randomUUID().toString();
+		}
+	}
+	
+	private static String bytesToHex(byte[] bytes) {
+    StringBuilder sb = new StringBuilder();
+    for (byte b : bytes) {
+        sb.append(String.format("%02x", b));
+    }
+    return sb.toString();
+	}
+}

src/main/java/com/marklogic/client/ext/document/RecordContent.java

@@ -0,0 +1,10 @@
+package com.marklogic.client.id.strategy;
+
+import java.util.UUID;
+import com.marklogic.client.io.marker.AbstractWriteHandle;
+
+public interface IdStrategy {
+	default String generateId(AbstractWriteHandle content, String topic, Integer partition, long offset) {
+		return UUID.randomUUID().toString();
+	}
+} 

src/main/java/com/marklogic/client/id/strategy/DefaultStrategy.java

@@ -0,0 +1,29 @@
+package com.marklogic.client.id.strategy;
+
+import java.util.Map;
+
+import com.marklogic.kafka.connect.sink.MarkLogicSinkConfig;
+
+public  class IdStrategyFactory {
+	
+	public static IdStrategy getIdStrategy(Map<String, String> kafkaConfig) {
+		String strategyType = kafkaConfig.get(MarkLogicSinkConfig.ID_STRATEGY);
+		String strategyPaths= kafkaConfig.get(MarkLogicSinkConfig.ID_STRATEGY_PATH);
+		
+		switch((strategyType != null) ? strategyType : "UUID") {
+			case "JSONPATH":
+				return (new JSONPathStrategy(strategyPaths.trim().split(",")[0]));
+			case "HASH":
+				return (new HashedJSONPathsStrategy(strategyPaths.trim().split(",")));
+			case "UUID":
+				return (new DefaultStrategy());
+			case "KAFKA_META_WITH_SLASH":
+				return (new KafkaMetaStrategy());
+			case "KAFKA_META_HASHED":
+				return (new HashedKafkaMetaStrategy());
+			default: 
+				return (new DefaultStrategy());
+		}
+	}
+	
+} 

src/main/java/com/marklogic/client/id/strategy/HashedJSONPathsStrategy.java

@@ -0,0 +1,37 @@
+package com.marklogic.client.id.strategy;
+
+import java.io.IOException;
+import java.util.UUID;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.marklogic.client.io.marker.AbstractWriteHandle;
+
+public class JSONPathStrategy implements IdStrategy{
+	
+	private static final Logger logger = LoggerFactory.getLogger(JSONPathStrategy.class);
+	
+	private String path;
+	
+	public JSONPathStrategy (String path) {
+		this.path = path;
+	}
+	
+	@Override
+	public String generateId(AbstractWriteHandle content, String topic, Integer partition, long offset) {
+		ObjectMapper om = new ObjectMapper();
+		try {
+			JsonNode node = om.readTree(content.toString());
+			String id = node.at(path).asText();
+			return id;
+		}
+		catch (IOException e) {
+			logger.warn("IOException. Not creating JSONPATH URI, instead generating UUID");
+			return UUID.randomUUID().toString();
+		}
+	}
+	
+} 

src/main/java/com/marklogic/client/id/strategy/HashedKafkaMetaStrategy.java

@@ -0,0 +1,14 @@
+package com.marklogic.client.id.strategy;
+
+import com.marklogic.client.io.marker.AbstractWriteHandle;
+
+public class KafkaMetaStrategy implements IdStrategy {
+
+	@Override
+	public String generateId(AbstractWriteHandle content, String topic, Integer partition, long offset) {
+		String id = "";
+		id = topic + "/" + partition.toString() + "/" + String.valueOf(offset);
+		return id;
+	}
+
+}