Persist OpenLineage event before updating marquez model (#2069)

RunTransitionListener is invoked after updating the marquez model,
but OpenLineage event is not available at that moment. This simply
switch the order of it, that persists OpenLineage event first and
then updates marquez model. With this change, OpenLineage event is
already persisted and available from RunTransitionListener.

Signed-off-by: Minkyu Park <minkyu.park.200@gmail.com>

Signed-off-by: Minkyu Park <minkyu.park.200@gmail.com>

api/src/main/java/marquez/service/OpenLineageService.java

@@ -65,19 +65,6 @@ public OpenLineageService(BaseDao baseDao, RunService runService, Executor execu
   }
 
   public CompletableFuture<Void> createAsync(LineageEvent event) {
-    CompletableFuture<Void> marquez =
-        CompletableFuture.supplyAsync(
-                withSentry(withMdc(() -> updateMarquezModel(event, mapper))), executor)
-            .thenAccept(
-                (update) -> {
-                  if (event.getEventType() != null) {
-                    if (event.getEventType().equalsIgnoreCase("COMPLETE")) {
-                      buildJobOutputUpdate(update).ifPresent(runService::notify);
-                    }
-                    buildJobInputUpdate(update).ifPresent(runService::notify);
-                  }
-                });
-
     UUID runUuid = runUuidFromEvent(event.getRun());
     CompletableFuture<Void> openLineage =
         CompletableFuture.runAsync(
@@ -94,6 +81,19 @@ public CompletableFuture<Void> createAsync(LineageEvent event) {
                             event.getProducer()))),
             executor);
 
+    CompletableFuture<Void> marquez =
+        CompletableFuture.supplyAsync(
+                withSentry(withMdc(() -> updateMarquezModel(event, mapper))), executor)
+            .thenAccept(
+                (update) -> {
+                  if (event.getEventType() != null) {
+                    if (event.getEventType().equalsIgnoreCase("COMPLETE")) {
+                      buildJobOutputUpdate(update).ifPresent(runService::notify);
+                    }
+                    buildJobInputUpdate(update).ifPresent(runService::notify);
+                  }
+                });
+
     return CompletableFuture.allOf(marquez, openLineage);
   }