ATLAS-4922: Atlas Async Import using Kafka #307

DishaTalreja3 · 2025-03-12T18:40:30Z

What changes were proposed in this pull request?

The existing synchronous import API causes performance bottlenecks with large or multiple requests. This patch introduces asynchronous imports using Kafka to queue requests via dedicated topics, enabling users to submit multiple import requests concurrently.

New APIs added to:

Perform asynchronous imports
Check import status (all operations or a specific import ID)
Abort a queued import operation

How was this patch tested?

Manual Testing, Unit Tests

…otification changes

mneethiraj

@DishaTalreja3 - I haven't completed review yet; sending comments so far. Please review and update the patch.

atlas-examples/sample-app/src/main/java/org/apache/atlas/examples/sampleapp/SampleApp.java

client/client-v2/src/main/java/org/apache/atlas/AtlasClientV2.java

mneethiraj · 2025-03-13T23:09:57Z

webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java

@@ -170,6 +177,7 @@ public class NotificationHookConsumer implements Service, ActiveStateChangeHandl
    private static final int    KAFKA_CONSUMER_SHUTDOWN_WAIT                    = 30000;
    private static final String ATLAS_HOOK_CONSUMER_THREAD_NAME                 = "atlas-hook-consumer-thread";
    private static final String ATLAS_HOOK_UNSORTED_CONSUMER_THREAD_NAME        = "atlas-hook-unsorted-consumer-thread";
+    private static final String ATLAS_IMPORT_CONSUMER_THREAD_PREFIX             = "atlas-import-hook-consumer-thread-";


atlas-import-hook-consumer-thread- => atlas-import-consumer-thread-

mneethiraj · 2025-03-13T23:16:40Z

webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java

+                    consumersIterator.remove();
+                }
+            }
+            notificationInterface.closeConsumer(ASYNC_IMPORT);


This would close all consumers of type ASYNC_IMPORT. Shouldn't only consumers of topic be closed instead? Perhaps only one consumer will be active for type ASYNC_IMPORT in the current implementation status; but this part of code should better handle multiple simultaneous imports.

mneethiraj · 2025-03-13T23:20:39Z

webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java


