nit: change the nomenclature to distributed tasks

Author: sumandas0

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

@@ -40,7 +40,7 @@ public enum AtlasConfiguration {
     NOTIFICATION_HOOK_TOPIC_NAME("atlas.notification.hook.topic.name", "ATLAS_HOOK"),
     NOTIFICATION_ENTITIES_TOPIC_NAME("atlas.notification.entities.topic.name", "ATLAS_ENTITIES"),
     NOTIFICATION_RELATIONSHIPS_TOPIC_NAME("atlas.notification.relationships.topic.name", "ATLAS_RELATIONSHIPS"),
-    NOTIFICATION_ATLAS_DISTRIBUTED_TASKS_TOPIC_NAME("atlas.notification.atlas.tasks.topic.name", "ATLAS_TASKS"),
+    NOTIFICATION_ATLAS_DISTRIBUTED_TASKS_TOPIC_NAME("atlas.notification.distributed.tasks.topic.name", "ATLAS_DISTRIBUTED_TASKS"),
 
     NOTIFICATION_HOOK_CONSUMER_TOPIC_NAMES("atlas.notification.hook.consumer.topic.names", "ATLAS_HOOK"), //  a comma separated list of topic names
     NOTIFICATION_ENTITIES_CONSUMER_TOPIC_NAMES("atlas.notification.entities.consumer.topic.names", "ATLAS_ENTITIES"), //  a comma separated list of topic names

notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java

@@ -75,7 +75,7 @@ public class KafkaNotification extends AbstractNotification implements Service {
             put(NotificationType.HOOK, ATLAS_HOOK_TOPIC);
             put(NotificationType.ENTITIES, ATLAS_ENTITIES_TOPIC);
             put(NotificationType.RELATIONSHIPS, ATLAS_RELATIONSHIPS_TOPIC);
-            put(NotificationType.ATLAS_TASKS, ATLAS_DISTRIBUTED_TASKS_TOPIC);
+            put(NotificationType.ATLAS_DISTRIBUTED_TASKS, ATLAS_DISTRIBUTED_TASKS_TOPIC);
 
         }
     };

notification/src/main/java/org/apache/atlas/notification/NotificationInterface.java

