Merge pull request #153 from marklogic/feature/biju-feedback

Little bits of beta feedback

Author: rjrudin

README.md

@@ -188,10 +188,16 @@ data:
 - `topic` = the topic identified by `ml.source.topic`
 
 The same query can also be executed by defining a serialized version of the plan; this can be useful when the plan has
-been constructed already via the MarkLogic Java Client:
+been constructed in another tool already, such as the MarkLogic Java Client, and can then be exported into its 
+serialized form:
 
     ml.source.optic.serialized={"$optic": {"ns": "op", "fn": "operators", "args": [{"ns": "op", "fn": "from-view", "args": ["demo", "purchases"]}]}}
 
+**Warning** - the Kafka `tasks.max` property is ignored by the MarkLogic Kafka source connector. Running 2 or more tasks
+with the same configuration would produce 2 or more copies of the same records and may also lead to inconsistent 
+results when using a constraint column. If you have a valid scenario for setting this property to 2 or higher, please 
+file an issue with your use case. 
+
 ### Selecting an output type
 
 By default, the connector will return each row as JSON. As mentioned above, it is recommended to configure the Kafka

build.gradle

@@ -5,7 +5,7 @@ plugins {
   id "com.github.jk1.dependency-license-report" version "1.3"
 
   // Only used for testing
-  id 'com.marklogic.ml-gradle' version '4.3.7'
+  id 'com.marklogic.ml-gradle' version '4.5.0'
   id 'jacoco'
   id "org.sonarqube" version "3.5.0.2730"
 }
@@ -47,12 +47,9 @@ dependencies {
   compileOnly "org.apache.kafka:connect-runtime:${kafkaVersion}"
   compileOnly "org.slf4j:slf4j-api:1.7.36"
 
-  implementation ('com.marklogic:ml-javaclient-util:4.5-SNAPSHOT') {
-    changing = true
-  }
-
+  implementation 'com.marklogic:ml-javaclient-util:4.5.0'
   // Force DHF to use the latest version of ml-app-deployer, which minimizes security vulnerabilities
-  implementation "com.marklogic:ml-app-deployer:4.5-SNAPSHOT"
+  implementation "com.marklogic:ml-app-deployer:4.5.0"
 
   implementation "com.fasterxml.jackson.dataformat:jackson-dataformat-csv:2.14.1"
 

src/main/java/com/marklogic/kafka/connect/source/MarkLogicSourceConnector.java

@@ -71,10 +71,13 @@ public Class<? extends Task> taskClass() {
 
     @Override
     public List<Map<String, String>> taskConfigs(final int taskCount) {
-        final List<Map<String, String>> configs = new ArrayList<>(taskCount);
-        for (int i = 0; i < taskCount; ++i) {
-            configs.add(config);
+        if (taskCount > 1) {
+            logger.warn("As of the 1.8.0 release, the Kafka tasks.max property is ignored and a single source connector " +
+                            "task is created. This prevents duplicate records from being created via multiple instances of " +
+                            "the task with the exact same config.");
         }
+        final List<Map<String, String>> configs = new ArrayList<>(1);
+        configs.add(config);
         return configs;
     }
 

src/test/java/com/marklogic/kafka/connect/source/CreateTasksTest.java

@@ -0,0 +1,27 @@
+package com.marklogic.kafka.connect.source;
+
+import org.junit.jupiter.api.Test;
+
+import java.util.List;
+import java.util.Map;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+public class CreateTasksTest extends AbstractIntegrationSourceTest {
+
+    @Test
+    void taskCountGreaterThanOne() {
+        Map<String, String> config = newMarkLogicConfig(testConfig);
+        config.put(MarkLogicSourceConfig.DSL_QUERY, AUTHORS_OPTIC_DSL);
+        MarkLogicSourceConnector connector = new MarkLogicSourceConnector();
+        connector.start(config);
+
+        List configs = connector.taskConfigs(2);
+        assertEquals(1, configs.size(),
+            "For the 1.8.0 release, no matter what the users sets the Kafka tasks.max property to, we are only " +
+                "supporting one task to prevent the user from a configuration that we do not believe will ever be " +
+                "valid. That is - 2+ tasks with the same Optic query and config will produce 2+ copies of every row, " +
+                "and it will also encounter race conditions if the constraint column config is stored in MarkLogic. " +
+                "This restriction can be relaxed in the future if users identify valid scenarios for having 2+ tasks.");
+    }
+}