Async the group de-index during upload in content-index service

  Recent performance problem we met for uploading & promotion is all
caused by the getAffectedBy operation, which is a time-consumed one for
a full scan on store data. After checking, I found that it is mainly
used for de-index or re-index content-index entries for the uploading
repo. I think this de-index will not be a block factor during the upload
or promotion process, which means we could make it async.
  In this commit, I switched back to use a normal executor service to
async the de-index of the affected group. One potential problem for this
solution is the executor queue size, which could cause OOM if there are
bunch of uploading happened in a short time period. We need to monitor
if this can happen.

Author: ligangty

addons/content-index/src/main/java/org/commonjava/indy/content/index/IndexingContentManagerDecorator.java

@@ -15,14 +15,15 @@
  */
 package org.commonjava.indy.content.index;
 
+import org.commonjava.cdi.util.weft.ExecutorConfig;
+import org.commonjava.cdi.util.weft.WeftManaged;
 import org.commonjava.indy.IndyWorkflowException;
 import org.commonjava.indy.content.ContentManager;
 import org.commonjava.indy.content.index.conf.ContentIndexConfig;
 import org.commonjava.indy.core.content.PathMaskChecker;
 import org.commonjava.indy.data.IndyDataException;
 import org.commonjava.indy.data.StoreDataManager;
 import org.commonjava.indy.measure.annotation.Measure;
-import org.commonjava.indy.measure.annotation.MetricNamed;
 import org.commonjava.indy.model.core.ArtifactStore;
 import org.commonjava.indy.model.core.Group;
 import org.commonjava.indy.model.core.HostedRepository;
@@ -51,10 +52,10 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
+import java.util.concurrent.Executor;
 
 import static org.commonjava.indy.core.content.group.GroupMergeHelper.GROUP_METADATA_EXISTS;
 import static org.commonjava.indy.core.content.group.GroupMergeHelper.GROUP_METADATA_GENERATED;
-import static org.commonjava.indy.measure.annotation.MetricNamed.DEFAULT;
 
 /**
  * Decorator for ContentManager which uses Infinispan to index content to avoid having to iterate all members of large
@@ -88,6 +89,11 @@ public abstract class IndexingContentManagerDecorator
     @Inject
     private ContentIndexConfig indexCfg;
 
+    @Inject
+    @WeftManaged
+    @ExecutorConfig( named="content-index-store-deindex", priority = 4, threads = 10)
+    private Executor deIndexExecutor;
+
     protected IndexingContentManagerDecorator()
     {
     }
@@ -112,6 +118,15 @@ protected IndexingContentManagerDecorator( final ContentManager delegate, final
         this.indexCfg = indexCfg;
     }
 
+    protected IndexingContentManagerDecorator( final ContentManager delegate, final StoreDataManager storeDataManager,
+                                               final SpecialPathManager specialPathManager,
+                                               final ContentIndexManager indexManager, final NotFoundCache nfc,
+                                               final ContentIndexConfig indexCfg, final Executor deIndexExecutor)
+    {
+        this(delegate, storeDataManager, specialPathManager, indexManager, nfc, indexCfg);
+        this.deIndexExecutor = deIndexExecutor;
+    }
+
     @Override
     public Transfer retrieveFirst( final List<? extends ArtifactStore> stores, final String path )
             throws IndyWorkflowException
@@ -163,12 +178,7 @@ public List<Transfer> retrieveAll( final List<? extends ArtifactStore> stores, f
             }
 
             return null;
-        } ).filter( Objects::nonNull ).forEachOrdered( ( transfer ) -> {
-            if ( transfer != null )
-            {
-                results.add( transfer );
-            }
-        } );
+        } ).filter( Objects::nonNull ).forEachOrdered( results::add );
 
         return results;
     }
@@ -200,7 +210,7 @@ public Transfer retrieve( final ArtifactStore store, final String path, final Ev
         else if ( isAuthoritativelyMissing( store ) )
         {
             logger.debug(
-                    "Not found indexed transfer: {} and authoritative index switched on. Considering not found and return null." );
+                    "Not found indexed transfer: {} and authoritative index switched on. Considering not found and return null.", transfer );
             return null;
         }
 
@@ -286,7 +296,7 @@ else if ( isAuthoritativelyMissing( store ) )
                     {
                         ; // metadata generated/exists but missing due to membership change, not add to nfc so next req can retry
                     }
-                    else if ( StoreType.hosted != type ) // don't track NFC for hosted repos
+                    else // don't track NFC for hosted repos
                     {
                         nfc.addMissing( resource );
                     }
@@ -458,7 +468,7 @@ public Transfer getTransfer( final ArtifactStore store, final String path, final
         else if ( isAuthoritativelyMissing( store ) )
         {
             logger.info(
-                    "Not found indexed transfer: {} and authoritative index switched on. Considering not found and return null." );
+                    "Not found indexed transfer: {} and authoritative index switched on. Considering not found and return null.", transfer );
             return null;
         }
 
@@ -510,7 +520,7 @@ else if ( isAuthoritativelyMissing( store ) )
         return transfer;
     }
 
-    @Measure( timers = @MetricNamed( DEFAULT ), exceptions = @MetricNamed( DEFAULT ) )
+    @Measure
     @Deprecated
     public Transfer getIndexedMemberTransfer( final Group group, final String path, TransferOperation op,
                                                ContentManagementFunction func, final EventMetadata metadata )
@@ -602,7 +612,7 @@ public Transfer getTransfer( final StoreKey storeKey, final String path, final T
 
         if ( isAuthoritativelyMissing( store ) )
         {
-            logger.debug( "Not found indexed transfer: {} and authoritative index switched on. Return null." );
+            logger.debug( "Not found indexed transfer: {} and authoritative index switched on. Return null.", transfer );
             return null;
         }
 
@@ -713,21 +723,25 @@ public Transfer store( final ArtifactStore store, final String path, final Input
             // may change the content index sequence based on the constituents sequence in parent groups
             if ( store.getType() == StoreType.hosted )
             {
-                try
-                {
-                    Set<Group> groups = storeDataManager.query().getGroupsAffectedBy( store.getKey() );
-                    if ( groups != null && !groups.isEmpty() && indexCfg.isEnabled() )
+                //FIXME: One potential problem here: The fixed thread pool is using a blocking queue to
+                // cache runnables, which could cause OOM if there are bunch of uploading happened in
+                // a short time period. We need to monitor if this could happen.
+                deIndexExecutor.execute( () -> {
+                    try
                     {
-                        groups.forEach( g -> indexManager.deIndexStorePath( g.getKey(), path ) );
+                        Set<Group> groups = storeDataManager.query().getGroupsAffectedBy( store.getKey() );
+                        if ( groups != null && !groups.isEmpty() && indexCfg.isEnabled() )
+                        {
+                            groups.forEach( g -> indexManager.deIndexStorePath( g.getKey(), path ) );
+                        }
                     }
-                }
-                catch ( IndyDataException e )
-                {
-                    throw new IndyWorkflowException(
-                            "Failed to get groups which contains: %s for NFC handling. Reason: %s", e, store.getKey(),
-                            e.getMessage() );
-                }
-
+                    catch ( IndyDataException e )
+                    {
+                        logger.error(
+                                String.format( "Failed to get groups which contains: %s for NFC handling. Reason: %s",
+                                               store.getKey(), e.getMessage() ), e );
+                    }
+                } );
             }
         }
 //        nfcClearByContaining( store, path );