DHFPROD-7289: Handling custom hook errors for 5.4

Author: rjrudin

marklogic-data-hub/src/main/java/com/marklogic/hub/step/impl/QueryStepRunner.java

@@ -428,6 +428,17 @@ private RunStepResponse runHarmonizer(RunStepResponse runStepResponse, Collectio
                     // if exception is thrown update the failed related metrics
                     stepMetrics.getFailedBatches().addAndGet(1);
                     stepMetrics.getFailedEvents().addAndGet(batch.getItems().length);
+
+                    if (flow != null && flow.isStopOnError()) {
+                        // Stop the job, and then we need to call processFailure to force the FlowRunner to stop the flow
+                        JobTicket jobTicket = ticketWrapper.get("jobTicket");
+                        if (jobTicket != null) {
+                            dataMovementManager.stopJob(jobTicket);
+                        }
+                        stepItemFailureListeners.forEach((StepItemFailureListener listener) -> {
+                            listener.processFailure(runStepResponse.getJobId(), null);
+                        });
+                    }
                 }
             })
             .onQueryFailure((QueryBatchException failure) -> {

marklogic-data-hub/src/main/resources/ml-modules/root/data-hub/5/impl/flow.sjs

@@ -262,20 +262,27 @@ class Flow {
       delete this.datahub.flow;
     }
 
-    if (this.isContextDB(this.globalContext.sourceDatabase) && !combinedOptions.stepUpdate) {
-      this.runStep(items, content, combinedOptions, flowName, stepNumber, flowStep);
-    } else {
-      const flowInstance = this;
-      xdmp.invoke(
-        '/data-hub/5/impl/invoke-step.sjs',
-        {flowInstance, items, content, combinedOptions, flowName, flowStep, stepNumber},
-        {
-          database: this.globalContext.sourceDatabase ? xdmp.database(this.globalContext.sourceDatabase) : xdmp.database(),
-          update: combinedOptions.stepUpdate ? 'true': 'false',
-          commit: 'auto',
-          ignoreAmps: true
-        }
-      );
+    try {
+      if (this.isContextDB(this.globalContext.sourceDatabase) && !combinedOptions.stepUpdate) {
+        this.runStep(items, content, combinedOptions, flowName, stepNumber, flowStep);
+      } else {
+        const flowInstance = this;
+        xdmp.invoke(
+          '/data-hub/5/impl/invoke-step.sjs',
+          {flowInstance, items, content, combinedOptions, flowName, flowStep, stepNumber},
+          {
+            database: this.globalContext.sourceDatabase ? xdmp.database(this.globalContext.sourceDatabase) : xdmp.database(),
+            update: combinedOptions.stepUpdate ? 'true': 'false',
+            commit: 'auto',
+            ignoreAmps: true
+          }
+        );
+      }  
+    } catch (error) {
+      this.globalContext.failedItems = items;
+      this.globalContext.batchErrors.push(error);
+      this.updateBatchDocument(flowName, flowStep, combinedOptions, items, {});
+      throw error;
     }
 
     let writeTransactionInfo = {};

marklogic-data-hub/src/test/java/com/marklogic/hub/flow/RunStepWithCustomHookTest.java

@@ -3,21 +3,26 @@
 import com.marklogic.hub.AbstractHubCoreTest;
 import com.marklogic.hub.HubClient;
 import com.marklogic.hub.job.JobStatus;
+import com.marklogic.hub.step.RunStepResponse;
 import com.marklogic.hub.test.ReferenceModelProject;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
+import java.util.HashMap;
+import java.util.Map;
+
 import static org.junit.jupiter.api.Assertions.*;
 
 public class RunStepWithCustomHookTest extends AbstractHubCoreTest {
 
     private final static String URI_CREATED_BY_HOOK = "/insertedByHook/customer1.json";
 
     HubClient client;
+    ReferenceModelProject project;
 
     @BeforeEach
     void beforeEach() {
-        ReferenceModelProject project = installReferenceModelProject();
+        project = installReferenceModelProject();
         runAsDataHubOperator();
         client = getHubClient();
         project.createRawCustomer(1, "Jane");
@@ -46,4 +51,61 @@ void afterHook() {
         assertNull(client.getStagingClient().newDocumentManager().exists(URI_CREATED_BY_HOOK));
         assertNotNull(client.getFinalClient().newDocumentManager().exists(URI_CREATED_BY_HOOK));
     }
+
+    @Test
+    void beforeHookThrowsErrorAndStopOnErrorIsTrue() {
+        project.createRawCustomer(2, "Janet");
+
+        Map<String, Object> options = new HashMap<>();
+        options.put("stopOnError", "true");
+        options.put("throwErrorForStepNumber", "1");
+        RunFlowResponse response = runFlow(new FlowInputs("customHookFlow", "1", "2").withOptions(options));
+
+        assertEquals(JobStatus.STOP_ON_ERROR.toString(), response.getJobStatus());
+        assertEquals("1", response.getLastAttemptedStep());
+        assertEquals("0", response.getLastCompletedStep());
+        assertEquals(1, response.getStepResponses().keySet().size(),
+            "The second step should not have been run since stopOnError=true");
+
+        RunStepResponse stepResponse = response.getStepResponses().get("1");
+        assertEquals("canceled step 1", stepResponse.getStatus());
+        assertEquals(2, stepResponse.getTotalEvents());
+        assertEquals(0, stepResponse.getSuccessfulEvents());
+        assertEquals(2, stepResponse.getFailedEvents(), "Both items are considered to have failed since the " +
+            "entire batch failed due to a custom hook error");
+        assertEquals(0, stepResponse.getSuccessfulBatches());
+        assertEquals(1, stepResponse.getFailedBatches());
+        assertEquals(1, stepResponse.getStepOutput().size());
+        assertTrue(stepResponse.getStepOutput().get(0).contains("Throwing error on purpose for step number"),
+            "Unexpected step error: " + stepResponse.getStepOutput().get(0));
+        assertFalse(stepResponse.isSuccess());
+    }
+
+    @Test
+    void afterHookThrowsErrorAndStopOnErrorIsTrue() {
+        project.createRawCustomer(2, "Janet");
+
+        Map<String, Object> options = new HashMap<>();
+        options.put("stopOnError", "true");
+        options.put("throwErrorForStepNumber", "2");
+        RunFlowResponse response = runFlow(new FlowInputs("customHookFlow", "2", "1").withOptions(options));
+
+        assertEquals(JobStatus.STOP_ON_ERROR.toString(), response.getJobStatus());
+        assertEquals("2", response.getLastAttemptedStep());
+        assertEquals("0", response.getLastCompletedStep());
+        assertEquals(1, response.getStepResponses().keySet().size(),
+            "The second step (step 1 in this scenario) should not have been run since stopOnError=true");
+
+        RunStepResponse stepResponse = response.getStepResponses().get("2");
+        assertEquals("canceled step 2", stepResponse.getStatus());
+        assertEquals(2, stepResponse.getTotalEvents());
+        assertEquals(0, stepResponse.getSuccessfulEvents());
+        assertEquals(2, stepResponse.getFailedEvents());
+        assertEquals(0, stepResponse.getSuccessfulBatches());
+        assertEquals(1, stepResponse.getFailedBatches());
+        assertEquals(1, stepResponse.getStepOutput().size());
+        assertTrue(stepResponse.getStepOutput().get(0).contains("Throwing error on purpose for step number"),
+            "Unexpected step error: " + stepResponse.getStepOutput().get(0));
+        assertFalse(stepResponse.isSuccess());
+    }
 }

marklogic-data-hub/src/test/resources/entity-reference-model/modules/root/custom-modules/custom-hooks/test-hook.sjs

@@ -1,12 +1,17 @@
 declareUpdate();
 
 var uris;
-var content;
+var content; // expected to be an array
 var options;
 var flowName;
 var stepNumber;
 var step;
 
-for (const contentObject of content) {
-  xdmp.documentInsert("/insertedByHook" + contentObject.uri, contentObject.value, contentObject.context.permissions);
+if (options.throwErrorForStepNumber === stepNumber) {
+  throw Error("Throwing error on purpose for step number: " + stepNumber);
 }
+
+// Using an array function to verify that an array is passed to the hook
+content.forEach(contentObject => {
+  xdmp.documentInsert("/insertedByHook" + contentObject.uri, contentObject.value, contentObject.context.permissions);
+});