SEARCH-2415: Performance tuning for MetadataTracker and ContentTracker.

Cherry picked from 7fbb361

search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ActivatableTracker.java

@@ -27,8 +27,6 @@ package org.alfresco.solr.tracker;
 
 import org.alfresco.solr.InformationServer;
 import org.alfresco.solr.client.SOLRAPIClient;
-import org.apache.solr.core.CoreDescriptorDecorator;
-import org.apache.solr.core.SolrCore;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 

search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/ContentTracker.java

@@ -53,7 +53,8 @@ public class ContentTracker extends ActivatableTracker
 {
     protected final static Logger LOGGER = LoggerFactory.getLogger(ContentTracker.class);
 
-    private static final int DEFAULT_CONTENT_TRACKER_MAX_PARALLELISM = 32;
+    // Keep this value to 1/4 of all the other pools, as ContentTracker Threads are heavier
+    private static final int DEFAULT_CONTENT_TRACKER_MAX_PARALLELISM = 8;
 
     private int contentTrackerParallelism;
     private int contentUpdateBatchSize;

search-services/alfresco-search/src/main/java/org/alfresco/solr/tracker/MetadataTracker.java

@@ -914,7 +914,7 @@ public class MetadataTracker extends ActivatableTracker
                             state.getTimeToStopIndexing());
                 }
 