-            startConsumers(executorService);
+    public void startImportNotificationConsumer(NotificationType notificationType, String importId, String topic) {


startImportNotificationConsumer(notificationType, importId, topic) ==> startAsyncImportConsumer(importId, topic)

mneethiraj · 2025-03-13T23:23:03Z

webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java

+            ListIterator<HookConsumer> consumersIterator = consumers.listIterator();
+            while (consumersIterator.hasNext()) {
+                HookConsumer consumer = consumersIterator.next();
+                if (consumer.getName().contains(consumerName)) {


contains(consumerName) => startsWith(consumerName)

mneethiraj · 2025-03-13T23:32:59Z

webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java

+                    new SynchronousQueue<>(), // Direct handoff queue
+                    new ThreadFactoryBuilder().setNameFormat(THREADNAME_PREFIX + " thread-%d").build());
+        }
+        executors = executorService;


Review all writes to executors i.e. following methods:

startInternal()

startConsumers()

stop()

Prior to this patch, these methods are to be called only once in an Atlas instance. However, now executors can be overwritten for every call to async import. Review the usage carefully and update to avoid inappropriate overwrites.

In this new approach the executors are overWritten in
startConsumers() - only when the passed in executorService is null (essentially at the start of the instance or start of the first async import request). Please correct me f I am wrong.

…e PR comments

mneethiraj · 2025-03-14T21:15:37Z

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

@@ -224,7 +224,7 @@ public enum AtlasErrorCode {
    GLOSSARY_IMPORT_FAILED(409, "ATLAS-409-00-011", "Glossary import failed"),
    METRICSSTAT_ALREADY_EXISTS(409, "ATLAS-409-00-012", "Metric Statistics already collected at {0}"),
    PENDING_TASKS_ALREADY_IN_PROGRESS(409, "ATLAS-409-00-013", "There are already {0} pending tasks in queue"),
-    IMPORT_ALREADY_IN_PROGRESS(409, "ATLAS-409-00-016", "Given import request {0} is already in progress or completed"),
+    ABORT_IMPORT(409, "ATLAS-409-00-016", "Import id={0} is currently in state {1}, cannot be aborted"),


ABORT_IMPORT => IMPORT_ABORT_NOT_ALLOWED

intg/src/main/java/org/apache/atlas/model/impexp/AtlasAsyncImportRequest.java

mneethiraj · 2025-03-14T23:11:44Z

intg/src/main/java/org/apache/atlas/model/impexp/AtlasAsyncImportRequest.java

+@JsonIgnoreProperties(ignoreUnknown = true)
+public class AtlasAsyncImportRequest extends AtlasBaseModelObject implements Serializable {
+    private static final long serialVersionUID = 1L;
+    public static final String ASYNC_IMPORT_TYPE_NAME = "__AtlasAsyncImportRequest";


Constants defined in line 47-49 seem to be used/needed only in Atlas server side. If yes, please move these to server side (like repository module).

These constants are used in this model class as well. Moving them to the repository module might introduce unnecessary dependencies.

None of the classes in model package referenes ASYNC_IMPORT_TYPE_NAME. This is referenced only from AsyncImportService and AtlasAsyncImportRequestDTO.

mneethiraj · 2025-03-14T23:21:23Z

intg/src/main/java/org/apache/atlas/model/impexp/AtlasAsyncImportRequest.java

+    }
+
+    @JsonIgnoreProperties(ignoreUnknown = true)
+    public static class ImportDetails {


add implements Serializable

failures is marked as final. How does this work drring deserialization?

failures is initialized with ConcurrentHashMap. Is this necessary? Json deserializer might assign a Hashmap instance.

mneethiraj · 2025-03-14T23:38:48Z

intg/src/main/java/org/apache/atlas/model/notification/ImportNotification.java

+@JsonIgnoreProperties(ignoreUnknown = true)
+@XmlRootElement
+@XmlAccessorType(XmlAccessType.PROPERTY)
+public class ImportNotification extends HookNotification implements Serializable {


ImportNotification is not instantiated anywhere. Consider following updates:

make ImportNotification constructor as protected

have AtlasTypeDefImportNotification and AtlasEntityImportNotification extend ImportNotification

move importId field to ImportNotification

mneethiraj · 2025-03-14T23:41:42Z

intg/src/main/java/org/apache/atlas/model/notification/ImportNotification.java

+        private String importId;
+
+        @JsonProperty
+        private AtlasEntityWithExtInfo entities;


entities => entity

mneethiraj · 2025-03-14T23:46:44Z

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

-        KafkaProducer producer = getOrCreateProducer(topic);
-
-        sendInternalToProducer(producer, topic, messages);
+        sendInternal(topic, messages, SORT_NOT_NEEDED);


ok. I get it now. SORT_NOT_NEEDED doesn't add much to readability, given the called method is right above. I suggest to simply send false here.

notification/src/main/java/org/apache/atlas/notification/rest/RestNotification.java

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

notification/src/test/java/org/apache/atlas/notification/spool/AtlasFileSpoolTest.java

…e PR comments

…sting override

…mments and made improvements

mneethiraj · 2025-03-25T00:32:57Z

intg/src/main/java/org/apache/atlas/model/impexp/AsyncImportStatus.java

+public class AsyncImportStatus {
+    private String importId;
+    private AtlasAsyncImportRequest.ImportStatus status;
+    private String importRequestReceivedAt;


should importRequestReceivedAt be of type Date?

I kept importRequestReceivedAt as a String to avoid the default timestamp serialization of Date and to ensure a more readable format.

mneethiraj · 2025-03-25T00:33:37Z

intg/src/main/java/org/apache/atlas/model/impexp/AsyncImportStatus.java

+    private String importId;
+    private AtlasAsyncImportRequest.ImportStatus status;
+    private String importRequestReceivedAt;
+    private String importRequestReceivedBy;


What does field importRequestReceivedBy capture? Is it username or hostname or anything else?

It's the username

importRequestReceivedBy is username? Which user would it be? Atlas server principal?

It is the requesting user, retrieved using Servlets.getUserName(httpServletRequest). This is also tracked in the existing import API and is part of the AtlasImportResult object.

intg/src/main/java/org/apache/atlas/model/impexp/AsyncImportStatus.java

mneethiraj · 2025-03-25T00:37:11Z

intg/src/main/java/org/apache/atlas/model/impexp/AtlasAsyncImportRequest.java

+@JsonIgnoreProperties(ignoreUnknown = true)
+public class AtlasAsyncImportRequest extends AtlasBaseModelObject implements Serializable {
+    private static final long serialVersionUID = 1L;
+    public static final String ASYNC_IMPORT_TYPE_NAME = "__AtlasAsyncImportRequest";


None of the classes in model package referenes ASYNC_IMPORT_TYPE_NAME. This is referenced only from AsyncImportService and AtlasAsyncImportRequestDTO.

intg/src/main/java/org/apache/atlas/model/impexp/AtlasAsyncImportRequest.java

intg/src/main/java/org/apache/atlas/model/notification/ImportNotification.java

intg/src/main/java/org/apache/atlas/utils/AtlasAsyncImportTestUtil.java

intg/src/test/java/org/apache/atlas/utils/TestAtlasAsyncImportTestUtil.java

repository/src/main/java/org/apache/atlas/repository/impexp/AsyncImportService.java

repository/src/main/java/org/apache/atlas/repository/impexp/ImportService.java

...ory/src/main/java/org/apache/atlas/repository/store/graph/v2/bulkimport/MigrationImport.java

webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java

… the same

…ortTaskListenerImplTest

DishaTalreja3 and others added 3 commits February 26, 2025 16:27

ATLAS-4922: Atlas Async Import using Kafka [1] - POJO & Data Access

2e9fba9

ATLAS-4922: Atlas Async Import using Kafka [2] - Client-side changes

3584a20

ATLAS-4922: Atlas Async Import using Kafka [3] - Implementation and N…

16b5449

…otification changes

DishaTalreja3 marked this pull request as ready for review March 12, 2025 18:41

rkundam requested review from rkundam, mneethiraj and pinal-shah March 12, 2025 20:33

ATLAS-4922: Atlas Async Import using Kafka [4] - Unit Tests

d678555

DishaTalreja3 force-pushed the ASYNC_IMPORT branch from a7dabe8 to d678555 Compare March 12, 2025 23:25

mneethiraj reviewed Mar 13, 2025

View reviewed changes

jackhalfalltrades force-pushed the ASYNC_IMPORT branch from 8c56a08 to 5a3a995 Compare March 14, 2025 19:08

ATLAS-4922: Atlas Async Import using Kafka [5] - Addressing few of th…

69ecd1b

…e PR comments

jackhalfalltrades force-pushed the ASYNC_IMPORT branch from 5a3a995 to 69ecd1b Compare March 14, 2025 19:24

mneethiraj reviewed Mar 14, 2025

View reviewed changes

Chandra Kanth Peravelli and others added 3 commits March 17, 2025 14:27

ATLAS-4922: Atlas Async Import using Kafka [6] - Addressing few of th…

6ae366a

…e PR comments

ATLAS-4922: Atlas Async Import using Kafka [7] - Adding support to te…

88ada02

…sting override

ATLAS-4922: Atlas Async Import using Kafka [8] - Addressed some PR co…

0f215e7

…mments and made improvements

DishaTalreja3 force-pushed the ASYNC_IMPORT branch from 7b2eb97 to 0f215e7 Compare March 25, 2025 03:31

addressed review comments

8fe6be6

mneethiraj reviewed Mar 25, 2025

View reviewed changes

addressed review comments

a2c731f

mneethiraj reviewed Mar 25, 2025

View reviewed changes

repository/src/main/java/org/apache/atlas/repository/impexp/ImportService.java Show resolved Hide resolved

repository/src/main/java/org/apache/atlas/repository/impexp/ImportService.java Show resolved Hide resolved

mneethiraj reviewed Mar 25, 2025

View reviewed changes

...ory/src/main/java/org/apache/atlas/repository/store/graph/v2/bulkimport/MigrationImport.java Outdated Show resolved Hide resolved

webapp/src/main/java/org/apache/atlas/notification/NotificationHookConsumer.java Outdated Show resolved Hide resolved

mneethiraj and others added 5 commits March 25, 2025 01:05

addressed review comments

94ff706

addressed review comments

bfc3c15

fixed unit tests

06507ee

code readability improvements

3d66b3f

fixed graph query

ff7c0ec

jackhalfalltrades force-pushed the ASYNC_IMPORT branch from af12b9f to 099b166 Compare March 27, 2025 14:21

ATLAS-4922: Ensuring executors are not overwritten and tests to prove…

83ed723

… the same

jackhalfalltrades force-pushed the ASYNC_IMPORT branch from 099b166 to 83ed723 Compare March 28, 2025 00:57

Chandra Kanth Peravelli and others added 8 commits March 28, 2025 01:56

ATLAS-4922: Fix failing tests

128bf1b

Fix getTopicName()

3a4c62a

ATLAS-4922: Fix flaky tests

6ccc2ae

ATLAS-4922: Fix failing tests

580a0c2

ATLAS-4922: Fix stuck in processing state in case of no topic found

484f27d

ATLAS-4922: Fix enunciate build errors

21d62ab

ATLAS-4922: Fix test failures in NotificationHookConsumerTest and Imp…

a65a5f8

…ortTaskListenerImplTest

Merge branch 'apache:master' into ASYNC_IMPORT

cf85053

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

ATLAS-4922: Atlas Async Import using Kafka #307

ATLAS-4922: Atlas Async Import using Kafka #307

DishaTalreja3 commented Mar 12, 2025

mneethiraj left a comment

mneethiraj Mar 13, 2025

mneethiraj Mar 13, 2025

mneethiraj Mar 13, 2025

mneethiraj Mar 13, 2025

mneethiraj Mar 13, 2025

jackhalfalltrades Mar 14, 2025

mneethiraj Mar 14, 2025

mneethiraj Mar 14, 2025

DishaTalreja3 Mar 20, 2025

mneethiraj Mar 25, 2025

mneethiraj Mar 14, 2025

mneethiraj Mar 14, 2025

mneethiraj Mar 14, 2025

mneethiraj Mar 14, 2025

mneethiraj Mar 25, 2025

DishaTalreja3 Mar 25, 2025

mneethiraj Mar 25, 2025

DishaTalreja3 Mar 25, 2025

mneethiraj Mar 25, 2025

DishaTalreja3 Mar 25, 2025

mneethiraj Mar 25, 2025


		startConsumers(executorService);
		public void startImportNotificationConsumer(NotificationType notificationType, String importId, String topic) {

ATLAS-4922: Atlas Async Import using Kafka #307

Are you sure you want to change the base?

ATLAS-4922: Atlas Async Import using Kafka #307

Conversation

DishaTalreja3 commented Mar 12, 2025

What changes were proposed in this pull request?

How was this patch tested?

mneethiraj left a 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