DG-1924 | Add 'assetsCountToPropagate' and 'assetsCountPropagated' in task vertex.

.github/workflows/maven.yml

@@ -29,6 +29,8 @@ on:
       - lineageondemand
       - cassandrapoliciesoptimisation
       - ixop
+      - makerlogic
+      - tagpropv1master
 
 jobs:
   build:
@@ -120,4 +122,4 @@ jobs:
       - name: Upload Trivy scan results to GitHub Security tab
         uses: github/codeql-action/[email protected]
         with:
-          sarif_file: 'trivy-image-results.sarif'
+          sarif_file: 'trivy-image-results.sarif'

common/src/main/java/org/apache/atlas/AtlasConstants.java

@@ -39,4 +39,6 @@ private AtlasConstants() {
     public static final String DEFAULT_TYPE_VERSION          = "1.0";
     public static final int    ATLAS_SHUTDOWN_HOOK_PRIORITY  = 30;
     public static final int TASK_WAIT_TIME_MS = 180_000;
+    public static final String ATLAS_KAFKA_TAG_TOPIC = "OBJECT_PROP_EVENTS";
+
 }

common/src/main/java/org/apache/atlas/repository/Constants.java

@@ -370,6 +370,11 @@ public final class Constants {
     public static final String TASK_CLASSIFICATION_ID       = encodePropertyKey(TASK_PREFIX + "classificationId");
     public static final String TASK_ENTITY_GUID             = encodePropertyKey(TASK_PREFIX + "entityGuid");
     public static final String TASK_CLASSIFICATION_TYPENAME = encodePropertyKey(TASK_PREFIX + "classificationTypeName");
+    public static final String TASK_ASSET_COUNT_TO_PROPAGATE = encodePropertyKey(TASK_PREFIX + "assetsCountToPropagate");
+    public static final String TASK_ASSET_COUNT_PROPAGATED = encodePropertyKey(TASK_PREFIX + "assetsCountPropagated");
+    public static final String TASK_ASSET_COUNT_FAILED = encodePropertyKey(TASK_PREFIX + "assetsFailedToPropagate");
+    public static final String TASK_CLEANUP_REQUIRED = encodePropertyKey(TASK_PREFIX + "cleanupRequired");
+    public static final String TASK_CLEANUP_STATUS = encodePropertyKey(TASK_PREFIX + "cleanupStatus");
     public static final String ACTIVE_STATE_VALUE           = "ACTIVE";
     public static final String TASK_HEADER_ATLAN_AGENT      = "x-atlan-agent";
     public static final String TASK_HEADER_ATLAN_AGENT_ID   = "x-atlan-agent-id";
@@ -490,6 +495,9 @@ public enum SupportedFileExtensions { XLSX, XLS, CSV }
     public static final String REQUEST_HEADER_USER_AGENT = "User-Agent";
     public static final String REQUEST_HEADER_HOST       = "Host";
 
+    //kafka partition value for OBJ_PROP_EVENTS
+    public static final String OBJ_PROP_EVENTS_PARTITION_COUNT = "10";
+
     public static final Set<String> SKIP_UPDATE_AUTH_CHECK_TYPES = new HashSet<String>() {{
         add(README_ENTITY_TYPE);
         add(LINK_ENTITY_TYPE);

common/src/main/java/org/apache/atlas/service/redis/AbstractRedisService.java

@@ -12,6 +12,7 @@
 import javax.annotation.PreDestroy;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.time.Duration;
 import java.util.Arrays;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
@@ -72,6 +73,26 @@ public void releaseDistributedLock(String key) {
         }
     }
 
+    @Override
+    public int getSetSize(String key) {
+        return redisCacheClient.getSet(convertToNamespace(key)).size();
+    }
+
+    @Override
+    public String getHashValue(String key, String field) {
+        return (String) redisCacheClient.getMap(convertToNamespace(key)).get(field);
+    }
+
+    @Override
+    public void expireKey(String key) {
+        redisCacheClient.getBucket(convertToNamespace(key)).expire(Duration.ofSeconds(60));
+    }
+
+    @Override
+    public void expireHash(String key) {
+        redisCacheClient.getMap(convertToNamespace(key)).expire(Duration.ofSeconds(60));
+    }
+
     @Override
     public String getValue(String key) {
         // If value doesn't exist, return null else return the value

common/src/main/java/org/apache/atlas/service/redis/RedisService.java

@@ -8,6 +8,12 @@ public interface RedisService {
 
   void releaseDistributedLock(String key);
 
+  int getSetSize(String key);
+
+  String getHashValue(String key, String field);
+
+  void expireHash(String key);
+
   String getValue(String key);
 
   String putValue(String key, String value);
@@ -18,4 +24,5 @@ public interface RedisService {
 
   Logger getLogger();
 
+  void expireKey(String key);
 }

common/src/main/java/org/apache/atlas/utils/KafkaUtils.java

@@ -108,6 +108,38 @@ public void createTopics(List<String> topicNames, int numPartitions, int replica
         }
     }
 
+    public void createTopics(List<String[]> topicDetails, int replicationFactor)
+            throws TopicExistsException, ExecutionException, InterruptedException {
+
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("==> createTopics()");
+        }
+
+        List<NewTopic> newTopicList = topicDetails.stream()
+                .map(details -> new NewTopic(details[0], Integer.parseInt(details[1]), (short) replicationFactor))
+                .collect(Collectors.toList());
+
+        CreateTopicsResult             createTopicsResult = adminClient.createTopics(newTopicList);
+        Map<String, KafkaFuture<Void>> futureMap          = createTopicsResult.values();
+
+        for (Map.Entry<String, KafkaFuture<Void>> futureEntry : futureMap.entrySet()) {
+            KafkaFuture<Void> future = futureEntry.getValue();
+
+            try {
+                future.get();
+            } catch (ExecutionException | InterruptedException e) {
+                LOG.error("Error while creating topic " + futureEntry.getKey(), e);
+                // Re-throw to handle these exceptions in the calling code.
+                throw e;
+            }
+        }
+
+        if (LOG.isDebugEnabled()) {
+            LOG.debug("<== createTopics()");
+        }
+    }
+
+
     public List<String> listAllTopics() throws ExecutionException, InterruptedException {
         if (LOG.isDebugEnabled()) {
             LOG.debug("==> KafkaUtils.listAllTopics() ");

distro/src/bin/atlas_config.py

@@ -500,7 +500,7 @@ def get_topics_to_create(confdir):
     if topic_list is not None:
         topics = topic_list.split(",")
     else:
-        topics = [getConfigWithDefault("atlas.notification.hook.topic.name", "ATLAS_HOOK"), getConfigWithDefault("atlas.notification.entities.topic.name", "ATLAS_ENTITIES"), getConfigWithDefault("atlas.notification.relationships.topic.name", "ATLAS_RELATIONSHIPS")]
+        topics = [getConfigWithDefault("atlas.notification.hook.topic.name", "ATLAS_HOOK"), getConfigWithDefault("atlas.notification.entities.topic.name", "ATLAS_ENTITIES"), getConfigWithDefault("atlas.notification.relationships.topic.name", "ATLAS_RELATIONSHIPS"), getConfigWithDefault("atlas.notification.propagation.topic.name", "TAG_PROP_EVENTS")]
     return topics
 
 def get_atlas_url_port(confdir):

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

@@ -40,6 +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_OBJ_PROPAGATION_TOPIC_NAME("atlas.notification.propagation.topic.name", "OBJECT_PROP_EVENTS"),
     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
@@ -84,7 +85,7 @@ public enum AtlasConfiguration {
     DSL_CACHED_TRANSLATOR("atlas.dsl.cached.translator", true),
     DEBUG_METRICS_ENABLED("atlas.debug.metrics.enabled", false),
     TASKS_USE_ENABLED("atlas.tasks.enabled", true),
-    ATLAS_DISTRIBUTED_TASK_ENABLED("atlas.distributed.task.enabled", false),
+    ATLAS_DISTRIBUTED_TASK_ENABLED("atlas.distributed.task.enabled", true),
     TASKS_REQUEUE_GRAPH_QUERY("atlas.tasks.requeue.graph.query", false),
     TASKS_IN_PROGRESS_GRAPH_QUERY("atlas.tasks.inprogress.graph.query", false),
     TASKS_REQUEUE_POLL_INTERVAL("atlas.tasks.requeue.poll.interval.millis", 60000),
@@ -137,7 +138,9 @@ public enum AtlasConfiguration {
     OTEL_RESOURCE_ATTRIBUTES("OTEL_RESOURCE_ATTRIBUTES", "service.name=atlas"),
     OTEL_SERVICE_NAME(" OTEL_SERVICE_NAME", "atlas"),
     OTEL_EXPORTER_OTLP_ENDPOINT("OTEL_EXPORTER_OTLP_ENDPOINT", "http://localhost:4317"),
-    ATLAS_BULK_API_MAX_ENTITIES_ALLOWED("atlas.bulk.api.max.entities.allowed", 10000);
+    ATLAS_BULK_API_MAX_ENTITIES_ALLOWED("atlas.bulk.api.max.entities.allowed", 10000),
+    ATLAS_DISTRIBUTED_TASK_MANAGEMENT_ENABLED("atlas.distributed.task.management.enabled", true),
+    NOTIFICATION_OBJ_PROPAGATION_TOPIC_PARTITIONS("atlas.obj.prop.topic.partitions", 10);
 
 
 

intg/src/main/java/org/apache/atlas/model/tasks/AtlasTask.java

@@ -80,6 +80,32 @@ public static Status from(String s) {
         }
     }
 
+    public enum CleanupStatus {
+        PENDING,
+        SUCCESS,
+        FAILED;
+
+        public static CleanupStatus from(String s) {
+            if(StringUtils.isEmpty(s)) {
+                return PENDING;
+            }
+
+            switch (s.toLowerCase()) {
+                case "pending":
+                    return PENDING;
+
+                case "success":
+                    return SUCCESS;
+
+                case "failed":
+                    return FAILED;
+
+                default:
+                    return PENDING;
+            }
+        }
+    }
+
     private String              type;
     private String              guid;
     private String              createdBy;
@@ -95,6 +121,11 @@ public static Status from(String s) {
     private String              classificationId;
     private String              entityGuid;
     private String              classificationTypeName;
+    private Long                assetsCountToPropagate;
+    private Long                assetsCountPropagated;
+    private Long                assetsFailedToPropagate;
+    private boolean             cleanupRequired;
+    private CleanupStatus       cleanupStatus;
 
     public AtlasTask() {
     }
@@ -111,6 +142,11 @@ public AtlasTask(String type, String createdBy, Map<String, Object> parameters,
         this.attemptCount       = 0;
         this.classificationId   = classificationId;
         this.entityGuid         = entityGuid;
+        this.assetsCountToPropagate = 0L;
+        this.assetsCountPropagated = 0L;
+        this.assetsFailedToPropagate = 0L;
+        this.cleanupRequired    = false;
+        this.cleanupStatus      = CleanupStatus.PENDING;
     }
 
     public String getGuid() {
@@ -175,6 +211,22 @@ public Status getStatus() {
         return this.status;
     }
 
+    public void setCleanupRequired(boolean cleanupRequired) {
+        this.cleanupRequired = cleanupRequired;
+    }
+
+    public boolean getCleanupRequired() {
+        return cleanupRequired;
+    }
+
+    public CleanupStatus getCleanupStatus() {
+        return this.cleanupStatus;
+    }
+
+    public void setCleanupStatus(CleanupStatus cleanupStatus) {
+        this.cleanupStatus = cleanupStatus;
+    }
+
     public int getAttemptCount() {
         return attemptCount;
     }
@@ -239,9 +291,34 @@ public String getEntityGuid() {
         return entityGuid;
     }
 
+    public void setAssetsCountToPropagate(Long assetsCount) {
+        this.assetsCountToPropagate = assetsCount;
+    }
+
+    public Long getAssetsCountToPropagate() {
+        return assetsCountToPropagate;
+    }
+
+    public void setAssetsCountPropagated(Long assetsCountPropagated) {
+        this.assetsCountPropagated = assetsCountPropagated;
+    }
+
+    public Long getAssetsCountPropagated(){
+        return  assetsCountPropagated;
+    }
+
+    public void setAssetsFailedToPropagate(Long assetsFailedToPropagate) {
+        this.assetsFailedToPropagate = assetsFailedToPropagate;
+    }
+
+    public Long getAssetsFailedToPropagate() {
+        return assetsFailedToPropagate;
+    }
+
     @JsonIgnore
     public void start() {
         this.setStatus(Status.IN_PROGRESS);
+        this.setCleanupStatus(CleanupStatus.PENDING);
         this.setStartTime(new Date());
     }
 
@@ -270,6 +347,11 @@ public String toString() {
                 ", attemptCount=" + attemptCount +
                 ", errorMessage='" + errorMessage + '\'' +
                 ", status=" + status +
+                ", assetsCountToPropagate=" + assetsCountToPropagate +
+                ", assetsCountPropagated=" + assetsCountPropagated +
+                ", assetsPropagationFailedCount=" + assetsFailedToPropagate +
+                ", cleanupRequired=" + cleanupRequired +
+                ", cleanupStatus=" + cleanupStatus +
                 '}';
     }
 }

notification/pom.xml

@@ -82,7 +82,10 @@
             <classifier>tests</classifier>
             <scope>test</scope>
         </dependency>
-
+        <dependency>
+            <groupId>org.apache.atlas</groupId>
+            <artifactId>atlas-graphdb-api</artifactId>
+        </dependency>
     </dependencies>
 
     <build>

notification/src/main/java/org/apache/atlas/hook/AtlasTopicCreator.java

@@ -31,7 +31,10 @@
 import org.slf4j.LoggerFactory;
 
 import java.io.IOException;
-import java.util.Arrays;
+import java.util.ArrayList;
+import java.util.List;
+import org.apache.atlas.AtlasConfiguration;
+import static org.apache.atlas.repository.Constants.OBJ_PROP_EVENTS_PARTITION_COUNT;
 
 /**
  * A class to create Kafka topics used by Atlas components.
@@ -61,19 +64,30 @@ public void createAtlasTopic(Configuration atlasProperties, String... topicNames
             if (!handleSecurity(atlasProperties)) {
                 return;
             }
-            try(KafkaUtils kafkaUtils = getKafkaUtils(atlasProperties)) {
-                int numPartitions = atlasProperties.getInt("atlas.notification.partitions", 1);
+            try (KafkaUtils kafkaUtils = getKafkaUtils(atlasProperties)) {
                 int numReplicas = atlasProperties.getInt("atlas.notification.replicas", 1);
-                kafkaUtils.createTopics(Arrays.asList(topicNames), numPartitions, numReplicas);
+                List<String[]> topicDetails = new ArrayList<>();
+
+                for (String topicName : topicNames) {
+                    if (AtlasConfiguration.NOTIFICATION_OBJ_PROPAGATION_TOPIC_NAME.getString().equals(topicName)) {
+                        topicDetails.add(new String[]{topicName, OBJ_PROP_EVENTS_PARTITION_COUNT}); // 5 partitions for 'TAG_PROP_EVENTS'
+                    } else {
+                        topicDetails.add(new String[]{topicName, String.valueOf(atlasProperties.getInt("atlas.notification.partitions", 1))}); // 1 partition for all others
+                    }
+                }
+
+                kafkaUtils.createTopics(topicDetails, numReplicas);
+
             } catch (Exception e) {
-                LOG.error("Error while creating topics e :" + e.getMessage(), e);
+                LOG.error("Error while creating topics: " + e.getMessage(), e);
             }
         } else {
             LOG.info("Not creating topics {} as {} is false", StringUtils.join(topicNames, ","),
                     ATLAS_NOTIFICATION_CREATE_TOPICS_KEY);
         }
     }
 
+
     @VisibleForTesting
     protected boolean handleSecurity(Configuration atlasProperties) {
         if (AuthenticationUtil.isKerberosAuthenticationEnabled(atlasProperties)) {