-
+                long idTrackerCycle = System.currentTimeMillis();
                 if (transactions.getTransactions().size() > 0)
                 {
                     LOGGER.info("{}-[CORE {}] Found {} transactions after lastTxCommitTime {}, transactions from {} to {}",
@@ -943,37 +943,49 @@ public class MetadataTracker extends ActivatableTracker
                     break;
                 }
 
-                long transaction_number = transactions.getTransactions().size();
-                final AtomicInteger counter = new AtomicInteger();
+                final AtomicInteger counterTransaction = new AtomicInteger();
                 Collection<List<Transaction>> txBatches = transactions.getTransactions().stream()
                         .peek(txnsFound::add)
                         .filter(this::isTransactionIndexed)
-                        .peek(transaction -> {
-                            if (LOGGER.isTraceEnabled())
-                            {
-                                LOGGER.trace("{}-[CORE {}] Tracking {} Transactions. Current Transaction Id to be indexed: {}",
-                                        Thread.currentThread().getId(), coreName, transaction_number, transaction.getId());
-                            }
-                        })
-                        .collect(Collectors.groupingBy(transaction -> counter.getAndAdd(
+                        .collect(Collectors.groupingBy(transaction -> counterTransaction.getAndAdd(
                                 (int) (transaction.getDeletes() + transaction.getUpdates())) / transactionDocsBatchSize))
                         .values();
 
                 // Index batches of transactions and the nodes updated or deleted within the transaction
+                List<List<Node>> nodeBatches = new ArrayList<>();
                 for (List<Transaction> batch : txBatches)
                 {
 
                     // Index nodes contained in the transactions
-                    int docCount =  indexBatchOfTransactions(batch);
-                    totalUpdatedDocs += docCount;
+                    long idTxBatch = System.currentTimeMillis();
+                    nodeBatches.addAll(buildBatchOfTransactions(batch, idTrackerCycle, idTxBatch));
+                }
+                
+                // Counter used to identify the worker inside the parallel stream processing
+                final AtomicInteger counterBatch = new AtomicInteger(0);
+                long idThread = Thread.currentThread().getId();
+                totalUpdatedDocs += forkJoinPool.submit(() ->
+                        nodeBatches.parallelStream().map(batch -> {
+                            int count = counterBatch.addAndGet(1);
+                            if (LOGGER.isTraceEnabled())
+                            {
+                                LOGGER.trace("{}:{}:{}-[CORE {}] indexing {} nodes ...",
+                                        idThread, idTrackerCycle, count,
+                                        coreName, batch.size());
+                            }
+                            new NodeIndexWorker(batch, infoSrv, idThread, idTrackerCycle, count).run();
+                            return batch.size();
+                        }).reduce(0, Integer::sum)).get();
 
+                for (List<Transaction> batch : txBatches)
+                {
                     // Add the transactions as found to avoid processing them again in the next iteration
                     batch.forEach(txnsFound::add);
 
                     // Index the transactions
                     indexTransactionsAfterWorker(batch);
                     long endElapsed = System.nanoTime();
-                    trackerStats.addElapsedNodeTime(docCount, endElapsed - startElapsed);
+                    trackerStats.addElapsedNodeTime(totalUpdatedDocs, endElapsed - startElapsed);
                     startElapsed = endElapsed;
                 }
 
@@ -1039,19 +1051,18 @@ public class MetadataTracker extends ActivatableTracker
 
 
     /**
-     * Index a batch of transactions.
+     * Build a batch of transactions.
      *
      * Updated or deleted nodes from these transactions are also packed into batches in order to get
      * the metadata of the nodes in smaller invocations to Repository
      *
      * @param txBatch Batch of transactions to be indexed
-     * @return Number of nodes indexed and last node indexed
+     * @param idTrackerCycle Id of the Tracker Cycle being executed
+     * @param idTxBatch Id of the Transaction Batch being executed
+     * @return List of Nodes to be indexed splitted by nodeBatchSize count
      *
-     * @throws AuthenticationException
-     * @throws IOException
-     * @throws JSONException
      */
-    private int indexBatchOfTransactions(List<Transaction> txBatch)
+    private List<List<Node>> buildBatchOfTransactions(List<Transaction> txBatch, long idTrackerCycle, long idTxBatch)
             throws AuthenticationException, IOException, JSONException, ExecutionException, InterruptedException 
     {
         
@@ -1068,7 +1079,7 @@ public class MetadataTracker extends ActivatableTracker
         // Skip getting nodes when no transactions left
         if (txIds.size() == 0)
         {
-            return 0;
+            return Collections.emptyList();
         }
         
         // Get Nodes Id properties for every transaction
@@ -1089,13 +1100,8 @@ public class MetadataTracker extends ActivatableTracker
         }
 
         // Group the nodes in batches of nodeBatchSize (or less)
-        List<List<Node>> nodeBatches = Lists.partition(nodes, nodeBatchSize);
+        return Lists.partition(nodes, nodeBatchSize);
 
-        return forkJoinPool.submit(() ->
-                nodeBatches.parallelStream().map(batch -> {
-                    new NodeIndexWorker(batch, infoSrv).run();
-                    return batch.size();
-                }).reduce(0, Integer::sum)).get();
     }
 
 
@@ -1106,11 +1112,21 @@ public class MetadataTracker extends ActivatableTracker
     {
         InformationServer infoServer;
         List<Node> nodes;
+        // Unique Id for the worker > thread : trackerCycle : worker
+        long idThread;
+        long idTrackerCycle;
+        int idWorker;
+        
+        // Link logger messages to parent Class MetadataTracker
+        protected Logger LOGGER = LoggerFactory.getLogger(MetadataTracker.class);
 
-        NodeIndexWorker(List<Node> nodes, InformationServer infoServer)
+        NodeIndexWorker(List<Node> nodes, InformationServer infoServer, long idThread, long idTrackerCycle, int idWorker)
         {
             this.infoServer = infoServer;
             this.nodes = nodes;
+            this.idThread = idThread;
+            this.idTrackerCycle = idTrackerCycle;
+            this.idWorker = idWorker;
         }
 
         @Override
@@ -1121,6 +1137,11 @@ public class MetadataTracker extends ActivatableTracker
             {
                 this.infoServer.indexNodes(filteredNodes, true);
             }
+            if (LOGGER.isTraceEnabled())
+            {
+                LOGGER.trace("{}:{}:{}-[CORE {}] ...indexed",
+                        idThread, idTrackerCycle, idWorker, coreName);
+            }
         }
         
         @Override

search-services/alfresco-search/src/main/resources/solr/instance/templates/rerank/conf/solrcore.properties

@@ -139,10 +139,12 @@ alfresco.contentUpdateBatchSize=1000
 alfresco.cascadeNodeBatchSize=10
 
 # Trackers thread pools
-#alfresco.metadataTrackerMaxParallelism=
-#alfresco.aclTrackerMaxParallelism=
-#alfresco.contentTrackerMaxParallelism=
-#alfresco.cascadeTrackerMaxParallelism
+# Keep Content Tracker max threads to 1/4 of other values,
+# as this threads are heavier than the other ones.
+#alfresco.metadata.tracker.maxParallelism=32
+#alfresco.acl.tracker.maxParallelism=32
+#alfresco.content.tracker.maxParallelism=8
+#alfresco.cascade.tracker.maxParallelism=32
 
 # Warming