DEVEXP-475 User can now force job to abort if write fails

Author: rjrudin

docs/configuration.md

@@ -77,11 +77,12 @@ information on how data is written to MarkLogic.
 
 | Option | Description                                                                       | 
 | --- |-----------------------------------------------------------------------------------|
+| spark.marklogic.write.abortOnFailure | Whether the Spark job should abort if a batch fails to be written; defaults to `true`. |
 | spark.marklogic.write.batchSize | The number of documents written in a call to MarkLogic; defaults to 100. |
-| spark.marklogic.write.threadCount | The number of threads used within each partition to send documents to MarkLogic; defaults to 4. |
 | spark.marklogic.write.collections | Comma-delimited string of collection names to add to each document |
 | spark.marklogic.write.permissions | Comma-delimited string of role names and capabilities to add to each document - e.g. role1,read,role2,update,role3,execute |
 | spark.marklogic.write.temporalCollection | Name of a temporal collection to assign each document to |
+| spark.marklogic.write.threadCount | The number of threads used within each partition to send documents to MarkLogic; defaults to 4. |
 | spark.marklogic.write.transform | Name of a REST transform to apply to each document |
 | spark.marklogic.write.transformParams | Comma-delimited string of transform parameter names and values - e.g. param1,value1,param2,value2 |
 | spark.marklogic.write.transformParamsDelimiter | Delimiter to use instead of a command for the `transformParams` option |

docs/writing.md

@@ -128,10 +128,10 @@ and during the writing of data to MarkLogic.
 For the first kind of error, the error will be immediately returned to the user and no data will have been written. 
 Such errors are often due to misconfiguration of the connector options and should be fixable. 
 
-For the second kind of error, the error will eventually be returned to the user, usually within seconds of it 
-occurring. The slight delay is due to the asynchronous nature of data being written by the connector. The error will 
-be logged by the connector and the write operation will be aborted. Any batches of documents that were written 
-successfully prior to the error occurring will still exist in the database. 
+For the second kind of error, the connector defaults to logging the error and asking Spark to abort the entire write 
+operation. Any batches of documents that were written successfully prior to the error occurring will still exist in the 
+database. To configure the connector to only log the error and continue writing batches of documents to MarkLogic, set 
+the `spark.marklogic.write.abortOnFailure` option to a value of `false`. 
 
 Similar to errors with reading data, the connector will strive to provide meaningful context when an error occurs to 
 assist with debugging the cause of the error. Any errors that cannot be fixed via changes to the options passed to the 

src/main/java/com/marklogic/spark/Options.java

