DG-1697: Adding threadpool for linking/unlink policy

intg/src/main/java/org/apache/atlas/AtlasConfiguration.java

@@ -113,6 +113,7 @@ public enum AtlasConfiguration {
 
     INDEXSEARCH_ASYNC_SEARCH_KEEP_ALIVE_TIME_IN_SECONDS("atlas.indexsearch.async.search.keep.alive.time.in.seconds", 300),
 
+    POLICY_OPERATIONS_NOTIFICATION_MAX_THREADS("atlas.policy.operations.max.threads", 5),
     ATLAS_MAINTENANCE_MODE("atlas.maintenance.mode", false);
 
 

repository/src/main/java/org/apache/atlas/repository/store/graph/v2/AtlasEntityStoreV2.java

@@ -20,6 +20,7 @@
 
 
 import com.google.common.annotations.VisibleForTesting;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 import org.apache.atlas.AtlasErrorCode;
 import org.apache.atlas.DeleteType;
 import org.apache.atlas.GraphTransactionInterceptor;
@@ -105,6 +106,8 @@
 import java.io.InputStream;
 import java.util.*;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.stream.Collectors;
 
 import static java.lang.Boolean.FALSE;
@@ -152,6 +155,8 @@ public class AtlasEntityStoreV2 implements AtlasEntityStore {
 
     private final ESAliasStore esAliasStore;
 
+    private final ExecutorService executorService;
+
     @Inject
     public AtlasEntityStoreV2(AtlasGraph graph, DeleteHandlerDelegate deleteDelegate, RestoreHandlerV1 restoreHandlerV1, AtlasTypeRegistry typeRegistry,
                               IAtlasEntityChangeNotifier entityChangeNotifier, EntityGraphMapper entityGraphMapper, TaskManagement taskManagement,
@@ -169,6 +174,11 @@ public AtlasEntityStoreV2(AtlasGraph graph, DeleteHandlerDelegate deleteDelegate
         this.atlasRelationshipStore = atlasRelationshipStore;
         this.featureFlagStore = featureFlagStore;
         this.esAliasStore = new ESAliasStore(graph, entityRetriever);
+        this.executorService =   Executors.newFixedThreadPool(AtlasConfiguration.POLICY_OPERATIONS_NOTIFICATION_MAX_THREADS.getInt(),
+                new ThreadFactoryBuilder()
+                        .setNameFormat("atlas-policy-operations-notification-%d")
+                        .build()
+        );
 
         try {
             this.discovery = new EntityDiscoveryService(typeRegistry, graph, null, null, null, null);
@@ -2785,7 +2795,7 @@ public void unlinkBusinessPolicy(String guid, List<String> unlinkGuids) throws A
             } finally {
                 RequestContext.clear();
             }
-        });
+        }, this.executorService);
         RequestContext.get().endMetricRecord(metric);
     }
 }