added work around for detecting MLCP errors

- work around is to set System.out to our own OutputStream before running MLCP

Author: gmarintes

marklogic-data-hub/src/main/java/com/marklogic/hub/DataHubContentPump.java

@@ -1,11 +1,25 @@
 package com.marklogic.hub;
 
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.PrintStream;
 import java.util.List;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+import java.util.regex.PatternSyntaxException;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import com.marklogic.contentpump.ContentPump;
 import com.marklogic.contentpump.utilities.OptionsFileUtil;
 
 public class DataHubContentPump extends ContentPump {
+    private final static Logger LOGGER = LoggerFactory.getLogger(DataHubContentPump.class);
 
     private String[] arguments;
 
@@ -21,19 +35,80 @@ public DataHubContentPump(String[] arguments) {
      * Run the Content Pump.
      *
      * @return true if the content pump executed successfully, false otherwise.
+     * @throws IOException 
      */
-    public boolean execute() {
+    public void execute() throws IOException {
         String[] expandedArgs = null;
-        int rc = 1;
+        
+        PrintStream sysout = System.out;
+        PrintStream mlcpOutputStream = null;
+        BufferedReader mlcpBufferedReader = null;
+        File mlcpOutputFile = null;
         try {
+            // redirect standard output
+            mlcpOutputFile = File.createTempFile("mlcp", ".txt");
+            mlcpOutputStream = new PrintStream(mlcpOutputFile);
+            System.setOut(mlcpOutputStream);
+            
+            // run mlcp
             expandedArgs = OptionsFileUtil.expandArguments(arguments);
-            rc = runCommand(expandedArgs);
+            runCommand(expandedArgs);
         } catch (Exception ex) {
             LOG.error("Error while expanding arguments", ex);
             System.err.println(ex.getMessage());
             System.err.println("Try 'mlcp help' for usage.");
+        } finally {
+            // close the mlcp output stream
+            if (mlcpOutputStream != null) {
+                mlcpOutputStream.close();
+            }
+            
+            // revert to the original standard output
+            System.setOut(sysout);
         }
+        
+        // read the mlcp output and get any error message
+        StringBuilder errorMessage = new StringBuilder();
+        try {
+            String regex = "([^\\s]*) \\s (\\[ [^ \\] ]*? \\]) \\s ([^\\s]*) \\s ([^\\s]*) \\s - \\s (.*)";
+            Pattern pattern = Pattern.compile(regex, Pattern.COMMENTS);
 
-        return rc == 0;
+            mlcpBufferedReader = new BufferedReader(new InputStreamReader(new FileInputStream(mlcpOutputFile)));
+            String line = null;
+            while ((line = mlcpBufferedReader.readLine()) != null) {
+                Matcher matcher = pattern.matcher(line);
+                if (matcher.matches()) {
+                    String logLevel = matcher.groupCount() >= 3 ? matcher.group(3) : "";
+                    String message = matcher.groupCount() >= 5 ? matcher.group(5) : "";
+                    if (logLevel.toLowerCase().equals("error")) {
+                        if (errorMessage.length() > 0) {
+                            errorMessage.append("\r\n");
+                        }
+                        errorMessage.append(message);
+                    }
+                }
+            }
+        } catch (PatternSyntaxException e) {
+            LOGGER.error("Unexpected error", e);
+        } catch (FileNotFoundException e) {
+            LOGGER.error("Unexpected error", e);
+        } catch (IOException e) {
+            LOGGER.error("Unexpected error", e);
+        } finally {
+            if (mlcpBufferedReader != null) {
+                try {
+                    mlcpBufferedReader.close();
+                } catch (IOException e) {
+                    // intentionally empty
+                }
+            }
+            
+            // delete the temporary file
+            mlcpOutputFile.delete();
+        }
+        
+        if (errorMessage.length() > 0) {
+            throw new IOException("Load data failed with:\r\n" + errorMessage.toString());
+        }
     }
 }

marklogic-data-hub/src/main/java/com/marklogic/hub/Mlcp.java

@@ -52,8 +52,7 @@ public void addSourceDirectory(String directoryPath, SourceOptions options) {
         sources.add(source);
     }
 
-    public boolean loadContent() {
-        boolean success = true;
+    public void loadContent() throws IOException {
         for (MlcpSource source : sources) {
             try {
                 List<String> arguments = new ArrayList<>();
@@ -75,13 +74,11 @@ public boolean loadContent() {
                 arguments.addAll(sourceArguments);
 
                 DataHubContentPump contentPump = new DataHubContentPump(arguments);
-                success = success && contentPump.execute();
-            }
-            catch (Exception e) {
-                LOGGER.error("Failed to load {}", source.getSourcePath(), e);
+                contentPump.execute();
+            } catch (IOException e) {
+                throw new IOException("Cannot load data from: " + source.getSourcePath() + " due to: " + e.getMessage());
             }
         }
-        return success;
     }
 
     protected void setHadoopHomeDir() throws IOException {

quick-start/src/main/resources/static/top/topController.js

@@ -65,7 +65,12 @@
           TaskManager.waitForTask(flow.inputFlowTaskId)
           .success(function (result) {
             if (!flow.inputFlowCancelled) {
-              DataHub.displayMessage('Load data successful.', 'success', 'notification', false);
+              if (result.success) {
+                DataHub.displayMessage('Load data successful.', 'success', 'notification', false);
+              }
+              else {
+                DataHub.displayMessage(result.errorMessage, 'error', 'notification', false);
+              }
             }
           })
           .error(function () {