Start work on v1.9

Update the version.
Cleanup a logging smell.
Remove a deprecated function in several places.

Author: BillFarber

build.gradle

@@ -40,7 +40,7 @@ dependencies {
   compileOnly "org.apache.kafka:connect-runtime:${kafkaVersion}"
   compileOnly "org.slf4j:slf4j-api:1.7.36"
 
-  implementation 'com.marklogic:ml-javaclient-util:4.6.0'
+  implementation 'com.marklogic:ml-javaclient-util:4.7.0'
   // Force DHF to use the latest version of ml-app-deployer, which minimizes security vulnerabilities
   implementation "com.marklogic:ml-app-deployer:4.6.0"
 

docker-compose.yml

@@ -4,7 +4,7 @@ name: marklogic-kafka
 services:
 
   marklogic:
-    image: "marklogicdb/marklogic-db:11.1.0-centos-1.1.0"
+    image: "marklogicdb/marklogic-db:11.2.0-centos-1.1.2"
     platform: linux/amd64
     environment:
       - INSTALL_CONVERTERS=true

gradle.properties

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

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

@@ -314,7 +314,7 @@ public void processFailure(WriteBatch batch, Throwable throwable) {
 
     private void logFailedWriteEvent(WriteEvent writeEvent, DocumentMetadataHandle metadata) {
         DocumentMetadataHandle.DocumentMetadataValues values = metadata.getMetadataValues();
-        if (values != null) {
+        if ((values != null) && logger.isErrorEnabled()) {
             logger.error("URI: {}; key: {}; partition: {}; offset: {}; timestamp: {}; topic: {}",
                 writeEvent.getTargetUri(), values.get("kafka-key"), values.get("kafka-partition"),
                 values.get("kafka-offset"), values.get("kafka-timestamp"), values.get("kafka-topic"));

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

@@ -16,7 +16,6 @@
 package com.marklogic.kafka.connect.source;
 
 import com.marklogic.client.document.DocumentWriteOperation;
-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;
@@ -43,7 +42,7 @@ public DocumentWriteOperation build(RecordContent recordContent) {
             metadata.getCollections().addAll(collections.trim().split(","));
         }
         if (StringUtils.hasText(permissions)) {
-            new DefaultDocumentPermissionsParser().parsePermissions(permissions.trim(), metadata.getPermissions());
+            metadata.getPermissions().addFromDelimitedString(permissions.trim());
         }
 
         if (StringUtils.hasText(uriPrefix)) {

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

@@ -18,7 +18,6 @@
 import com.marklogic.client.DatabaseClient;
 import com.marklogic.client.document.DocumentDescriptor;
 import com.marklogic.client.document.JSONDocumentManager;
-import com.marklogic.client.ext.util.DefaultDocumentPermissionsParser;
 import com.marklogic.client.io.DocumentMetadataHandle;
 import com.marklogic.client.io.Format;
 import com.marklogic.client.io.JacksonHandle;
@@ -29,7 +28,6 @@
 import java.util.Map;
 
 public class MarkLogicConstraintValueStore extends ConstraintValueStore {
-    private final DefaultDocumentPermissionsParser permissionsParser = new DefaultDocumentPermissionsParser();
     private final DatabaseClient databaseClient;
     private final String constraintStorageUri;
     private String constraintStoragePermissions;
@@ -69,7 +67,7 @@ public void storeConstraintState(String previousMaxConstraintColumnValue, int la
 
     private void insertPermissions(DocumentMetadataHandle metadataHandle) {
         if (StringUtils.hasText(constraintStoragePermissions)) {
-            permissionsParser.parsePermissions(constraintStoragePermissions, metadataHandle.getPermissions());
+            metadataHandle.getPermissions().addFromDelimitedString(constraintStoragePermissions);
         }
     }
 

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

@@ -15,7 +15,6 @@
  */
 package com.marklogic.kafka.connect.source;
 
-import com.marklogic.client.ext.util.DefaultDocumentPermissionsParser;
 import com.marklogic.client.io.DocumentMetadataHandle;
 import com.marklogic.kafka.connect.MarkLogicConfig;
 import org.apache.kafka.common.config.ConfigDef;
@@ -51,7 +50,6 @@ enum OUTPUT_TYPE {JSON, XML, CSV}
     public static final String WAIT_TIME = "ml.source.waitTime";
 
     public static final ConfigDef CONFIG_DEF = getConfigDef();
-    private static final DefaultDocumentPermissionsParser permissionsParser = new DefaultDocumentPermissionsParser();
     private static final String GROUP = "MarkLogic Source Settings";
 
     private static ConfigDef getConfigDef() {
@@ -113,7 +111,7 @@ public void ensureValid(String name, Object value) {
             if (StringUtils.hasText((String) value)) {
                 DocumentMetadataHandle metadata = new DocumentMetadataHandle();
                 try {
-                    permissionsParser.parsePermissions((String) value, metadata.getPermissions());
+                    metadata.getPermissions().addFromDelimitedString((String) value);
                 } catch (IllegalArgumentException ex) {
                     throw new ConfigException(ex.getMessage());
                 }

src/test/avro/avroTestClass-schema.avsc

@@ -20,7 +20,7 @@
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
-public class WriteAvroDataTest extends AbstractIntegrationSinkTest {
+class WriteAvroDataTest extends AbstractIntegrationSinkTest {
 
     /**
      * Verifies that an instance of an Avro-generated class - AvroTestClass - can be serialized to a byte array and

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

@@ -7,7 +7,7 @@
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 
-public class CreateTasksTest extends AbstractIntegrationSourceTest {
+class CreateTasksTest extends AbstractIntegrationSourceTest {
 
     @Test
     void taskCountGreaterThanOne() {