DHFPROD-7570: stopOnError now works when the collector fails

I had to update a couple existing tests, which I think in hindsight were erroneously expecting the wrong job status. If stopOnError=true, and the collector fails, then a user expects the flow to be stopped and no other steps to execute, which means the job status should be stop-on-error.

Author: rjrudin

marklogic-data-hub/src/main/java/com/marklogic/hub/flow/impl/FlowRunnerImpl.java

@@ -337,8 +337,7 @@ public void run() {
                         .onItemFailed((jobId, itemId)-> {
                             errorCount.incrementAndGet();
                             if(flow.isStopOnError()){
-                                jobStoppedOnError.set(true);
-                                stopJob(jobId);
+                                stopJobOnError(jobId);
                             }
                         });
 
@@ -347,6 +346,10 @@ public void run() {
                     }
                     stepResp = stepRunner.run();
                     stepRunner.awaitCompletion();
+                    final boolean stepFailed = stepResp.getStatus() != null && stepResp.getStatus().startsWith("failed");
+                    if (stepFailed && runningFlow.isStopOnError()) {
+                        stopJobOnError(runningJobId);
+                    }
                 }
                 catch (Exception e) {
                     stepResp = RunStepResponse.withFlow(flow).withStep(stepNum);
@@ -370,8 +373,7 @@ public void run() {
                     }
 
                     if(runningFlow.isStopOnError()) {
-                        jobStoppedOnError.set(true);
-                        stopJob(runningJobId);
+                        stopJobOnError(runningJobId);
                     }
                 }
                 finally {
@@ -450,6 +452,12 @@ else if (!isJobSuccess.get()) {
                 }
             }
         }
