#1279 size the task queue for uri collectors and document processors

Author: ehennum

marklogic-client-api-functionaltests/src/test/java/com/marklogic/client/datamovement/functionaltests/TestSplitters.java

@@ -476,7 +476,6 @@ public void testWriteOpsMultipleThreads() throws KeyManagementException, NoSuchA
         try {
             tempJsonFile1 = File.createTempFile("TestWriteOps1", ".json");
             BufferedWriter bwJson1 = new BufferedWriter(new FileWriter(tempJsonFile1));
-            DocumentManager docMgrJson1 = client.newJSONDocumentManager();
             dmManager = client.newDataMovementManager();
             WriteBatcher wbatcher1 = dmManager.newWriteBatcher();
 
@@ -530,15 +529,22 @@ public void testWriteOpsMultipleThreads() throws KeyManagementException, NoSuchA
             t1.join();
             t2.join();
 
-            Thread.sleep(1000);
+            wbatcher1.awaitCompletion();
+
             // Verify docs count
             QueryBatcher queryBatcherdMgr = dmManager.newQueryBatcher(
                     new StructuredQueryBuilder().collection(collectionName))
                     .withBatchSize(20)
-                    //.withThreadCount(1)
+                    .withThreadCount(1)
                     .onUrisReady((batch) -> {
                         cnt1.addAndGet(batch.getItems().length);
-                    });
+                        })
+                    .onQueryFailure(failure -> {
+                        System.out.println("query failed");
+                        failure.printStackTrace(System.out);
+                        dmManager.stopJob(failure.getBatcher());
+                        });
+
             dmManager.startJob(queryBatcherdMgr);
             queryBatcherdMgr.awaitCompletion();
             assertEquals(2 * nDocs, cnt1.get());
@@ -565,7 +571,6 @@ public void testWriteOpsMultipleThreads() throws KeyManagementException, NoSuchA
                 jsonfs1.close();
             if (jsonfs2 != null)
                 jsonfs2.close();
-            docMgrJson1 = null;
         } catch (Exception ex) {
             System.out.println("Exceptions thrown testSplitterWithMultipleThreads " + ex.getMessage());
         } finally {

marklogic-client-api/src/main/java/com/marklogic/client/datamovement/impl/QueryBatcherImpl.java

@@ -492,7 +492,7 @@ private synchronized void initialize() {
                     "onUrisReady listeners={}, failure listeners={}",
       getBatchSize(), getDocToUriBatchRatio(), getThreadThrottleFactor(), getThreadCount(),
             urisReadyListeners.size(), failureListeners.size());
-    threadPool = new QueryThreadPoolExecutor(getThreadCount(), this);
+    threadPool = new QueryThreadPoolExecutor(getThreadCount(), getDocToUriBatchRatio(), this);
   }
 
   /* When withForestConfig is called before the job starts, it just provides
@@ -1141,9 +1141,9 @@ public void rejectedExecution(Runnable runnable, ThreadPoolExecutor executor) {
   private class QueryThreadPoolExecutor extends ThreadPoolExecutor {
     private Object objectToNotifyFrom;
 
-    QueryThreadPoolExecutor(int threadCount, Object objectToNotifyFrom) {
+    QueryThreadPoolExecutor(int threadCount, int docToUriBatchRatio, Object objectToNotifyFrom) {
       super(threadCount, threadCount, 0, TimeUnit.MILLISECONDS,
-        new LinkedBlockingQueue<Runnable>(threadCount * 25), new BlockingRunsPolicy());
+        new LinkedBlockingQueue<Runnable>(threadCount * (docToUriBatchRatio + 2)), new BlockingRunsPolicy());
       this.objectToNotifyFrom = objectToNotifyFrom;
     }