Runless events - consume dataset event (#2641)

* Runless events - consume dataset event Signed-off-by: Pawel Leszczynski <leszczynski.pawel@gmail.com> * Runless events - add container to store daos Signed-off-by: Pawel Leszczynski <leszczynski.pawel@gmail.com> * Runless events - extract common methods Signed-off-by: Pawel Leszczynski <leszczynski.pawel@gmail.com> * Runless events - run upsert builder Signed-off-by: Pawel Leszczynski <leszczynski.pawel@gmail.com> * Runless events - review feedback Signed-off-by: Pawel Leszczynski <leszczynski.pawel@gmail.com> * fix daos container - speeds up tests twice Signed-off-by: Pawel Leszczynski <leszczynski.pawel@gmail.com> --------- Signed-off-by: Pawel Leszczynski <leszczynski.pawel@gmail.com> Co-authored-by: Willy Lulciuc <willy@datakin.com>
MarquezProject · Nov 6, 2023 · 42cadbb · 42cadbb
1 parent 7c19162
commit 42cadbb
Show file tree

Hide file tree

Showing 14 changed files with 646 additions and 268 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -26,6 +26,10 @@
 * Web: fix Unix epoch time display for null `endedAt` values [`#2647`](https://github.com/MarquezProject/marquez/pull/2647) [@merobi-hub](https://github.com/merobi-hub)  
     *Fixes the issue of the GUI displaying Unix epoch time (midnight on January 1, 1970) in the case of running jobs/null `endedAt` values.*
 
+### Added
+* API: support `DatasetEvent` [`#2641`](https://github.com/MarquezProject/marquez/pull/2641) [@pawel-big-lebowski]( https://github.com/pawel-big-lebowski)
+    *Save into Marquez model datasets sent via `DatasetEvent` event type 
+
 ## [0.41.0](https://github.com/MarquezProject/marquez/compare/0.40.0...0.41.0) - 2023-09-20
 ### Added
 * API: add support for the following parameters in the `SearchDao` [`#2556`](https://github.com/MarquezProject/marquez/pull/2556) [@tati](https://github.com/tati) [@wslulciuc](https://github.com/wslulciuc)  

diff --git a/api/src/main/java/marquez/api/OpenLineageResource.java b/api/src/main/java/marquez/api/OpenLineageResource.java
@@ -39,6 +39,7 @@
 import marquez.db.OpenLineageDao;
 import marquez.service.ServiceFactory;
 import marquez.service.models.BaseEvent;
+import marquez.service.models.DatasetEvent;
 import marquez.service.models.LineageEvent;
 import marquez.service.models.NodeId;
 
@@ -67,15 +68,11 @@ public void create(@Valid @NotNull BaseEvent event, @Suspended final AsyncRespon
     if (event instanceof LineageEvent) {
       openLineageService
           .createAsync((LineageEvent) event)
-          .whenComplete(
-              (result, err) -> {
-                if (err != null) {
-                  log.error("Unexpected error while processing request", err);
-                  asyncResponse.resume(Response.status(determineStatusCode(err)).build());
-                } else {
-                  asyncResponse.resume(Response.status(201).build());
-                }
-              });
+          .whenComplete((result, err) -> onComplete(result, err, asyncResponse));
+    } else if (event instanceof DatasetEvent) {
+      openLineageService
+          .createAsync((DatasetEvent) event)
+          .whenComplete((result, err) -> onComplete(result, err, asyncResponse));
     } else {
       log.warn("Unsupported event type {}. Skipping without error", event.getClass().getName());
 
@@ -84,6 +81,15 @@ public void create(@Valid @NotNull BaseEvent event, @Suspended final AsyncRespon
     }
   }
 
+  private void onComplete(Void result, Throwable err, AsyncResponse asyncResponse) {
+    if (err != null) {
+      log.error("Unexpected error while processing request", err);
+      asyncResponse.resume(Response.status(determineStatusCode(err)).build());
+    } else {
+      asyncResponse.resume(Response.status(201).build());
+    }
+  }
+
   private int determineStatusCode(Throwable e) {
     if (e instanceof CompletionException) {
       return determineStatusCode(e.getCause());

diff --git a/api/src/main/java/marquez/db/DatasetFacetsDao.java b/api/src/main/java/marquez/db/DatasetFacetsDao.java
@@ -12,6 +12,7 @@
 import java.util.Spliterators;
 import java.util.UUID;
 import java.util.stream.StreamSupport;
+import javax.annotation.Nullable;
 import lombok.NonNull;
 import marquez.common.Utils;
 import marquez.service.models.LineageEvent;
@@ -126,9 +127,9 @@ void insertDatasetFacet(
   default void insertDatasetFacetsFor(
       @NonNull UUID datasetUuid,
       @NonNull UUID datasetVersionUuid,
-      @NonNull UUID runUuid,
+      @Nullable UUID runUuid,
       @NonNull Instant lineageEventTime,
-      @NonNull String lineageEventType,
+      @Nullable String lineageEventType,
       @NonNull LineageEvent.DatasetFacets datasetFacets) {
     final Instant now = Instant.now();