Refactor: Extracted marklogic-spark-api

We now have a loose coupling between the connector and the langchain4j stuff. For the 2.5.0 release, it'll all be combined, but this prepares us for upgrading langchain4j in the near future.

Haven't gotten jacoco working across multiple subprojects. That's an issue in Flux too. Going to figure that out soon.

Author: rjrudin

build.gradle

@@ -1,8 +1,32 @@
+plugins {
+  id "java-library"
+  id "jacoco-report-aggregation"
+  id "org.sonarqube" version "5.1.0.4882"
+}
+
+// See https://docs.gradle.org/current/samples/sample_jvm_multi_project_with_code_coverage_standalone.html .
+// This isn't picking up jacoco coverage of all the subprojects. Will figure that out soon.
+dependencies {
+  jacocoAggregation project(':marklogic-langchain4j')
+  jacocoAggregation project(':marklogic-spark-api')
+  jacocoAggregation project(':marklogic-spark-langchain4j')
+  jacocoAggregation project(':marklogic-spark-connector')
+}
+
+sonar {
+  properties {
+    property "sonar.projectKey", "marklogic-spark"
+    property "sonar.host.url", "http://localhost:9000"
+  }
+}
+
 subprojects {
   apply plugin: "java-library"
+  apply plugin: "jacoco"
+  apply plugin: "org.sonarqube"
 
   group = "com.marklogic"
-  version '2.5-SNAPSHOT'
+  version "2.5-SNAPSHOT"
 
   java {
     sourceCompatibility = 11
@@ -19,9 +43,18 @@ subprojects {
 
   test {
     useJUnitPlatform()
+    finalizedBy jacocoTestReport
     testLogging {
       events 'started', 'passed', 'skipped', 'failed'
       exceptionFormat 'full'
     }
   }
+
+  // See https://docs.gradle.org/current/userguide/jacoco_plugin.html .
+  jacocoTestReport {
+    dependsOn test
+    reports {
+      xml.required = true
+    }
+  }
 }

gradle/wrapper/gradle-wrapper.properties

@@ -1,6 +1,6 @@
 distributionBase=GRADLE_USER_HOME
 distributionPath=wrapper/dists
-distributionUrl=https\://services.gradle.org/distributions/gradle-8.9-bin.zip
+distributionUrl=https\://services.gradle.org/distributions/gradle-8.11-bin.zip
 networkTimeout=10000
 validateDistributionUrl=true
 zipStoreBase=GRADLE_USER_HOME

marklogic-spark-api/README.md

@@ -0,0 +1,2 @@
+This module defines the public API that marklogic-spark-langchain4j can depend on without depending on Spark or the
+connector itself.

marklogic-spark-api/build.gradle

@@ -0,0 +1,3 @@
+dependencies {
+  implementation "org.slf4j:jcl-over-slf4j:2.0.13"
+}

marklogic-spark-connector/src/main/java/com/marklogic/spark/ConnectorException.java renamed to marklogic-spark-api/src/main/java/com/marklogic/spark/ConnectorException.java

@@ -0,0 +1,60 @@
+/*
+ * Copyright © 2024 MarkLogic Corporation. All Rights Reserved.
+ */
+package com.marklogic.spark;
+
+import java.io.Serializable;
+import java.util.Map;
+import java.util.stream.Stream;
+
+public abstract class Context implements Serializable {
+
+    protected final Map<String, String> properties;
+
+    protected Context(Map<String, String> properties) {
+        this.properties = properties;
+    }
+
+    public final boolean hasOption(String... options) {
+        return Stream.of(options)
+            .anyMatch(option -> properties.get(option) != null && properties.get(option).trim().length() > 0);
+    }
+
+    public final String getStringOption(String option) {
+        return getStringOption(option, null);
+    }
+
+    public final String getStringOption(String option, String defaultValue) {
+        return hasOption(option) ? properties.get(option).trim() : defaultValue;
+    }
+
+    public final int getIntOption(String optionName, int defaultValue, int minimumValue) {
+        return (int) getNumericOption(optionName, defaultValue, minimumValue);
+    }
+
+    public final long getNumericOption(String optionName, long defaultValue, long minimumValue) {
+        try {
+            long value = this.getProperties().containsKey(optionName) ?
+                Long.parseLong(this.getProperties().get(optionName)) :
+                defaultValue;
+            if (value != defaultValue && value < minimumValue) {
+                throw new ConnectorException(String.format("The value of '%s' must be %d or greater.", getOptionNameForMessage(optionName), minimumValue));
+            }
+            return value;
+        } catch (NumberFormatException ex) {
+            throw new ConnectorException(String.format("The value of '%s' must be numeric.", getOptionNameForMessage(optionName)), ex);
+        }
+    }
+
+    public final boolean getBooleanOption(String option, boolean defaultValue) {
+        return hasOption(option) ? Boolean.parseBoolean(getStringOption(option)) : defaultValue;
+    }
+    
+    public final String getOptionNameForMessage(String option) {
+        return Util.getOptionNameForErrorMessage(option);
+    }
+
+    public Map<String, String> getProperties() {
+        return properties;
+    }
+}

marklogic-spark-api/src/main/java/com/marklogic/spark/Context.java

@@ -3,7 +3,6 @@
  */
 package com.marklogic.spark;
 
-import com.marklogic.client.document.DocumentManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -73,21 +72,4 @@ static String getOptionNameForErrorMessage(String option) {
         String optionName = bundle.getString(option);
         return optionName != null && optionName.trim().length() > 0 ? optionName.trim() : option;
     }
-
-    static Set<DocumentManager.Metadata> getRequestedMetadata(ContextSupport context) {
-        Set<DocumentManager.Metadata> set = new HashSet<>();
-        if (context.hasOption(Options.READ_DOCUMENTS_CATEGORIES)) {
-            for (String category : context.getStringOption(Options.READ_DOCUMENTS_CATEGORIES).split(",")) {
-                if ("content".equalsIgnoreCase(category)) {
-                    continue;
-                }
-                if ("metadata".equalsIgnoreCase(category)) {
-                    set.add(DocumentManager.Metadata.ALL);
-                } else {
-                    set.add(DocumentManager.Metadata.valueOf(category.toUpperCase()));
-                }
-            }
-        }
-        return set;
-    }
 }

marklogic-spark-connector/src/main/java/com/marklogic/spark/Options.java renamed to marklogic-spark-api/src/main/java/com/marklogic/spark/Options.java

@@ -3,8 +3,6 @@ plugins {
   id 'com.gradleup.shadow' version '8.3.3'
   id "com.marklogic.ml-gradle" version "5.0.0"
   id 'maven-publish'
-  id "jacoco"
-  id "org.sonarqube" version "5.1.0.4882"
 }
 
 configurations {
@@ -29,12 +27,14 @@ configurations.all {
 
 dependencies {
   // This is compileOnly as any environment this is used in will provide the Spark dependencies itself.
-  compileOnly ('org.apache.spark:spark-sql_2.12:' + sparkVersion) {
+  compileOnly('org.apache.spark:spark-sql_2.12:' + sparkVersion) {
     // Excluded from our ETL tool for size reasons, so excluded here as well to ensure we don't need it.
     exclude module: "rocksdbjni"
   }
 
-  shadowDependencies (project(":marklogic-langchain4j")) {
+  shadowDependencies project(":marklogic-spark-api")
+
+  shadowDependencies("com.marklogic:marklogic-client-api:7.0.0") {
     // The Java Client uses Jackson 2.15.2; Scala 3.4.x does not yet support that and will throw the following error:
     // Scala module 2.14.2 requires Jackson Databind version >= 2.14.0 and < 2.15.0 - Found jackson-databind version 2.15.2
     // So the 4 Jackson modules are excluded to allow for Spark's to be used.
@@ -43,7 +43,7 @@ dependencies {
   }
 
   // Required for converting JSON to XML. Using 2.15.2 to align with Spark 3.5.3.
-  shadowDependencies ("com.fasterxml.jackson.dataformat:jackson-dataformat-xml:2.15.2") {
+  shadowDependencies("com.fasterxml.jackson.dataformat:jackson-dataformat-xml:2.15.2") {
     // Not needed, as the modules in this group that this dependency depends on are all provided by Spark.
     exclude group: "com.fasterxml.jackson.core"
   }
@@ -52,26 +52,30 @@ dependencies {
   shadowDependencies 'com.squareup.okhttp3:okhttp:4.12.0'
 
   // Supports reading and writing RDF data.
-  shadowDependencies ("org.apache.jena:jena-arq:4.10.0") {
+  shadowDependencies("org.apache.jena:jena-arq:4.10.0") {
     exclude group: "com.fasterxml.jackson.core"
     exclude group: "com.fasterxml.jackson.dataformat"
   }
 
+  // Allows us to test the langchain4j-specific features without including them in the connector by default.
+  testImplementation project(":marklogic-langchain4j")
+  testImplementation project(":marklogic-spark-langchain4j")
+
   // Supports testing the embedder feature.
   testImplementation "dev.langchain4j:langchain4j-embeddings-all-minilm-l6-v2:0.35.0"
 
   // Needed for some XML operations that are far easier with JDOM2 than with DOM.
   shadowDependencies "org.jdom:jdom2:2.0.6.1"
 
-  testImplementation ('com.marklogic:ml-app-deployer:5.0.0') {
+  testImplementation('com.marklogic:ml-app-deployer:5.0.0') {
     exclude group: "com.fasterxml.jackson.core"
     exclude group: "com.fasterxml.jackson.dataformat"
 
     // Use the Java Client declared above.
     exclude module: "marklogic-client-api"
   }
 
-  testImplementation ('com.marklogic:marklogic-junit5:1.5.0') {
+  testImplementation('com.marklogic:marklogic-junit5:1.5.0') {
     exclude group: "com.fasterxml.jackson.core"
     exclude group: "com.fasterxml.jackson.dataformat"
 
@@ -85,26 +89,10 @@ dependencies {
 }
 
 test {
-  finalizedBy jacocoTestReport
   // Allows mlHost to override the value in gradle.properties, which the test plumbing will default to.
   environment "mlHost", mlHost
 }
 
-// See https://docs.gradle.org/current/userguide/jacoco_plugin.html .
-jacocoTestReport {
-  dependsOn test
-  reports {
-    xml.required = true
-  }
-}
-
-sonar {
-  properties {
-    property "sonar.projectKey", "marklogic-spark"
-    property "sonar.host.url", "http://localhost:9000"
-  }
-}
-
 task reloadTestData(type: com.marklogic.gradle.task.MarkLogicTask) {
   description = "Convenience task for clearing the test database and reloading the test data; only intended for a connector developer to use."
   doLast {

marklogic-spark-connector/src/main/java/com/marklogic/spark/Util.java renamed to marklogic-spark-api/src/main/java/com/marklogic/spark/Util.java

@@ -5,19 +5,22 @@
 
 import com.marklogic.client.DatabaseClient;
 import com.marklogic.client.DatabaseClientFactory;
+import com.marklogic.client.document.DocumentManager;
 import com.marklogic.client.extra.okhttpclient.OkHttpClientConfigurator;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.io.Serializable;
 import java.util.HashMap;
+import java.util.HashSet;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
 
-public class ContextSupport implements Serializable {
+public class ContextSupport extends Context implements Serializable {
 
     protected static final Logger logger = LoggerFactory.getLogger(ContextSupport.class);
-    private final Map<String, String> properties;
     private final boolean configuratorWasAdded;
 
     // Java Client 6.5.0 has a bug in it (to be fixed in 6.5.1 or 6.6.0) where multiple threads that use a configurator
@@ -26,7 +29,7 @@ public class ContextSupport implements Serializable {
     private static final Object CLIENT_LOCK = new Object();
 
     public ContextSupport(Map<String, String> properties) {
-        this.properties = properties;
+        super(properties);
         this.configuratorWasAdded = addOkHttpConfiguratorIfNecessary();
     }
 
@@ -92,10 +95,6 @@ public final boolean optionExists(String option) {
         return value != null && value.trim().length() > 0;
     }
 
-    public final String getOptionNameForMessage(String option) {
-        return Util.getOptionNameForErrorMessage(option);
-    }
-
     private void parseConnectionString(String value, Map<String, String> connectionProps) {
         ConnectionString connectionString = new ConnectionString(value, getOptionNameForMessage("spark.marklogic.client.uri"));
         connectionProps.put(Options.CLIENT_USERNAME, connectionString.getUsername());
@@ -109,24 +108,6 @@ private void parseConnectionString(String value, Map<String, String> connectionP
         }
     }
 
-    public final int getIntOption(String optionName, int defaultValue, int minimumValue) {
-        return (int) getNumericOption(optionName, defaultValue, minimumValue);
-    }
-
-    public final long getNumericOption(String optionName, long defaultValue, long minimumValue) {
-        try {
-            long value = this.getProperties().containsKey(optionName) ?
-                Long.parseLong(this.getProperties().get(optionName)) :
-                defaultValue;
-            if (value != defaultValue && value < minimumValue) {
-                throw new ConnectorException(String.format("The value of '%s' must be %d or greater.", getOptionNameForMessage(optionName), minimumValue));
-            }
-            return value;
-        } catch (NumberFormatException ex) {
-            throw new ConnectorException(String.format("The value of '%s' must be numeric.", getOptionNameForMessage(optionName)), ex);
-        }
-    }
-
     /**
      * Only intended for "write" use cases; an error on "read" is always expected to be propagated to the user.
      *
@@ -141,30 +122,10 @@ public final boolean isDirectConnection() {
         return value != null && value.equalsIgnoreCase(DatabaseClient.ConnectionType.DIRECT.name());
     }
 
-    public final boolean hasOption(String... options) {
-        return Util.hasOption(this.properties, options);
-    }
-
-    public final String getStringOption(String option) {
-        return getStringOption(option, null);
-    }
-
-    public final String getStringOption(String option, String defaultValue) {
-        return hasOption(option) ? properties.get(option).trim() : defaultValue;
-    }
-
-    public final boolean getBooleanOption(String option, boolean defaultValue) {
-        return hasOption(option) ? Boolean.parseBoolean(getStringOption(option)) : defaultValue;
-    }
-
     public final boolean isStreamingFiles() {
         return "true".equalsIgnoreCase(getStringOption(Options.STREAM_FILES));
     }
 
-    public Map<String, String> getProperties() {
-        return properties;
-    }
-
     /**
      * @return true if a configurator was added
      */
@@ -197,4 +158,21 @@ private boolean addOkHttpConfiguratorIfNecessary() {
         }
         return false;
     }
+
+    public static Set<DocumentManager.Metadata> getRequestedMetadata(ContextSupport context) {
+        Set<DocumentManager.Metadata> set = new HashSet<>();
+        if (context.hasOption(Options.READ_DOCUMENTS_CATEGORIES)) {
+            for (String category : context.getStringOption(Options.READ_DOCUMENTS_CATEGORIES).split(",")) {
+                if ("content".equalsIgnoreCase(category)) {
+                    continue;
+                }
+                if ("metadata".equalsIgnoreCase(category)) {
+                    set.add(DocumentManager.Metadata.ALL);
+                } else {
+                    set.add(DocumentManager.Metadata.valueOf(category.toUpperCase()));
+                }
+            }
+        }
+        return set;
+    }
 }