Throw HTTP 400 in case of plugin errors

ssambasu · MarkLogic Builder · commit a0e2ca88b232 · 2019-09-13T01:17:11.000+05:30
diff --git a/marklogic-data-hub/src/main/java/com/marklogic/hub/flow/impl/FlowRunnerImpl.java b/marklogic-data-hub/src/main/java/com/marklogic/hub/flow/impl/FlowRunnerImpl.java
@@ -19,6 +19,7 @@
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.marklogic.client.DatabaseClient;
+import com.marklogic.client.FailedRequestException;
 import com.marklogic.client.datamovement.*;
 import com.marklogic.client.datamovement.impl.JobTicketImpl;
 import com.marklogic.client.extensions.ResourceManager;
@@ -34,7 +35,9 @@
 import com.marklogic.hub.job.Job;
 import com.marklogic.hub.job.JobManager;
 import com.marklogic.hub.job.JobStatus;
+import org.apache.commons.lang3.StringUtils;
 
+import java.io.IOException;
 import java.io.PrintWriter;
 import java.io.StringWriter;
 import java.util.*;
@@ -369,7 +372,8 @@ public RunFlowResponse run(String jobId, String[] items) {
         }
 
         public RunFlowResponse run(String jobId, String[] items, Map<String, Object> options) {
-            RunFlowResponse resp;
+            RunFlowResponse resp = null;
+            ObjectMapper objectMapper = new ObjectMapper();
             try {
                 RequestParameters params = new RequestParameters();
                 params.add("entity-name", flow.getEntityName());
@@ -378,7 +382,6 @@ public RunFlowResponse run(String jobId, String[] items, Map<String, Object> opt
                 params.put("identifiers", items);
                 params.put("target-database", targetDatabase);
                 if (options != null) {
-                    ObjectMapper objectMapper = new ObjectMapper();
                     params.put("options", objectMapper.writeValueAsString(options));
                 }
                 ResourceServices.ServiceResultIterator resultItr = this.getServices().post(params, new StringHandle("{}").withFormat(Format.JSON));
@@ -389,7 +392,6 @@ public RunFlowResponse run(String jobId, String[] items, Map<String, Object> opt
                     else {
                         ResourceServices.ServiceResult res = resultItr.next();
                         StringHandle handle = new StringHandle();
-                        ObjectMapper objectMapper = new ObjectMapper();
                         resp = objectMapper.readValue(res.getContent(handle).get(), RunFlowResponse.class);
                     }
                 }
@@ -400,10 +402,25 @@ public RunFlowResponse run(String jobId, String[] items, Map<String, Object> opt
                 }
             }
             catch(Exception e) {
-                e.printStackTrace();
-                throw new RuntimeException(e);
+                resp = handleFlowRunnerException (e);
             }
             return resp;
         }
     }
+
+    protected RunFlowResponse handleFlowRunnerException (Exception e) {
+        ObjectMapper objectMapper = new ObjectMapper();
+        RunFlowResponse resp = null;
+        if(e instanceof FailedRequestException && StringUtils.containsIgnoreCase(((FailedRequestException) e).getFailedRequest().getStatus(), "Plugin error")){
+            try {
+                resp = objectMapper.readValue(((FailedRequestException)e).getFailedRequest().getMessage(), RunFlowResponse.class);
+            } catch (IOException ex) {
+                throw new RuntimeException("Unexpected IO error while parsing exception from running flow; original exception: " + e.getMessage());
+            }
+        }
+        else{
+            throw new RuntimeException(e);
+        }
+        return resp;
+    }
 }
diff --git a/marklogic-data-hub/src/main/resources/ml-modules/root/data-hub/4/extensions/flow.xqy b/marklogic-data-hub/src/main/resources/ml-modules/root/data-hub/4/extensions/flow.xqy
@@ -157,7 +157,7 @@ declare function post(
             then trace:error-trace($item-context, $batch-error, xdmp:elapsed-time() - $before)
             else trace:error-trace($item-context, $ex, xdmp:elapsed-time() - $before)
           }
