Merging 1.0 fixes into 2.0

paxtonhare · paxtonhare · commit 0913016f61d3 · 2016-12-07T13:43:01.000-05:00
- fixes for #336
- fixes for #331
diff --git a/marklogic-data-hub/src/main/resources/ml-modules/services/flow.xqy b/marklogic-data-hub/src/main/resources/ml-modules/services/flow.xqy
@@ -98,7 +98,11 @@ declare function delete(
   $params  as map:map
   ) as document-node()?
 {
-  let $_ := flow:invalidate-flow-caches()
-  return
-    document { () }
+  debug:dump-env("INVALIDATE FLOW CACHES"),
+
+  perf:log('/v1/resources/flow:delete', function() {
+    let $_ := flow:invalidate-flow-caches()
+    return
+      document { () }
+  })
 };
diff --git a/quick-start/src/main/java/com/marklogic/quickstart/Application.java b/quick-start/src/main/java/com/marklogic/quickstart/Application.java
@@ -15,6 +15,8 @@
  */
 package com.marklogic.quickstart;
 
+import com.marklogic.contentpump.ContentPump;
+import org.apache.commons.lang.ArrayUtils;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.boot.builder.SpringApplicationBuilder;
@@ -31,7 +33,13 @@ protected SpringApplicationBuilder configure(SpringApplicationBuilder applicatio
         return application.sources(Application.class);
     }
 
-    public static void main(String[] args) {
-        SpringApplication.run(Application.class, args);
+    public static void main(String[] args) throws Exception {
+        if (args.length > 0 && args[0].equals("mlcp")) {
+            String[] newArgs = (String[]) ArrayUtils.subarray(args, 1, args.length);
+            ContentPump.main(newArgs);
+        }
+        else {
+            SpringApplication.run(Application.class, args);
+        }
     }
 }
diff --git a/quick-start/src/main/java/com/marklogic/quickstart/service/FileSystemEventListener.java b/quick-start/src/main/java/com/marklogic/quickstart/service/FileSystemEventListener.java
@@ -9,5 +9,5 @@
 @Component
 public interface FileSystemEventListener {
 
-    void onWatchEvent(HubConfig hubConfig, Path path);
+    void onWatchEvent(HubConfig hubConfig);
 }
diff --git a/quick-start/src/main/java/com/marklogic/quickstart/service/FileSystemWatcherService.java b/quick-start/src/main/java/com/marklogic/quickstart/service/FileSystemWatcherService.java
@@ -64,12 +64,12 @@ public void removeListener(FileSystemEventListener listener) {
         listeners.remove(listener);
     }
 
