ADMIN-3 | Send notifications to Kafka topic after planning is done for tag propagation #4085

abhijeet-atlan · 2025-02-04T12:26:00Z

Change description

Key Changes:

Kafka Integration:
- Added TAG_PROP_EVENTS topic for tag propagation messages.
- Defined a partition count (TAG_PROP_EVENTS_PARTITION_COUNT = 5) for better load distribution.
- Updated AtlasTopicCreator and KafkaNotification to support message partitioning.
Notification Improvements:
- Introduced partition-aware Kafka message handling for classification propagation.
- Updated NotificationInterface and AbstractNotification to support partitioned message sending.

Ref: V1.5

Description here

Type of change

Bug fix (fixes an issue)
New feature (adds functionality)

Related issues

Fix #1

Checklists

Development

Lint rules pass locally
Application changes have been tested thoroughly
Automated tests covering modified code pass

Security

Security impact of change has been considered
Code follows company security practices and guidelines

Code review

Pull request has a descriptive title and context useful to a reviewer. Screenshots or screencasts are attached as necessary
"Ready for review" label attached and reviewers assigned
Changes have been reviewed by at least one other contributor
Pull request linked to task tracker where applicable

jnkrmg · 2025-02-05T00:54:39Z

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 = "TAG_PROP_EVENTS";


We will create a new topic which will handle other object propagation use-cases w a more generic name.

That is upcoming in a TRD. Will share and we can make changes accordingly.

sure, will wait. Include me in the discussions too

We will create a new topic which will handle other object propagation use-cases w a more generic name.
Can we change this to some generic name then ?

Or you want to go ahead with this for now?

Let's change the name it OBJECT_PROP_EVENTS.

jnkrmg · 2025-02-05T00:55:40Z

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

@@ -480,6 +480,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 TAG_PROP_EVENTS
+    public static final String TAG_PROP_EVENTS_PARTITION_COUNT = "5";


Partition count might also get changed once we determine how we combine more use-cases into the same Kafka topic.

sure, will wait. Include me in the discussions too

jnkrmg · 2025-02-05T00:58:56Z

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

@@ -418,4 +478,18 @@ private boolean isKafkaConsumerOpen(KafkaConsumer consumer) {
        return ret;
    }

+    public Map<String, Object> createKafkaMessage(AtlasVertex vertex, AtlasGraph graph, String classificationType, String tagVertedId) {


The function name is too generic and its body is doing something v specific to tag propagation. Rename the func.

changed it to createTagPropKafkaMessage

jnkrmg · 2025-02-05T01:00:31Z

notification/src/main/java/org/apache/atlas/notification/spool/Spooler.java

@@ -70,6 +71,11 @@ public void sendInternal(NotificationType type, List<String> messages) {
        }
    }

