Merge branch 'feature/speed-it-up' into develop

Author: paxtonhare

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

@@ -146,9 +146,10 @@ private ConfigurableApplicationContext buildApplicationContext(Flow flow, JobSta
         return ctx;
     }
 
-    private JobParameters buildJobParameters(Flow flow, int batchSize) {
+    private JobParameters buildJobParameters(Flow flow, int batchSize, int threadCount) {
         JobParametersBuilder jpb = new JobParametersBuilder();
-        jpb.addLong("chunk", Integer.toUnsignedLong(batchSize));
+        jpb.addLong("batchSize", Integer.toUnsignedLong(batchSize));
+        jpb.addLong("threadCount", Integer.toUnsignedLong(threadCount));
         jpb.addString("uid", UUID.randomUUID().toString());
         jpb.addString("flowType", flow.getType().toString());
         jpb.addString("entity", flow.getEntityName());
@@ -163,12 +164,12 @@ private JobParameters buildJobParameters(Flow flow, int batchSize) {
      * @param statusListener - the callback to receive job status updates
      * @return a JobExecution instance
      */
-    public JobExecution runFlow(Flow flow, int batchSize, JobStatusListener statusListener) {
+    public JobExecution runFlow(Flow flow, int batchSize, int threadCount, JobStatusListener statusListener) {
         JobExecution result = null;
         try {
             ConfigurableApplicationContext ctx = buildApplicationContext(flow, statusListener);
 
-            JobParameters params = buildJobParameters(flow, batchSize);
+            JobParameters params = buildJobParameters(flow, batchSize, threadCount);
             JobLauncher launcher = ctx.getBean(JobLauncher.class);
             Job job = ctx.getBean(Job.class);
             result = launcher.run(job, params);

marklogic-data-hub/src/main/java/com/marklogic/spring/batch/hub/FlowWriter.java

@@ -2,6 +2,8 @@
 
 import java.util.List;
 
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import org.springframework.batch.item.ItemWriter;
 
 import com.marklogic.client.DatabaseClient;
@@ -18,36 +20,28 @@ public class FlowWriter extends ResourceManager  implements ItemWriter<String> {
 
     private DatabaseClient client;
     private Flow flow;
+    StringHandle handle;
+    RequestParameters params = new RequestParameters();
 
     public FlowWriter(DatabaseClient client, Flow flow) {
         super();
         this.flow = flow;
         this.client = client;
         this.client.init(NAME, this);
+        handle = new StringHandle(flow.serialize(true));
+        handle.setFormat(Format.XML);
     }
 
     @Override
     public void write(List<? extends String> items) {
 
-        Transaction transaction = null;
         try {
-            transaction = client.openTransaction();
-            for (String id: items) {
-                RequestParameters params = new RequestParameters();
-                params.add("identifier", id);
-
-                StringHandle handle = new StringHandle(flow.serialize(true));
-                handle.setFormat(Format.XML);
-                this.getServices().post(params, handle, transaction);
-            }
-            transaction.commit();
+            params.put("identifier", items.toArray(new String[items.size()]));
+            this.getServices().post(params, handle);
         }
         catch(Exception e) {
-          if (transaction != null) {
-              transaction.rollback();
-          }
             e.printStackTrace();
-          throw e;
-      }
+            throw e;
+        }
     }
 }

marklogic-data-hub/src/main/java/com/marklogic/spring/batch/hub/HubJobRunner.java

@@ -11,7 +11,6 @@
 import org.springframework.batch.core.launch.JobLauncher;
 import org.springframework.context.ConfigurableApplicationContext;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
-import org.springframework.core.env.JOptCommandLinePropertySource;
 
 import java.util.Arrays;
 import java.util.Date;

marklogic-data-hub/src/main/java/com/marklogic/spring/batch/hub/RunHarmonizeFlowConfig.java

@@ -11,8 +11,10 @@
 import org.springframework.batch.item.support.PassThroughItemProcessor;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.beans.factory.annotation.Qualifier;
+import org.springframework.beans.factory.annotation.Value;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
+import org.springframework.core.task.SimpleAsyncTaskExecutor;
 
 import java.util.List;
 
@@ -48,17 +50,19 @@ public void afterJob(JobExecution jobExecution) {
 
     @Bean
     @JobScope
-    protected Step step1() {
+    protected Step step1(@Value("#{jobParameters['batchSize']}") int batchSize, @Value("#{jobParameters['threadCount']}") int threadCount) {
 
         Collector c = flow.getCollector();
         if (c instanceof ServerCollector) {
             ((ServerCollector)c).setClient(getDatabaseClient());
         }
 
         ItemProcessor<String, String> ip = new PassThroughItemProcessor<>();
+        SimpleAsyncTaskExecutor taskExecutor = new SimpleAsyncTaskExecutor();
+        taskExecutor.setConcurrencyLimit(threadCount);
 
         return stepBuilderFactory.get("step1")
-            .<String, String>chunk(getChunkSize())
+            .<String, String>chunk(batchSize)
             .reader(new CollectorReader(c))
             .processor(ip)
             .listener(new ItemWriteListener<String>() {
@@ -101,6 +105,7 @@ public void afterChunkError(ChunkContext context) {
 
                 }
             })
+            .taskExecutor(taskExecutor)
             .build();
     }
 }

marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.hub/lib/debug-lib.xqy

@@ -80,8 +80,9 @@ declare function debug:dump-env($name as xs:string?)
     if ($request-path = '/MarkLogic/rest-api/endpoints/resource-service-query.xqy') then
       let $params := fn:string-join(
         for $f in xdmp:get-request-field-names()[fn:starts-with(., "rs:")]
+        let $value := xdmp:get-request-field($f)
         return
-          $f || "=" || xdmp:get-request-field($f),
+          $f || "=" || fn:string-join($value, ", "),
       "&")
       return
         "/v1/resources/" || xdmp:get-request-field("name") || "?" || $params
@@ -108,9 +109,9 @@ declare function debug:dump-env($name as xs:string?)
         "    " || $h || " => " || xdmp:get-request-header($h),
       "",
       "  [Request Params]",
-      for $p in xdmp:get-request-field-names()
+      for $p in xdmp:get-request-field-names()[fn:not(fn:starts-with(., "rs:"))]
       return
-        "    " || $p || " => " || xdmp:get-request-field($p),
+        "    " || $p || " => " || fn:string-join(xdmp:get-request-field($p), ", "),
       let $body := xdmp:get-request-body()
       return
         if (fn:exists($body)) then

marklogic-data-hub/src/main/resources/ml-modules/root/com.marklogic.hub/lib/trace-lib.xqy

@@ -33,10 +33,15 @@ declare variable $FORMAT-JSON := "json";
 
 declare %private variable $current-trace-settings := map:map();
 
-declare %private variable $current-trace := map:new((
-  map:entry("traceId", xdmp:random()),
-  map:entry("created", fn:current-dateTime())
-));
+declare %private variable $current-trace := trace:new-trace();
+
+declare function trace:new-trace()
+{
+  map:new((
+    map:entry("traceId", xdmp:random()),
+    map:entry("created", fn:current-dateTime())
+  ))
+};
 
 declare function trace:enable-tracing($enabled as xs:boolean)
 {
@@ -77,7 +82,7 @@ declare function trace:init-trace($format as xs:string)
 
 declare function trace:write-trace()
 {
-  if (trace:enabled() or trace:has-errors()) then
+  if (trace:enabled() or trace:has-errors()) then (
     let $format := map:get($current-trace-settings, "data-format")
     let $trace :=
       if ($format eq $FORMAT-JSON) then
@@ -145,7 +150,9 @@ declare function trace:write-trace()
       map:new((
         map:entry("database", xdmp:database($config:TRACING-DATABASE)),
         map:entry("transactionMode", "update-auto-commit")
-      )))
+      ))),
+    xdmp:set($current-trace, trace:new-trace())
+  )
   else ()
 };
 

marklogic-data-hub/src/main/resources/ml-modules/services/flow.xqy

@@ -77,12 +77,12 @@ declare %rapi:transaction-mode("update") function post(
 
   perf:log('/v1/resources/flow:post', function() {
     let $flow as element(hub:flow) := $input/hub:flow
-    let $identifier := map:get($params, "identifier")
     let $options := map:new((
       map:entry("entity", $flow/hub:entity/fn:data()),
       map:entry("flow", $flow/hub:name/fn:data()),
       map:entry("flowType", $flow/hub:type/fn:data())
     ))
+    for $identifier in map:get($params, "identifier")
     let $_ := flow:run-flow($flow, $identifier, $options)
     return
       document { () }

marklogic-data-hub/src/test/java/com/marklogic/hub/EndToEndTestSjsJson.java

@@ -66,7 +66,7 @@ public void runFlows() throws IOException, ParserConfigurationException, SAXExce
         stagingDocMgr.write("/input.json", new JacksonHandle(getJsonFromResource("e2e-test/staged.json")));
 
         JobFinishedListener harmonizeFlowListener = new JobFinishedListener();
-        fm.runFlow(harmonizeFlow, 10, harmonizeFlowListener);
+        fm.runFlow(harmonizeFlow, 10, 1, harmonizeFlowListener);
         harmonizeFlowListener.waitForFinish();
         String expected = getResource("e2e-test/final.json");
         String actual = finalDocMgr.read("/input.json").next().getContent(new StringHandle()).get();

marklogic-data-hub/src/test/java/com/marklogic/hub/EndToEndTestSjsXml.java

@@ -70,7 +70,7 @@ public void runFlows() throws IOException, ParserConfigurationException, SAXExce
         stagingDocMgr.write("/input.xml", new DOMHandle(getXmlFromResource("e2e-test/staged.xml")));
 
         JobFinishedListener harmonizeFlowListener = new JobFinishedListener();
-        fm.runFlow(harmonizeFlow, 10, harmonizeFlowListener);
+        fm.runFlow(harmonizeFlow, 10, 1, harmonizeFlowListener);
         harmonizeFlowListener.waitForFinish();
         assertXMLEqual(getXmlFromResource("e2e-test/final.xml"), finalDocMgr.read("/input.xml").next().getContent(new DOMHandle()).get());
     }

marklogic-data-hub/src/test/java/com/marklogic/hub/EndToEndTestXqyJson.java

@@ -65,7 +65,7 @@ public void runFlowwithTriplesNodeStar() throws IOException, ParserConfiguration
         stagingDocMgr.write("/input.json", new JacksonHandle(getJsonFromResource("e2e-test/staged.json")));
 
         JobFinishedListener harmonizeFlowListener = new JobFinishedListener();
-        fm.runFlow(harmonizeFlow, 10, harmonizeFlowListener);
+        fm.runFlow(harmonizeFlow, 10, 1, harmonizeFlowListener);
         harmonizeFlowListener.waitForFinish();
         String expected = getResource("e2e-test/final.json");
         String actual = finalDocMgr.read("/input.json").next().getContent(new StringHandle()).get();
@@ -82,7 +82,7 @@ public void runFlowWithTriplesJsonArray() throws IOException, ParserConfiguratio
         stagingDocMgr.write("/input.json", new JacksonHandle(getJsonFromResource("e2e-test/staged.json")));
 
         JobFinishedListener harmonizeFlowListener = new JobFinishedListener();
-        fm.runFlow(harmonizeFlow, 10, harmonizeFlowListener);
+        fm.runFlow(harmonizeFlow, 10, 1, harmonizeFlowListener);
         harmonizeFlowListener.waitForFinish();
         String expected = getResource("e2e-test/final.json");
         String actual = finalDocMgr.read("/input.json").next().getContent(new StringHandle()).get();