-    private void notifyListeners(HubConfig hubConfig, Path path) {
+    private void notifyListeners(HubConfig hubConfig) {
         // notify global listeners
         synchronized (listeners) {
             for (FileSystemEventListener listener : listeners) {
                 try {
-                    listener.onWatchEvent(hubConfig, path);
+                    listener.onWatchEvent(hubConfig);
                 }
                 catch (Exception e) {
                     logger.error("Exception occured on listener", e);
@@ -146,50 +146,28 @@ public synchronized void destroy() throws Exception {
     private class DirectoryWatcherThread extends Thread {
 
         private HubConfig hubConfig;
-        private final int DELAY = 500;
+        private final int DELAY = 1000;
 
-        // Use a SET to prevent duplicates from being added when multiple events on the
-        // same file arrive in quick succession.
-        HashSet<Path> filesToReload = new HashSet<>();
         Timer processDelayTimer = null;
 
         DirectoryWatcherThread(String name, HubConfig hubConfig) {
             super(name);
             this.hubConfig = hubConfig;
         }
 
-        private synchronized void addFileToProcess(Path path) {
-            boolean alreadyAdded = !filesToReload.add(path);
-            logger.info("Queuing file for processing: "
-                + path.toString() + (alreadyAdded?"(already queued)":""));
+        private synchronized void queueReload() {
             if (processDelayTimer != null) {
                 processDelayTimer.cancel();
             }
             processDelayTimer = new Timer();
             processDelayTimer.schedule(new TimerTask() {
                 @Override
                 public void run() {
-                    processFiles();
+                    notifyListeners(hubConfig);
                 }
             }, DELAY);
         }
 
-        private synchronized void processFiles() {
-            // Iterate over the set of file to be processed
-            for (Iterator<Path> it = filesToReload.iterator(); it.hasNext();) {
-                Path path = it.next();
-
-                // Sometimes you just have to do what you have to do...
-                logger.info("Processing file: " + path.toString());
-
-                // notify listeners
-                notifyListeners(hubConfig, path);
-
-                // Remove this file from the set.
-                it.remove();
-            }
-        }
-
         @Override
         public void run() {
             for (;;) {
@@ -220,8 +198,8 @@ public void run() {
                     Path child = dir.resolve(context);
 
                     // print out event
-                    logger.info("Event received: {} for: {}", event.kind().name(), child);
-                    addFileToProcess(child);
+                    logger.debug("Event received: {} for: {}", event.kind().name(), child);
+                    queueReload();
 
                     // if directory is created, then register it and its sub-directories
                     // we are always listening recursively
diff --git a/quick-start/src/main/java/com/marklogic/quickstart/service/MlcpTasklet.java b/quick-start/src/main/java/com/marklogic/quickstart/service/MlcpTasklet.java
@@ -2,25 +2,30 @@
 
 import com.fasterxml.jackson.databind.JsonNode;
 import com.fasterxml.jackson.databind.ObjectMapper;
-import com.marklogic.contentpump.ContentPump;
 import com.marklogic.contentpump.bean.MlcpBean;
 import com.marklogic.hub.HubConfig;
 import com.marklogic.hub.JobStatusListener;
 import com.marklogic.quickstart.util.StreamGobbler;
 import org.apache.commons.io.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.batch.core.StepContribution;
 import org.springframework.batch.core.scope.context.ChunkContext;
 import org.springframework.batch.core.step.tasklet.Tasklet;
 import org.springframework.batch.repeat.RepeatStatus;
 
 import java.io.File;
 import java.io.IOException;
+import java.net.URL;
+import java.net.URLClassLoader;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.function.Consumer;
 
 class MlcpTasklet implements Tasklet {
 
+    protected final Logger logger = LoggerFactory.getLogger(getClass());
+
     private HubConfig hubConfig;
     private JsonNode mlcpOptions;
     private JobStatusListener statusListener;
@@ -66,34 +71,53 @@ public RepeatStatus execute(StepContribution contribution, ChunkContext chunkCon
     }
 
     private void runMlcp(long jobId, MlcpBean bean) throws IOException, InterruptedException {
+        ClassLoader cl = ClassLoader.getSystemClassLoader();
+
+        URL[] urls = ((URLClassLoader)cl).getURLs();
+
         String javaHome = System.getProperty("java.home");
         String javaBin = javaHome +
             File.separator + "bin" +
             File.separator + "java";
         String classpath = System.getProperty("java.class.path");
-        String className = ContentPump.class.getCanonicalName();
-
-        File loggerFile = File.createTempFile("mlcp-", "-logger");
-        String loggerData = "log4j.rootLogger=INFO,console\n" +
-            "log4j.appender.console=org.apache.log4j.ConsoleAppender\n" +
-            "log4j.appender.console.target=System.err\n" +
-            "log4j.appender.console.layout=org.apache.log4j.PatternLayout\n" +
-            "log4j.appender.console.layout.ConversionPattern=%d{yy/MM/dd HH:mm:ss} %p %c{2}: %m%n\n" +
+
+        File loggerFile = File.createTempFile("mlcp-", "-logger.xml");
+        String loggerData = "<configuration>\n" +
             "\n" +
-            "# To suppress not native warn on Mac and Solaris\n" +
-            "log4j.logger.org.apache.hadoop.util.NativeCodeLoader=ERROR\n" +
+            "  <appender name=\"STDOUT\" class=\"ch.qos.logback.core.ConsoleAppender\">\n" +
+            "    <!-- encoders are assigned the type\n" +
+            "         ch.qos.logback.classic.encoder.PatternLayoutEncoder by default -->\n" +
+            "    <encoder>\n" +
+            "      <pattern>%d{HH:mm:ss.SSS} [%thread] %-5level %logger{36} - %msg%n</pattern>\n" +
+            "    </encoder>\n" +
+            "  </appender>\n" +
             "\n" +
-            "# To enable debug\n" +
-            "#log4j.logger.com.marklogic.mapreduce=DEBUG\n" +
-            "#log4j.logger.com.marklogic.contentpump=DEBUG\n";
+            "  <logger name=\"org.apache.http\" level=\"WARN\"/>\n" +
+            "\n" +
+            "  <logger name=\"com.marklogic.spring.batch.core.repository.dao.MarkLogicStepExecutionDao\" level=\"WARN\"/>\n" +
+            "  <logger name=\"com.marklogic.spring.batch.core.repository.dao.MarkLogicJobExecutionDao\" level=\"WARN\"/>\n" +
+            "  <logger name=\"com.marklogic.client.impl.DocumentManagerImpl\" level=\"WARN\"/>\n" +
+            "  <logger name=\"com.marklogic.client.impl.DatabaseClientImpl\" level=\"WARN\"/>\n" +
+            "  <logger name=\"com.marklogic\" level=\"INFO\"/>\n" +
+            "  <logger name=\"com.marklogic.appdeployer\" level=\"INFO\"/>\n" +
+            "  <logger name=\"com.marklogic.hub\" level=\"INFO\"/>\n" +
+            "  <logger name=\"com.marklogic.contentpump\" level=\"INFO\"/>\n" +
+            "  <logger name=\"org.apache.catalina.webresources.Cache\" level=\"ERROR\"/>\n" +
+            "  <logger name=\"org.apache.hadoop.util.Shell\" level=\"OFF\"/>\n" +
+            "  <logger name=\"org.apache.hadoop.util.NativeCodeLoader\" level=\"ERROR\"/>\n" +
+            "\n" +
+            "  <root level=\"WARN\">\n" +
+            "    <appender-ref ref=\"STDOUT\" />\n" +
+            "  </root>\n" +
+            "</configuration>\n";
         FileUtils.writeStringToFile(loggerFile, loggerData);
 
         ArrayList<String> args = new ArrayList<>();
         args.add(javaBin);
-        args.add("-Dlog4j.configurationFile=" + loggerFile.getAbsolutePath());
-        args.add("-cp");
+        args.add("-Dlogback.configurationFile=" + loggerFile.toURI());//logback);
+        args.add("-jar");
         args.add(classpath);
-        args.add(className);
+        args.add("mlcp");
         args.addAll(Arrays.asList(bean.buildArgs()));
 
         ProcessBuilder pb = new ProcessBuilder(args);
@@ -104,7 +128,7 @@ private void runMlcp(long jobId, MlcpBean bean) throws IOException, InterruptedE
 
             @Override
             public void accept(String status) {
-
+                System.out.println(status);
                 // don't log an error if the winutils binary is missing
                 if (status.contains("ERROR") && !status.contains("winutils binary")) {
                     hasError = true;
diff --git a/quick-start/src/main/java/com/marklogic/quickstart/util/StreamGobbler.java b/quick-start/src/main/java/com/marklogic/quickstart/util/StreamGobbler.java
@@ -16,7 +16,6 @@ public StreamGobbler(InputStream inputStream, Consumer<String> consumeInputLine)
 
     @Override
     public void run() {
-        System.out.println("gobbler running");
         new BufferedReader(new InputStreamReader(inputStream)).lines().forEach(consumeInputLine);
     }
 }
diff --git a/quick-start/src/main/java/com/marklogic/quickstart/web/ProjectsController.java b/quick-start/src/main/java/com/marklogic/quickstart/web/ProjectsController.java
@@ -352,8 +352,13 @@ public void onValidate(JsonNode validation) {
         template.convertAndSend("/topic/validate-status", validation);
     }
 
+    /**
+     * Called when the filesystem watcher detects a file change. We then install the user modules
+     * @param hubConfig - must pass the hub config because this runs in a separate thread and doesn't
+     *                  have access to the current spring boot context
+     */
     @Override
-    public void onWatchEvent(HubConfig hubConfig, Path path) {
+    public void onWatchEvent(HubConfig hubConfig) {
         installUserModules(hubConfig, false);
     }
 
diff --git a/quick-start/src/main/resources/log4j.properties b/quick-start/src/main/resources/log4j.properties
@@ -1 +1,4 @@
 log4j.logger.httpclient.wire=DEBUG
+log4j.logger.org.apache.hadoop.util.NativeCodeLoader=ERROR
+log4j.logger.com.marklogic.mapreduce=ERROR
+log4j.logger.com.marklogic.contentpump=ERROR
diff --git a/quick-start/src/main/resources/logback-spring.xml b/quick-start/src/main/resources/logback-spring.xml
@@ -19,7 +19,9 @@
   <logger name="com.marklogic.appdeployer" level="INFO"/>
   <logger name="com.marklogic.hub" level="INFO"/>
   <logger name="com.marklogic.contentpump" level="INFO"/>
+  <logger name="com.marklogic.mapreduce" level="INFO"/>
   <logger name="org.apache.catalina.webresources.Cache" level="SEVERE"/>
+  <logger name="org.apache.hadoop.util.NativeCodeLoader" level="SEVERE"/>
 
   <root level="WARN">
     <appender-ref ref="STDOUT" />

Original file line number	Diff line number	Diff line change
`@@ -9,5 +9,5 @@`
`9`	`9`	`@Component`
`10`	`10`	`public interface FileSystemEventListener {`
`11`	`11`
`12`		`- void onWatchEvent(HubConfig hubConfig, Path path);`
	`12`	`+ void onWatchEvent(HubConfig hubConfig);`
`13`	`13`	`}`
Original file line number	Diff line number	Diff line change
`@@ -16,7 +16,6 @@ public StreamGobbler(InputStream inputStream, Consumer<String> consumeInputLine)`
`16`	`16`
`17`	`17`	`@Override`
`18`	`18`	`public void run() {`
`19`		`- System.out.println("gobbler running");`
`20`	`19`	`new BufferedReader(new InputStreamReader(inputStream)).lines().forEach(consumeInputLine);`
`21`	`20`	`}`
`22`	`21`	`}`
Original file line number	Diff line number	Diff line change
`@@ -352,8 +352,13 @@ public void onValidate(JsonNode validation) {`
`352`	`352`	`template.convertAndSend("/topic/validate-status", validation);`
`353`	`353`	`}`
`354`	`354`
	`355`	`+ /**`
	`356`	`+ * Called when the filesystem watcher detects a file change. We then install the user modules`
	`357`	`+ * @param hubConfig - must pass the hub config because this runs in a separate thread and doesn't`
	`358`	`+ * have access to the current spring boot context`
	`359`	`+ */`
`355`	`360`	`@Override`
`356`		`- public void onWatchEvent(HubConfig hubConfig, Path path) {`
	`361`	`+ public void onWatchEvent(HubConfig hubConfig) {`
`357`	`362`	`installUserModules(hubConfig, false);`
`358`	`363`	`}`
`359`	`364`