Merge pull request #368 from marklogic/feature/error-fix

Throw full exception on write failure

Author: rjrudin

CONTRIBUTING.md

@@ -98,7 +98,7 @@ This will produce a single jar file for the connector in the `./build/libs` dire
 
 You can then launch PySpark with the connector available via:
 
-    pyspark --jars build/libs/marklogic-spark-connector-2.4-SNAPSHOT.jar
+    pyspark --jars marklogic-spark-connector/build/libs/marklogic-spark-connector-2.5-SNAPSHOT.jar
 
 The below command is an example of loading data from the test application deployed via the instructions at the top of 
 this page. 
@@ -142,7 +142,7 @@ For a quick test of writing documents, use the following:
 
 ```
 
-spark.read.option("header", True).csv("src/test/resources/data.csv")\
+spark.read.option("header", True).csv("marklogic-spark-connector/src/test/resources/data.csv")\
     .repartition(2)\
     .write.format("marklogic")\
     .option("spark.marklogic.client.uri", "spark-test-user:spark@localhost:8000")\
@@ -174,7 +174,7 @@ The Spark master GUI is at <http://localhost:8080>. You can use this to view det
 
 Now that you have a Spark cluster running, you just need to tell PySpark to connect to it:
 
-    pyspark --master spark://NYWHYC3G0W:7077 --jars build/libs/marklogic-spark-connector-2.4-SNAPSHOT.jar
+    pyspark --master spark://NYWHYC3G0W:7077 --jars marklogic-spark-connector/build/libs/marklogic-spark-connector-2.5-SNAPSHOT.jar
 
 You can then run the same commands as shown in the PySpark section above. The Spark master GUI will allow you to 
 examine details of each of the commands that you run.
@@ -193,12 +193,12 @@ You will need the connector jar available, so run `./gradlew clean shadowJar` if
 You can then run a test Python program in this repository via the following (again, change the master address as 
 needed); note that you run this outside of PySpark, and `spark-submit` is available after having installed PySpark:
 
-    spark-submit --master spark://NYWHYC3G0W:7077 --jars build/libs/marklogic-spark-connector-2.4-SNAPSHOT.jar src/test/python/test_program.py
+    spark-submit --master spark://NYWHYC3G0W:7077 --jars marklogic-spark-connector/build/libs/marklogic-spark-connector-2.5-SNAPSHOT.jar src/test/python/test_program.py
 
 You can also test a Java program. To do so, first move the `com.marklogic.spark.TestProgram` class from `src/test/java`
 to `src/main/java`. Then run `./gradlew clean shadowJar` to rebuild the connector jar. Then run the following:
 
-    spark-submit --master spark://NYWHYC3G0W:7077 --class com.marklogic.spark.TestProgram build/libs/marklogic-spark-connector-2.4-SNAPSHOT.jar
+    spark-submit --master spark://NYWHYC3G0W:7077 --class com.marklogic.spark.TestProgram marklogic-spark-connector/build/libs/marklogic-spark-connector-2.5-SNAPSHOT.jar
 
 Be sure to move `TestProgram` back to `src/test/java` when you are done. 
 

marklogic-spark-connector/build.gradle

@@ -100,11 +100,6 @@ dependencies {
   testImplementation "org.skyscreamer:jsonassert:1.5.1"
 }
 
-//test {
-//  // Allows mlHost to override the value in gradle.properties, which the test plumbing will default to.
-//  environment "mlHost", mlHost
-//}
-
 if (JavaVersion.current().isCompatibleWith(JavaVersion.VERSION_17)) {
   test {
     // See https://stackoverflow.com/questions/72724816/running-unit-tests-with-spark-3-3-0-on-java-17-fails-with-illegalaccesserror-cl

marklogic-spark-connector/src/main/java/com/marklogic/spark/ContextSupport.java

@@ -16,7 +16,6 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.TimeUnit;
-import java.util.stream.Stream;
 
 public class ContextSupport extends Context implements Serializable {
 

marklogic-spark-connector/src/main/java/com/marklogic/spark/writer/WriteBatcherDataWriter.java

@@ -227,9 +227,12 @@ private synchronized void throwWriteFailureIfExists() {
             if (failure instanceof ConnectorException) {
                 throw (ConnectorException) failure;
             }
-            // Only including the message seems sufficient here, as Spark is logging the stacktrace. And the user
-            // most likely only needs to know the message.
-            throw new ConnectorException(failure.getMessage());
+            // Originally, only the failure message was included under the impression that the Spark environment was
+            // logging the full stacktrace. That either was not the case or is no longer the case on Spark 3.5.x.
+            // So the original exception is retained. But oddly, this results in a SparkException with a null cause - ???.
+            // That doesn't really impact a user - it's a SparkException regardless - but caused some tests to no longer
+            // be able to catch a ConnectorException.
+            throw new ConnectorException(failure.getMessage(), failure);
         }
     }
 

marklogic-spark-connector/src/main/java/com/marklogic/spark/writer/file/ContentWriter.java

@@ -11,7 +11,6 @@
 import com.marklogic.spark.ConnectorException;
 import com.marklogic.spark.ContextSupport;
 import com.marklogic.spark.Options;
-import com.marklogic.spark.Util;
 import com.marklogic.spark.reader.document.DocumentRowSchema;
 import org.apache.commons.io.IOUtils;
 import org.apache.spark.sql.catalyst.InternalRow;

marklogic-spark-connector/src/test/java/com/marklogic/spark/AbstractIntegrationTest.java

@@ -158,7 +158,8 @@ protected final ConnectorException assertThrowsConnectorException(Runnable r) {
         SparkException ex = assertThrows(SparkException.class, () -> r.run());
         assertTrue(ex.getCause() instanceof ConnectorException,
             "Expect the Spark-thrown SparkException to wrap our ConnectorException, which is an exception that we " +
-                "intentionally throw when an error condition is detected.");
+                "intentionally throw when an error condition is detected. " +
+                "Actual exception cause type: " + ex.getCause());
         return (ConnectorException) ex.getCause();
     }
 

marklogic-spark-connector/src/test/java/com/marklogic/spark/reader/file/ReadGenericFilesTest.java

@@ -11,6 +11,7 @@
 import com.marklogic.spark.AbstractIntegrationTest;
 import com.marklogic.spark.ConnectorException;
 import com.marklogic.spark.Options;
+import org.apache.spark.SparkException;
 import org.apache.spark.sql.DataFrameWriter;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
@@ -72,7 +73,7 @@ void wrongEncoding() {
             .option(Options.WRITE_PERMISSIONS, DEFAULT_PERMISSIONS)
             .mode(SaveMode.Append);
 
-        ConnectorException ex = assertThrowsConnectorException(() -> writer.save());
+        SparkException ex = assertThrows(SparkException.class, () -> writer.save());
         assertTrue(ex.getMessage().contains("document is not UTF-8 encoded"), "Actual error: " + ex.getMessage());
     }
 

marklogic-spark-connector/src/test/java/com/marklogic/spark/reader/file/StreamGenericFilesTest.java

@@ -4,8 +4,8 @@
 package com.marklogic.spark.reader.file;
 
 import com.marklogic.spark.AbstractIntegrationTest;
-import com.marklogic.spark.ConnectorException;
 import com.marklogic.spark.Options;
+import org.apache.spark.SparkException;
 import org.apache.spark.sql.DataFrameWriter;
 import org.apache.spark.sql.Dataset;
 import org.apache.spark.sql.Row;
@@ -73,7 +73,7 @@ void handleFailureWhileStreaming() {
             .option(Options.WRITE_PERMISSIONS, "not-an-actual-role,read")
             .mode(SaveMode.Append);
 
-        ConnectorException ex = assertThrowsConnectorException(() -> writer.save());
+        SparkException ex = assertThrows(SparkException.class, () -> writer.save());
         assertTrue(ex.getMessage().contains("SEC-ROLEDNE: xdmp:role(\"not-an-actual-role\")"),
             "This verifies that when the connector uses GenericDocumentManager to PUT a single document, any error " +
                 "is still wrapped in a ConnectorException. Actual error message: " + ex.getMessage());

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

@@ -3,12 +3,13 @@
  */
 package com.marklogic.spark.writer;
 
-import com.marklogic.spark.ConnectorException;
 import com.marklogic.spark.Options;
+import org.apache.spark.SparkException;
 import org.apache.spark.sql.DataFrameWriter;
 import org.apache.spark.sql.SaveMode;
 import org.junit.jupiter.api.Test;
 
+import static org.junit.jupiter.api.Assertions.assertThrows;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 class WritePartialBatchTest extends AbstractWriteTest {
@@ -39,7 +40,7 @@ void shouldThrowError() {
             .option(Options.WRITE_PERMISSIONS, DEFAULT_PERMISSIONS)
             .mode(SaveMode.Append);
 
-        ConnectorException ex = assertThrowsConnectorException(() -> writer.save());
+        SparkException ex = assertThrows(SparkException.class, () -> writer.save());
         assertTrue(ex.getMessage().contains("Document is not JSON"), "Verifying that trying to write non-JSON " +
             "documents with a .json extension should produce an error; unexpected error: " + ex.getMessage());
     }

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

@@ -151,7 +151,7 @@ void invalidBatchSize() {
      */
     @Test
     void userNotPermittedToWriteAndFailOnCommit() {
-        ConnectorException ex = assertThrowsConnectorException(() -> newWriter()
+        SparkException ex = assertThrows(SparkException.class, () -> newWriter()
             .option(Options.CLIENT_USERNAME, "spark-no-write-user")
             .option(Options.WRITE_BATCH_SIZE, 500)
             .save()
@@ -181,7 +181,7 @@ void invalidPassword() {
      */
     @Test
     void userNotPermittedToWriteAndFailOnWrite() {
-        ConnectorException ex = assertThrowsConnectorException(() -> newWriter()
+        SparkException ex = assertThrows(SparkException.class, () -> newWriter()
             .option(Options.CLIENT_USERNAME, "spark-no-write-user")
             .option(Options.WRITE_BATCH_SIZE, 1)
             .option(Options.WRITE_THREAD_COUNT, 1)
@@ -224,7 +224,7 @@ void dontAbortOnFailure() {
         assertEquals(1, successCount.get());
     }
 
-    private void verifyFailureIsDueToLackOfPermission(ConnectorException ex) {
+    private void verifyFailureIsDueToLackOfPermission(SparkException ex) {
         assertTrue(ex.getMessage().contains("Server Message: You do not have permission to this method and URL"),
             "Unexpected cause message: " + ex.getMessage());
         verifyNoDocsWereWritten();