Merge pull request #2 from rjrudin/feature/1-transform

#1 Adding support for a REST transform with parameters

Author: BillFarber

src/main/java/com/marklogic/kafka/connect/sink/MarkLogicSinkConfig.java

@@ -26,6 +26,9 @@ public class MarkLogicSinkConfig extends AbstractConfig {
 
 	public static final String DMSDK_BATCH_SIZE = "ml.dmsdk.batchSize";
 	public static final String DMSDK_THREAD_COUNT = "ml.dmsdk.threadCount";
+	public static final String DMSDK_TRANSFORM = "ml.dmsdk.transform";
+	public static final String DMSDK_TRANSFORM_PARAMS = "ml.dmsdk.transformParams";
+	public static final String DMSDK_TRANSFORM_PARAMS_DELIMITER = "ml.dmsdk.transformParamsDelimiter";
 
 	public static final String DOCUMENT_COLLECTIONS = "ml.document.collections";
 	public static final String DOCUMENT_PERMISSIONS = "ml.document.permissions";
@@ -48,6 +51,9 @@ public class MarkLogicSinkConfig extends AbstractConfig {
 		.define(CONNECTION_EXTERNAL_NAME, Type.STRING, Importance.LOW, "External name for Kerberos authentication")
 		.define(DMSDK_BATCH_SIZE, Type.INT, 100, Importance.HIGH, "Number of documents to write in each batch")
 		.define(DMSDK_THREAD_COUNT, Type.INT, 8, Importance.HIGH, "Number of threads for DMSDK to use")
+		.define(DMSDK_TRANSFORM, Type.STRING, Importance.MEDIUM, "Name of a REST transform to use when writing documents")
+		.define(DMSDK_TRANSFORM_PARAMS, Type.STRING, Importance.MEDIUM, "Delimited set of transform names and values")
+		.define(DMSDK_TRANSFORM_PARAMS_DELIMITER, Type.STRING, ",", Importance.LOW, "Delimiter for transform parameter names and values; defaults to a comma")
 		.define(DOCUMENT_COLLECTIONS, Type.STRING, Importance.MEDIUM, "String-delimited collections to add each document to")
 		.define(DOCUMENT_FORMAT, Type.STRING, Importance.LOW, "Defines format of each document; can be one of json, xml, text, binary, or unknown")
 		.define(DOCUMENT_MIMETYPE, Type.STRING, Importance.LOW, "Defines the mime type of each document; optional, and typically the format is set instead of the mime type")

src/main/java/com/marklogic/kafka/connect/sink/MarkLogicSinkTask.java

@@ -3,6 +3,7 @@
 import com.marklogic.client.DatabaseClient;
 import com.marklogic.client.datamovement.DataMovementManager;
 import com.marklogic.client.datamovement.WriteBatcher;
+import com.marklogic.client.document.ServerTransform;
 import com.marklogic.kafka.connect.DefaultDatabaseClientCreator;
 import org.apache.kafka.connect.sink.SinkRecord;
 import org.apache.kafka.connect.sink.SinkTask;
@@ -37,11 +38,49 @@ public void start(final Map<String, String> config) {
 			.withBatchSize(Integer.valueOf(config.get(MarkLogicSinkConfig.DMSDK_BATCH_SIZE)))
 			.withThreadCount(Integer.valueOf(config.get(MarkLogicSinkConfig.DMSDK_THREAD_COUNT)));
 
+		ServerTransform transform = buildServerTransform(config);
+		if (transform != null) {
+			writeBatcher.withTransform(transform);
+		}
+
 		dataMovementManager.startJob(writeBatcher);
 
 		logger.info("Started");
 	}
 
+	/**
+	 * Builds a REST ServerTransform object based on the DMSDK parameters in the given config. If no transform name
+	 * is configured, then null will be returned.
+	 *
+	 * @param config
+	 * @return
+	 */
+	protected ServerTransform buildServerTransform(final Map<String, String> config) {
+		String transform = config.get(MarkLogicSinkConfig.DMSDK_TRANSFORM);
+		if (transform != null && transform.trim().length() > 0) {
+			ServerTransform t = new ServerTransform(transform);
+			String params = config.get(MarkLogicSinkConfig.DMSDK_TRANSFORM_PARAMS);
+			if (params != null && params.trim().length() > 0) {
+				String delimiter = config.get(MarkLogicSinkConfig.DMSDK_TRANSFORM_PARAMS_DELIMITER);
+				if (delimiter != null && delimiter.trim().length() > 0) {
+					String[] tokens = params.split(delimiter);
+					for (int i = 0; i < tokens.length; i += 2) {
+						if (i + 1 >= tokens.length) {
+							throw new IllegalArgumentException(String.format("The value of the %s property does not have an even number of " +
+								"parameter names and values; property value: %s", MarkLogicSinkConfig.DMSDK_TRANSFORM_PARAMS, params));
+						}
+						t.addParameter(tokens[i], tokens[i + 1]);
+					}
+				} else {
+					logger.warn(String.format("Unable to apply transform parameters to transform: %s; please set the " +
+						"delimiter via the %s property", transform, MarkLogicSinkConfig.DMSDK_TRANSFORM_PARAMS_DELIMITER));
+				}
+			}
+			return t;
+		}
+		return null;
+	}
+
 	/**
 	 * Creates a new DatabaseClient based on the configuration properties found in the given map.
 	 *

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

@@ -0,0 +1,63 @@
+package com.marklogic.kafka.connect.sink;
+
+import com.marklogic.client.document.ServerTransform;
+import org.junit.jupiter.api.Test;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+public class BuildServerTransformTest {
+
+	private MarkLogicSinkTask task = new MarkLogicSinkTask();
+	private Map<String, String> config = new HashMap<>();
+
+	@Test
+	public void noTransform() {
+		assertNull(task.buildServerTransform(config));
+	}
+
+	@Test
+	public void noParams() {
+		config.put(MarkLogicSinkConfig.DMSDK_TRANSFORM, "noParams");
+		ServerTransform t = task.buildServerTransform(config);
+		assertEquals("noParams", t.getName());
+		assertTrue(t.keySet().isEmpty());
+	}
+
+	@Test
+	public void oneParam() {
+		config.put(MarkLogicSinkConfig.DMSDK_TRANSFORM, "oneParam");
+		config.put(MarkLogicSinkConfig.DMSDK_TRANSFORM_PARAMS, "param1,value1");
+		config.put(MarkLogicSinkConfig.DMSDK_TRANSFORM_PARAMS_DELIMITER, ",");
+		ServerTransform t = task.buildServerTransform(config);
+		assertEquals(1, t.keySet().size());
+		assertEquals("value1", t.get("param1").get(0));
+	}
+
+	@Test
+	public void twoParamsWithCustomDelimiter() {
+		config.put(MarkLogicSinkConfig.DMSDK_TRANSFORM, "twoParams");
+		config.put(MarkLogicSinkConfig.DMSDK_TRANSFORM_PARAMS, "param1;value1;param2;value2");
+		config.put(MarkLogicSinkConfig.DMSDK_TRANSFORM_PARAMS_DELIMITER, ";");
+		ServerTransform t = task.buildServerTransform(config);
+		assertEquals(2, t.keySet().size());
+		assertEquals("value1", t.get("param1").get(0));
+		assertEquals("value2", t.get("param2").get(0));
+	}
+
+	@Test
+	public void malformedParams() {
+		config.put(MarkLogicSinkConfig.DMSDK_TRANSFORM, "malformedParams");
+		config.put(MarkLogicSinkConfig.DMSDK_TRANSFORM_PARAMS, "param1,value1,param2");
+		config.put(MarkLogicSinkConfig.DMSDK_TRANSFORM_PARAMS_DELIMITER, ",");
+		try {
+			task.buildServerTransform(config);
+			fail("The call should have failed because the params property does not have an even number of parameter " +
+				"names and values");
+		} catch (IllegalArgumentException ex) {
+			assertTrue(ex.getMessage().startsWith("The value of the ml.dmsdk.transformParams property"));
+		}
+	}
+}