+    @Override
+    public void sendInternal(NotificationType notificationType, List<String> messages, Integer partition) throws NotificationException {


Raise a NotImplemented exception from this method if it's only written to satisfy an interface.

jnkrmg · 2025-02-05T01:00:58Z

repository/pom.xml

+        <dependency>
+            <groupId>org.apache.atlas</groupId>
+            <artifactId>atlas-notification</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.atlas</groupId>
+            <artifactId>atlas-notification</artifactId>
+        </dependency>


Duplicated?

jnkrmg · 2025-02-05T01:02:02Z

repository/src/main/java/org/apache/atlas/repository/store/graph/v1/DeleteHandlerV1.java

@@ -98,21 +100,9 @@ public abstract class DeleteHandlerV1 {
    private   final TaskUtil             taskUtil;
    private static final int CHUNK_SIZE            = AtlasConfiguration.TASKS_GRAPH_COMMIT_CHUNK_SIZE.getInt();

-    public void updateTaskVertexProperty(String propertyKey, long value, boolean isIncremental, BiConsumer<AtlasTask, Long> taskSetter) {


Rebase and bring the latest v1 changes into this branch. Will continue reviewing once that is done.

nikhilbonte21 · 2025-02-07T05:33:28Z

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 = "TAG_PROP_EVENTS";


We will create a new topic which will handle other object propagation use-cases w a more generic name.
Can we change this to some generic name then ?

nikhilbonte21 · 2025-02-07T05:33:59Z

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 = "TAG_PROP_EVENTS";


Or you want to go ahead with this for now?

nikhilbonte21 · 2025-02-07T05:36:26Z

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 = "TAG_PROP_EVENTS";


In the other python script, we are also considering the atlas.notification.propagation.topic.name from application properties, I think we should follow same here

Referring to distro/src/bin/atlas_config.py

nikhilbonte21 · 2025-02-07T05:37:32Z

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)


Why we need this method if the script distro/src/bin/atlas_config.py also does create topics?

nikhilbonte21 · 2025-02-07T05:42:36Z

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

+                List<String[]> topicDetails = new ArrayList<>();
+
+                for (String topicName : topicNames) {
+                    if (AtlasConfiguration.NOTIFICATION_PROPAGATION_TOPIC_NAME.getString().equals(topicName)) {


We could have created An Enum for all Kafka topics with all the details of every topic needed to avoid such special handling for a particular topic, fine for now shared this for future reference

nikhilbonte21 · 2025-02-07T06:06:10Z

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

+        List<MessageContext> messageContexts = new ArrayList<>();
+
+        for (String message : messages) {
+            ProducerRecord record = new ProducerRecord(topic, partition, null, message);


Once we accept message key instead of partition, use the proper ProducerRecord constructor here which accepts topic, key & value

nikhilbonte21 · 2025-02-07T06:11:23Z

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

+                    Map<String, Object> kafkaMessage = kfknotif.createTagPropKafkaMessage(vertex, graph, CLASSIFICATION_PROPAGATION_ADD, classificationVertex.getIdForDisplay());
+                    int partition = Math.abs((Integer) kafkaMessage.get("parentTaskGuid")) % numPartitions;
+                    LOG.debug("sending message with  guid={} to partition={}",kafkaMessage.get("parentTaskVertexId"), partition);
+                    kfknotif.sendInternal(NotificationInterface.NotificationType.EMIT_PLANNED_RELATIONSHIPS, Collections.singletonList(kafkaMessage.toString()), partition);
+                    LOG.debug("Message with guid={} sent to partition={} sent successfully.",kafkaMessage.get("parentTaskVertexId"), partition );


Try to simplify 2 method calls into a single method call in kfknotif may be

nikhilbonte21 · 2025-02-07T06:12:18Z

setup.py

+#!/usr/bin/env python3
+
+import os
+import subprocess
+import sys
+import platform
+import time
+
+
+
+
+


I think this file should not be in your PR, do rebase once

nikhilbonte21 · 2025-02-07T06:12:51Z

webapp/src/main/java/org/apache/atlas/Atlas.java

@@ -65,6 +65,8 @@
 import static org.apache.atlas.repository.Constants.INDEX_PREFIX;
 import static org.apache.atlas.repository.Constants.VERTEX_INDEX;

+import org.apache.atlas.hook.AtlasTopicCreator;


I think this is not needed, remove all changes in this file

nikhilbonte21 · 2025-02-07T06:14:51Z

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

+                            int partition = Math.abs((Integer) kafkaMessage.get("parentTaskGuid")) % numPartitions;
+                            LOG.debug("sending message with  guid={} to partition={}",kafkaMessage.get("parentTaskVertexId"), partition);
+                            kfknotif.sendInternal(NotificationInterface.NotificationType.EMIT_PLANNED_RELATIONSHIPS, Collections.singletonList(kafkaMessage.toString()), partition);
+                            LOG.debug("Message with guid={} sent to partition={} sent successfully.",kafkaMessage.get("parentTaskVertexId"), partition );


Do we really need one debug log per message? consider removing this & if needed write a generic log outside the for

Applicable for all other similar places

abhijeet-atlan requested review from sumandas0, nikhilbonte21, jnkrmg and hr2904 February 4, 2025 13:36

jnkrmg requested changes Feb 5, 2025

View reviewed changes

hr2904 and others added 7 commits February 5, 2025 09:50

Init script

d046141

PR Fixes #1

288062c

added code to send kafka message to topic 'TAG_PROP_EVENTS'

ae9dec1

added abstract method to send kafka message to a partition

a13f17e

Update maven.yml

d364c8e

corrected delete handler

b9b6887

Merge with master

317322e

abhijeet-atlan force-pushed the tagpropv1.5master branch from 6a0fb40 to 317322e Compare February 5, 2025 04:27

abhijeet-atlan added 2 commits February 5, 2025 10:03

Review comments fixes

7cf3072

Removed dup dependency

ba5d144

abhijeet-atlan requested a review from jnkrmg February 5, 2025 04:36

nikhilbonte21 requested changes Feb 7, 2025

View reviewed changes

manikant-prasad changed the title ~~DG1925 | Send notifications to Kafka topic after planning is done for tag propagation~~ ADMIN-3 | Send notifications to Kafka topic after planning is done for tag propagation Feb 10, 2025

payload change for kafka message

c7c8c4e

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ADMIN-3 | Send notifications to Kafka topic after planning is done for tag propagation #4085

ADMIN-3 | Send notifications to Kafka topic after planning is done for tag propagation #4085

abhijeet-atlan commented Feb 4, 2025 •

edited

Loading

jnkrmg Feb 5, 2025

abhijeet-atlan Feb 5, 2025

nikhilbonte21 Feb 7, 2025

nikhilbonte21 Feb 7, 2025

jnkrmg Feb 14, 2025

jnkrmg Feb 5, 2025

abhijeet-atlan Feb 5, 2025

jnkrmg Feb 5, 2025

abhijeet-atlan Feb 5, 2025

jnkrmg Feb 5, 2025

abhijeet-atlan Feb 5, 2025

jnkrmg Feb 5, 2025

abhijeet-atlan Feb 5, 2025

jnkrmg Feb 5, 2025

abhijeet-atlan Feb 5, 2025

nikhilbonte21 Feb 7, 2025

nikhilbonte21 Feb 7, 2025

nikhilbonte21 Feb 7, 2025

nikhilbonte21 Feb 7, 2025

nikhilbonte21 Feb 7, 2025

nikhilbonte21 Feb 7, 2025

nikhilbonte21 Feb 7, 2025

nikhilbonte21 Feb 7, 2025

nikhilbonte21 Feb 7, 2025

nikhilbonte21 Feb 7, 2025

nikhilbonte21 Feb 7, 2025

nikhilbonte21 Feb 7, 2025

ADMIN-3 | Send notifications to Kafka topic after planning is done for tag propagation #4085

Are you sure you want to change the base?

ADMIN-3 | Send notifications to Kafka topic after planning is done for tag propagation #4085

Conversation

abhijeet-atlan commented Feb 4, 2025 • edited Loading

Change description

Type of change

Related issues

Checklists

Development

Security

Code review

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

abhijeet-atlan commented Feb 4, 2025 •

edited

Loading