@@ -52,7 +52,7 @@ enum NotificationType {
 
         RELATIONSHIPS(new EntityMessageDeserializer()),
 
-        ATLAS_TASKS(new AtlasDistributedTaskMessageDeserializer());
+        ATLAS_DISTRIBUTED_TASKS(new AtlasDistributedTaskMessageDeserializer());
 
         private final AtlasNotificationMessageDeserializer deserializer;
 

notification/src/main/java/org/apache/atlas/notification/task/AtlasTaskNotificationSender.java notification/src/main/java/org/apache/atlas/notification/task/AtlasDistributedTaskNotificationSender.java

@@ -1,6 +1,8 @@
-package org.apache.atlas.notification;
+package org.apache.atlas.notification.task;
 
 import org.apache.atlas.model.notification.AtlasDistributedTaskNotification;
+import org.apache.atlas.notification.NotificationException;
+import org.apache.atlas.notification.NotificationInterface;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
@@ -11,15 +13,15 @@
 import java.util.Map;
 
 @Component
-public class AtlasTaskNotificationSender {
-    private static final Logger LOG = LoggerFactory.getLogger(AtlasTaskNotificationSender.class);
+public class AtlasDistributedTaskNotificationSender {
+    private static final Logger LOG = LoggerFactory.getLogger(AtlasDistributedTaskNotificationSender.class);
 
     private final NotificationInterface notificationInterface;
 
     private final static Long batchSize = 40000L;
 
     @Autowired
-    public AtlasTaskNotificationSender(NotificationInterface notificationInterface) {
+    public AtlasDistributedTaskNotificationSender(NotificationInterface notificationInterface) {
         this.notificationInterface = notificationInterface;
     }
 
@@ -35,15 +37,15 @@ public AtlasDistributedTaskNotification createRelationshipCleanUpTask(String ver
 
     public void send(AtlasDistributedTaskNotification notification) {
         try {
-            notificationInterface.send(NotificationInterface.NotificationType.ATLAS_TASKS, notification);
+            notificationInterface.send(NotificationInterface.NotificationType.ATLAS_DISTRIBUTED_TASKS, notification);
         } catch (NotificationException e) {
             LOG.error("Failed to send notification for task: {}", notification, e);
         }
     }
 
     public void send(List<AtlasDistributedTaskNotification> notifications) {
         try {
-            notificationInterface.send(NotificationInterface.NotificationType.ATLAS_TASKS, notifications);
+            notificationInterface.send(NotificationInterface.NotificationType.ATLAS_DISTRIBUTED_TASKS, notifications);
         } catch (NotificationException e) {
             LOG.error("Failed to send notifications for tasks: {}", notifications, e);
         }

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

@@ -37,7 +37,7 @@
 import org.apache.atlas.model.notification.AtlasDistributedTaskNotification;
 import org.apache.atlas.model.tasks.AtlasTask;
 import org.apache.atlas.model.typedef.AtlasBaseTypeDef;
-import org.apache.atlas.notification.AtlasTaskNotificationSender;
+import org.apache.atlas.notification.task.AtlasDistributedTaskNotificationSender;
 import org.apache.atlas.repository.Constants;
 import org.apache.atlas.repository.RepositoryException;
 import org.apache.atlas.repository.graph.GraphHelper;
@@ -143,13 +143,13 @@ public class AtlasEntityStoreV2 implements AtlasEntityStore {
 
     private final ESAliasStore esAliasStore;
     private final IAtlasMinimalChangeNotifier atlasAlternateChangeNotifier;
-    private final AtlasTaskNotificationSender taskNotificationSender;
+    private final AtlasDistributedTaskNotificationSender taskNotificationSender;
 
     @Inject
     public AtlasEntityStoreV2(AtlasGraph graph, DeleteHandlerDelegate deleteDelegate, RestoreHandlerV1 restoreHandlerV1, AtlasTypeRegistry typeRegistry,
                               IAtlasEntityChangeNotifier entityChangeNotifier, EntityGraphMapper entityGraphMapper, TaskManagement taskManagement,
                               AtlasRelationshipStore atlasRelationshipStore, FeatureFlagStore featureFlagStore,
-                              IAtlasMinimalChangeNotifier atlasAlternateChangeNotifier, AtlasTaskNotificationSender taskNotificationSender) {
+                              IAtlasMinimalChangeNotifier atlasAlternateChangeNotifier, AtlasDistributedTaskNotificationSender taskNotificationSender) {
 
         this.graph                = graph;
         this.deleteDelegate       = deleteDelegate;
@@ -591,10 +591,6 @@ public EntityMutationResponse deleteById(final String guid) throws AtlasBaseExce
 
             AtlasAuthorizationUtils.verifyDeleteEntityAccess(typeRegistry, entityHeader, "delete entity: guid=" + guid);
 
-            if (AtlasConfiguration.NOTIFICATION_ATLAS_DISTRIBUTED_TASKS_TOPIC_NAME.getBoolean()) {
-                checkAndCreateProcessRelationshipsCleanupTaskNotification(typeRegistry.getEntityTypeByName(entityHeader.getTypeName()), vertex);
-            }
-
             deletionCandidates.add(vertex);
         } else {
             if (LOG.isDebugEnabled()) {
@@ -641,10 +637,6 @@ public EntityMutationResponse deleteByIds(final List<String> guids) throws Atlas
 
             AtlasAuthorizationUtils.verifyDeleteEntityAccess(typeRegistry, entityHeader, "delete entity: guid=" + guid);
 
-            if (AtlasConfiguration.NOTIFICATION_ATLAS_DISTRIBUTED_TASKS_TOPIC_NAME.getBoolean()) {
-                checkAndCreateProcessRelationshipsCleanupTaskNotification(typeRegistry.getEntityTypeByName(entityHeader.getTypeName()), vertex);
-            }
-
             deletionCandidates.add(vertex);
         }
 
@@ -755,10 +747,6 @@ public EntityMutationResponse deleteByUniqueAttributes(AtlasEntityType entityTyp
             AtlasAuthorizationUtils.verifyDeleteEntityAccess(typeRegistry, entityHeader,
                     "delete entity: typeName=" + entityType.getTypeName() + ", uniqueAttributes=" + uniqAttributes);
 
-            if (AtlasConfiguration.NOTIFICATION_ATLAS_DISTRIBUTED_TASKS_TOPIC_NAME.getBoolean()) {
-                checkAndCreateProcessRelationshipsCleanupTaskNotification(typeRegistry.getEntityTypeByName(entityHeader.getTypeName()), vertex);
-            }
-
             deletionCandidates.add(vertex);
         } else {
             if (LOG.isDebugEnabled()) {
@@ -1744,7 +1732,6 @@ private EntityMutationContext preCreateOrUpdate(EntityStream entityStream, Entit
 
             if (entity != null) { // entity would be null if guid is not in the stream but referenced by an entity in the stream
                 AtlasEntityType entityType = typeRegistry.getEntityTypeByName(entity.getTypeName());
-                // If typeName is Process or ColumnProcess we send task notification
 
                 if (entityType == null) {
                     throw new AtlasBaseException(element.getValue(), AtlasErrorCode.TYPE_NAME_INVALID, TypeCategory.ENTITY.name(), entity.getTypeName());
@@ -2125,6 +2112,10 @@ private EntityMutationResponse deleteVertices(Collection<AtlasVertex> deletionCa
 
                 String typeName = getTypeName(vertex);
 
+                if(ATLAS_DISTRIBUTED_TASK_ENABLED.getBoolean()) {
+                    checkAndCreateProcessRelationshipsCleanupTaskNotification(typeRegistry.getEntityTypeByName(typeName), vertex);
+                }
+
                 List<PreProcessor> preProcessors = getPreProcessor(typeName);
                 for(PreProcessor processor : preProcessors){
                     processor.processDelete(vertex);