Added tests for transform errors

One slight change to ForestReader, just for sanity purposes. But this just verifies existing behavior.

Author: rjrudin

marklogic-spark-connector/src/main/java/com/marklogic/spark/reader/document/ForestReader.java

@@ -17,6 +17,7 @@
 import com.marklogic.spark.ContextSupport;
 import com.marklogic.spark.Options;
 import com.marklogic.spark.ReadProgressLogger;
+import org.apache.commons.io.IOUtils;
 import org.apache.spark.sql.catalyst.InternalRow;
 import org.apache.spark.sql.connector.read.PartitionReader;
 import org.slf4j.Logger;
@@ -160,7 +161,7 @@ private DocumentPage readPage(List<String> uris) {
 
     private void closeCurrentDocumentPage() {
         if (currentDocumentPage != null) {
-            currentDocumentPage.close();
+            IOUtils.closeQuietly(currentDocumentPage);
             currentDocumentPage = null;
         }
     }

marklogic-spark-connector/src/test/java/com/marklogic/spark/reader/document/ReadDocumentRowsTest.java

@@ -308,6 +308,21 @@ void withTransform() {
         assertEquals("value2", params.get("param2").asText());
     }
 
+    @Test
+    void transformThrowsError() {
+        Dataset<Row> dataset = startRead()
+            .option(Options.READ_DOCUMENTS_STRING_QUERY, "Vivianne")
+            .option(Options.READ_DOCUMENTS_TRANSFORM, "throwError")
+            .load();
+
+        SparkException ex = assertThrows(SparkException.class, () -> dataset.count());
+        assertTrue(ex.getMessage().contains("This is an intentional error for testing purposes."),
+            "When the transform throws an error, our connector throws a ConnectorException, but Spark seems to wrap " +
+                "its stacktrace into a SparkException, such that we can't access the original ConnectorException " +
+                "object. But the transform error should be in the error message. " +
+                "Actual message: " + ex.getMessage());
+    }
+
     private DataFrameReader startRead() {
         return newSparkSession().read()
             .format(CONNECTOR_IDENTIFIER)

marklogic-spark-connector/src/test/java/com/marklogic/spark/writer/WriteRowsWithTransformTest.java

@@ -76,11 +76,23 @@ void withParamsAndCustomDelimiter() {
         assertEquals("value,2", doc.get("params").get("param2").asText());
     }
 
+    @Test
+    void transformThrowsError() {
+        ConnectorException ex = assertThrowsConnectorException(() -> newWriterForSingleRow()
+            .option(Options.WRITE_TRANSFORM_NAME, "throwError")
+            .save());
+
+        assertTrue(ex.getMessage().contains("This is an intentional error for testing purposes."),
+            "The transform is expected to throw an error which should be caught by " +
+                "WriteBatcherDataWriter and then thrown as a ConnectorException. " +
+                "Actual error: " + ex.getMessage());
+    }
+
     @Test
     void invalidTransform() {
         ConnectorException ex = assertThrowsConnectorException(() -> newWriterForSingleRow()
-                .option(Options.WRITE_TRANSFORM_NAME, "this-doesnt-exist")
-                .save());
+            .option(Options.WRITE_TRANSFORM_NAME, "this-doesnt-exist")
+            .save());
 
         assertTrue(ex.getMessage().contains("Extension this-doesnt-exist or a dependency does not exist"),
             "The connector can't easily validate that a REST transform is valid, but the expectation is that the " +

test-app/src/main/ml-modules/transforms/throwError.sjs

@@ -0,0 +1,6 @@
+function transform(context, params, content) {
+  fn.error(xs.QName("ERROR"), "This is an intentional error for testing purposes.");
+  return content;
+};
+
+exports.transform = transform;