@@ -25,6 +25,7 @@ public interface Options {
 
     String WRITE_BATCH_SIZE = "spark.marklogic.write.batchSize";
     String WRITE_THREAD_COUNT = "spark.marklogic.write.threadCount";
+    String WRITE_ABORT_ON_FAILURE = "spark.marklogic.write.abortOnFailure";
 
     String WRITE_COLLECTIONS = "spark.marklogic.write.collections";
     String WRITE_PERMISSIONS = "spark.marklogic.write.permissions";

src/main/java/com/marklogic/spark/writer/MarkLogicDataWriter.java

@@ -62,10 +62,12 @@ class MarkLogicDataWriter implements DataWriter<InternalRow> {
         this.databaseClient = writeContext.connectToMarkLogic();
         this.dataMovementManager = this.databaseClient.newDataMovementManager();
         this.writeBatcher = writeContext.newWriteBatcher(this.dataMovementManager);
-        this.writeBatcher.onBatchFailure((batch, failure) -> {
-            // Logging not needed here, as WriteBatcherImpl already logs this at the warning level.
-            this.writeFailure.compareAndSet(null, failure);
-        });
+        if (writeContext.isAbortOnFailure()) {
+            this.writeBatcher.onBatchFailure((batch, failure) -> {
+                // Logging not needed here, as WriteBatcherImpl already logs this at the warning level.
+                this.writeFailure.compareAndSet(null, failure);
+            });
+        }
         this.dataMovementManager.startJob(this.writeBatcher);
     }
 

src/main/java/com/marklogic/spark/writer/WriteContext.java

@@ -42,7 +42,7 @@ public StructType getSchema() {
         return schema;
     }
 
-    public WriteBatcher newWriteBatcher(DataMovementManager dataMovementManager) {
+    WriteBatcher newWriteBatcher(DataMovementManager dataMovementManager) {
         WriteBatcher writeBatcher = dataMovementManager
             .newWriteBatcher()
             .withBatchSize((int) getNumericOption(Options.WRITE_BATCH_SIZE, 100, 1))
@@ -62,7 +62,7 @@ public WriteBatcher newWriteBatcher(DataMovementManager dataMovementManager) {
         return writeBatcher;
     }
 
-    public DocBuilder newDocBuilder() {
+    DocBuilder newDocBuilder() {
         DocBuilderFactory factory = new DocBuilderFactory()
             .withCollections(getProperties().get(Options.WRITE_COLLECTIONS))
             .withPermissions(getProperties().get(Options.WRITE_PERMISSIONS));
@@ -85,6 +85,10 @@ public DocBuilder newDocBuilder() {
         return factory.newDocBuilder();
     }
 
+    boolean isAbortOnFailure() {
+        return !"false".equalsIgnoreCase(getProperties().get(Options.WRITE_ABORT_ON_FAILURE));
+    }
+
     private void configureRestTransform(WriteBatcher writeBatcher) {
         String transformName = getProperties().get(Options.WRITE_TRANSFORM_NAME);
         if (transformName != null && transformName.trim().length() > 0) {

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

@@ -30,6 +30,8 @@
 
 public class WriteRowsTest extends AbstractWriteTest {
 
+    private final static String TEMPORAL_COLLECTION = "temporal-collection";
+
     @Test
     void defaultBatchSizeAndThreadCount() {
         newWriter().save();
@@ -58,15 +60,14 @@ void twoPartitions() {
 
     @Test
     void temporalTest() {
-        String temporalCollection = "temporal-collection";
         newWriterWithDefaultConfig("temporal-data.csv", 1)
-            .option(Options.WRITE_TEMPORAL_COLLECTION, temporalCollection)
+            .option(Options.WRITE_TEMPORAL_COLLECTION, TEMPORAL_COLLECTION)
             .save();
 
         String uri = getUrisInCollection(COLLECTION, 1).get(0);
         // Temporal doc is written to the temporal collection; "latest" since it's the latest version for that URI;
         // and to a collection matching the URI of the document.
-        assertInCollections(uri, COLLECTION, temporalCollection, "latest", uri);
+        assertInCollections(uri, COLLECTION, TEMPORAL_COLLECTION, "latest", uri);
     }
 
     @Test
@@ -176,6 +177,19 @@ void invalidPermissionsConfig() {
             "string: rest-reader,read,rest-writer", ex.getCause().getMessage());
     }
 
+    @Test
+    void dontAbortOnFailure() {
+        newWriterWithDefaultConfig("temporal-data-with-invalid-rows.csv", 1)
+            .option(Options.WRITE_TEMPORAL_COLLECTION, TEMPORAL_COLLECTION)
+            // Force each row in the CSV to be written in its own batch, ensuring that the one row that should succeed
+            // will in fact succeed (if it's stuck in a batch with other bad rows, it'll fail too).
+            .option(Options.WRITE_BATCH_SIZE, 1)
+            .option(Options.WRITE_ABORT_ON_FAILURE, false)
+            .save();
+
+        assertCollectionSize("9 of the batches should have failed, with the 10th batch succeeding", COLLECTION, 1);
+    }
+
     private void verifyFailureIsDueToLackOfPermission(SparkException ex) {
         assertTrue(ex.getCause() instanceof IOException, "Unexpected cause: " + ex.getCause().getClass());
         assertTrue(ex.getCause().getMessage().contains("Server Message: You do not have permission to this method and URL"),

src/test/resources/temporal-data-with-invalid-rows.csv

@@ -0,0 +1,11 @@
+id,content,systemStart,systemEnd,validStart,validEnd,columnWithOnlyWhitespace
+1,this should fail to write,,,,,"   "
+2,this should fail to write,,,,,"   "
+3,this should fail to write,,,,,"   "
+4,this should fail to write,,,,,"   "
+5,this should fail to write,,,,,"   "
+6,this should fail to write,,,,,"   "
+7,this should fail to write,,,,,"   "
+8,this should fail to write,,,,,"   "
+9,this should fail to write,,,,,"   "
+10,hello world,2014-04-03T11:00:00,2014-04-03T16:00:00,2014-04-03T11:00:00,2014-04-03T16:00:00,"   "