+
+        private void stopJobOnError(String jobId) {
+            jobStoppedOnError.set(true);
+            stopJob(jobId);
+        }
+
     }
 
     public void awaitCompletion() {

marklogic-data-hub/src/test/java/com/marklogic/hub/flow/impl/FlowRunnerTest.java

@@ -130,7 +130,7 @@ void customStepReferencesModulePathThatDoesntExist() {
         List<String> errors = stepResponse.getStepOutput();
         assertEquals(1, errors.size(), "Expecting an error due to the missing module");
         assertTrue(errors.get(0).contains("Unable to access module: /custom-modules/custom/value-step/main.sjs. " +
-            "Verify that this module is in your modules database and that your user account has a role that grants read and execute permission to this module"),
+                "Verify that this module is in your modules database and that your user account has a role that grants read and execute permission to this module"),
             "Did not find expected message in error; error: " + errors.get(0));
 
         assertEquals(2, stepResponse.getTotalEvents(), "Expecting 2, as there are 2 URIs matching the collection query");
@@ -235,7 +235,7 @@ public void testIngestCSVasXML() {
             "Per DHFPROD-6665, the DHF file ingester now includes a no-namespaced 'root' element around each of the elements constructed from a " +
                 "row in the delimited file. This matches what MLCP does, thereby avoiding confusion from a user getting different results from DHF " +
                 "and from MLCP.");
-     }
+    }
 
     @SuppressWarnings("deprecation")
     protected RunFlowResponse runFlow(String flowName, String commaDelimitedSteps, String jobId, Map<String, Object> options, Map<String, Object> stepConfig) {
@@ -286,15 +286,18 @@ public void testSourceQueryLimit(){
         verifyJobFinished(resp);
         assertEquals(2, getDocCount(HubConfig.DEFAULT_FINAL_NAME, "xml-map"));
 
+        final String stopOnErrorReason = "Per DHFPROD-7570, if the sourceQuery fails for any reason, and the flow has stopOnError=true, then " +
+            "the flow should be stopped and the job status should be stop-on-error.";
+
         opts.put("sourceQueryLimit", -2);
         resp = runFlow("testFlow", "6", UUID.randomUUID().toString(), opts, new HashMap<>());
         flowRunner.awaitCompletion();
-        assertEquals("failed", resp.getJobStatus(), "Unexpected job status: " + resp.toJson());
+        assertEquals(JobStatus.STOP_ON_ERROR.toString(), resp.getJobStatus(), stopOnErrorReason);
 
         opts.put("sourceQueryLimit", "invalidValue");
         resp = runFlow("testFlow", "6", UUID.randomUUID().toString(), opts, new HashMap<>());
         flowRunner.awaitCompletion();
-        assertEquals("failed", resp.getJobStatus(), "Unexpected job status: " + resp.toJson());
+        assertEquals(JobStatus.STOP_ON_ERROR.toString(), resp.getJobStatus(), stopOnErrorReason);
     }
 
     @Test
@@ -343,8 +346,8 @@ public void testInvalidQueryCollector(){
 
         RunFlowResponse resp = runFlow("testFlow", "1,6", UUID.randomUUID().toString(), opts, null);
         flowRunner.awaitCompletion();
-        assertEquals(JobStatus.FINISHED_WITH_ERRORS.toString(), resp.getJobStatus(), "Since one step completed and " +
-            "the other failed, the status should be finished with errors: " + resp.toJson());
+        assertEquals(JobStatus.STOP_ON_ERROR.toString(), resp.getJobStatus(), "Per DHFPROD-7570, since the sourceQuery failed and " +
+            "stopOnError=true, the job status should be stop-on-error, even though one step did complete successfully");
         assertEquals(1, getDocCount(HubConfig.DEFAULT_STAGING_NAME, "xml-coll"));
         assertEquals(0, getDocCount(HubConfig.DEFAULT_FINAL_NAME, "xml-map"));
 
@@ -355,8 +358,9 @@ public void testInvalidQueryCollector(){
 
         resp = runFlow("testFlow", "6", UUID.randomUUID().toString(), opts, null);
         flowRunner.awaitCompletion();
-        assertEquals(JobStatus.FAILED.toString(), resp.getJobStatus(), "Since all steps failed (there was just one step), " +
-            "the status should be failed: " + resp.toJson());
+
+        assertEquals(JobStatus.STOP_ON_ERROR.toString(), resp.getJobStatus(), "Per DHFPROD-7570, since the sourceQuery failed " +
+            "and stopOnError=true, the job status should be stop-on-error");
         assertEquals(0, getDocCount(HubConfig.DEFAULT_FINAL_NAME, "xml-map"));
         stepResponse = resp.getStepResponses().get("6");
         assertEquals("failed step 6", stepResponse.getStatus());
@@ -496,7 +500,7 @@ void stopOnErrorWhenStepFailsItem() {
             .withOption("throwErrorOnPurpose", true)
         );
 
-        assertEquals("stop-on-error", response.getJobStatus());
+        assertEquals(JobStatus.STOP_ON_ERROR.toString(), response.getJobStatus());
         assertEquals("1", response.getLastAttemptedStep());
         assertEquals("0", response.getLastCompletedStep());
 
@@ -512,6 +516,35 @@ void stopOnErrorWhenStepFailsItem() {
             "The second step should not have been run since stopOnError=true");
     }
 
+    @Test
+    void stopOnErrorIsTrueAndSourceQueryThrowsError() {
+        installReferenceModelProject().createRawCustomer(1, "Jane");
+
+        RunFlowResponse response = runFlow(new FlowInputs("customHookFlow", "1", "2")
+            .withOption("stopOnError", true)
+            .withOption("sourceQuery", "cts.collectionQuery('this should throw an error")
+        );
+
+        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(0, stepResponse.getTotalEvents(), "Total events is zero because the collector failed");
+        assertEquals(0, stepResponse.getSuccessfulEvents());
+        assertEquals(0, stepResponse.getFailedEvents());
+        assertEquals(0, stepResponse.getSuccessfulBatches());
+        assertEquals(0, stepResponse.getFailedBatches());
+        assertFalse(stepResponse.isSuccess());
+        assertEquals("failed step 1", stepResponse.getStatus(), "The step status is 'failed' since the sourceQuery could not be evaluated");
+        assertEquals(1, stepResponse.getStepOutput().size());
+        assertTrue(stepResponse.getStepOutput().get(0).contains("Unable to collect items to process for flow customHookFlow and step 1"));
+        assertTrue(stepResponse.getStepOutput().get(0).contains("Invalid or unexpected token"),
+            "The stepOutput should have a stacktrace that includes the ML-generated error message");
+    }
+
     @Test
     public void testIngestBinaryAndTxt(){
         runAsDataHubOperator();
@@ -587,7 +620,7 @@ public void testStopJob() {
         final String status = resp.getJobStatus() != null ? resp.getJobStatus().toLowerCase() : "";
         assertEquals(JobStatus.CANCELED.toString().toLowerCase(), status,
             "Expected the response status to be canceled, since the job was stopped before it finished, but was instead: " + status
-         + ". If this failed in Jenkins, it likely can be ignored because we don't have a firm idea of how long the " +
+                + ". If this failed in Jenkins, it likely can be ignored because we don't have a firm idea of how long the " +
                 "thread that stops the job should wait until it stops the job; response: " + resp.toJson());
     }