-        let $resp :=
+        let $output :=
           document {
             object-node {
               "totalCount": fn:count($identifiers),
@@ -167,6 +167,11 @@ declare function post(
               "errors": $errors
             }
           }
+        let $resp :=
+          if (trace:get-error-count() > 0) then
+            fn:error((),"RESTAPI-SRVEXERR", (400, "Plugin error", text{$output}))
+          else
+            $output
         return
           $resp
       else
diff --git a/marklogic-data-hub/src/main/resources/ml-modules/root/data-hub/4/extensions/sjsflow.sjs b/marklogic-data-hub/src/main/resources/ml-modules/root/data-hub/4/extensions/sjsflow.sjs
@@ -120,6 +120,9 @@ function post(context, params, input) {
       "failedItems": tracelib.getFailedItems(),
       "errors": errors
     }
+    if(resp.errorCount > 0){
+      fn.error(null, "RESTAPI-SRVEXERR", Sequence.from([400, "Plugin error", resp]));
+    }
   }
   else {
     resp = 'error';
diff --git a/marklogic-data-hub/src/test/java/com/marklogic/hub/flow/impl/FlowRunnerImplTest.java b/marklogic-data-hub/src/test/java/com/marklogic/hub/flow/impl/FlowRunnerImplTest.java
@@ -0,0 +1,78 @@
+package com.marklogic.hub.flow.impl;
+
+import com.marklogic.bootstrap.Installer;
+import com.marklogic.client.FailedRequestException;
+import com.marklogic.client.impl.FailedRequest;
+import com.marklogic.hub.ApplicationConfig;
+import com.marklogic.hub.HubTestBase;
+import com.marklogic.hub.flow.RunFlowResponse;
+import org.custommonkey.xmlunit.XMLUnit;
+
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit.jupiter.SpringExtension;
+
+import java.io.IOException;
+@ExtendWith(SpringExtension.class)
+@ContextConfiguration(classes = ApplicationConfig.class)
+class FlowRunnerImplTest extends HubTestBase {
+    @BeforeEach
+    public void setup() throws IOException {
+        XMLUnit.setIgnoreWhitespace(true);
+        createProjectDir();
+        getDataHubAdminConfig();
+    }
+
+    @AfterAll
+    public static void cleanup() throws IOException {
+        new Installer().teardownProject();
+    }
+    @Test
+    public void testFlowRunnerException() {
+        RunFlowResponse resp = null;
+        FlowRunnerImpl fr = new FlowRunnerImpl(adminHubConfig);
+        FailedRequest req = new FailedRequest();
+        req.setMessageCode("RESTAPI-SRVEXERR");
+        req.setMessageString("{\n" +
+            "\t\"totalCount\": 3,\n" +
+            "\t\"errorCount\": 3,\n" +
+            "\t\"completedItems\": [],\n" +
+            "\t\"failedItems\": [\"/10287.json\", \"/10310.json\", \"/10328.json\"],\n" +
+            "\t\"errors\": []\n" +
+            "}");
+        req.setStatusString("Plugin error");
+
+       resp = fr.handleFlowRunnerException(new FailedRequestException("failed to apply resource at resources/ml:flow: Plugin error", req));
+       Assertions.assertEquals(3, resp.errorCount);
+       Assertions.assertEquals(3, resp.totalCount);
+    }
+
+    @Test
+    public void testFlowRunnerExceptionWithoutPluginError() {
+        RunFlowResponse resp = null;
+        FlowRunnerImpl fr = new FlowRunnerImpl(adminHubConfig);
+        FailedRequest req = new FailedRequest();
+        req.setMessageCode("RESTAPI-SRVEXERR");
+        req.setMessageString("{\n" +
+            "\t\"totalCount\": 3,\n" +
+            "\t\"errorCount\": 3,\n" +
+            "\t\"completedItems\": [],\n" +
+            "\t\"failedItems\": [\"/10287.json\", \"/10310.json\", \"/10328.json\"],\n" +
+            "\t\"errors\": []\n" +
+            "}");
+        req.setStatusString("Failed");
+        try {
+            fr.handleFlowRunnerException(new FailedRequestException("Not a 'Plugin error' exception", req));
+        }
+        catch (Exception e) {
+            Assertions.assertTrue(e instanceof  RuntimeException);
+            e.printStackTrace();
+            System.out.println(e.getMessage());
+            Assertions.assertTrue(e.getMessage().contains("com.marklogic.client.FailedRequestException: Local message: Not a 'Plugin error' exception"));
+        }
+    }
+}

Original file line number	Diff line number	Diff line change
`@@ -120,6 +120,9 @@ function post(context, params, input) {`
`120`	`120`	`"failedItems": tracelib.getFailedItems(),`
`121`	`121`	`"errors": errors`
`122`	`122`	`}`
	`123`	`+ if(resp.errorCount > 0){`
	`124`	`+ fn.error(null, "RESTAPI-SRVEXERR", Sequence.from([400, "Plugin error", resp]));`
	`125`	`+ }`
`123`	`126`	`}`
`124`	`127`	`else {`
`125`	`128`	`resp = 'error';`