Add parentRun alias to LineageEvent RunFacet to support older OpenLin…

CHANGELOG.md

@@ -1,6 +1,8 @@
 # Changelog
 
 ## [Unreleased](https://github.com/MarquezProject/marquez/compare/0.26.0...HEAD)
+### Fixed
+* Add support for `parentRun` facet as reported by older Airflow OpenLineage versions [@collado-mike](https://github.com/collado-mike)
 ## [0.26.0](https://github.com/MarquezProject/marquez/compare/0.25.0...0.26.0) - 2022-09-15
 
 ### Added

api/src/main/java/marquez/service/models/LineageEvent.java

@@ -5,6 +5,7 @@
 
 package marquez.service.models;
 
+import com.fasterxml.jackson.annotation.JsonAlias;
 import com.fasterxml.jackson.annotation.JsonAnyGetter;
 import com.fasterxml.jackson.annotation.JsonAnySetter;
 import com.fasterxml.jackson.annotation.JsonIgnore;
@@ -71,7 +72,11 @@ public static class Run extends BaseJsonModel {
   public static class RunFacet {
 
     @Valid private NominalTimeRunFacet nominalTime;
-    @Valid private ParentRunFacet parent;
+
+    @JsonAlias(
+        "parentRun") // the Airflow integration previously reported parentRun instead of parent
+    @Valid
+    private ParentRunFacet parent;
 
     @Builder.Default @JsonIgnore private Map<String, Object> additional = new LinkedHashMap<>();
 

api/src/test/java/marquez/OpenLineageIntegrationTest.java

@@ -31,6 +31,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Optional;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
@@ -235,6 +236,139 @@ public void testOpenLineageJobHierarchyAirflowIntegration()
     assertThat(runsList).isNotEmpty().hasSize(1);
   }
 
+  @Test
+  public void testOpenLineageJobHierarchyAirflowIntegrationWithParentRunFacet()
+      throws ExecutionException, InterruptedException, TimeoutException {
+    OpenLineage ol = new OpenLineage(URI.create("http://openlineage.test.com/"));
+    ZonedDateTime startOfHour =
+        Instant.now()
+            .atZone(LineageTestUtils.LOCAL_ZONE)
+            .with(ChronoField.MINUTE_OF_HOUR, 0)
+            .with(ChronoField.SECOND_OF_MINUTE, 0);
+    ZonedDateTime endOfHour = startOfHour.plusHours(1);
+    String airflowParentRunId = UUID.randomUUID().toString();
+    String task1Name = "task1";
+    String task2Name = "task2";
+    String dagName = "the_dag";
+    RunEvent airflowTask1 =
+        createAirflowRunEvent(
+            ol,
+            startOfHour,
+            endOfHour,
+            airflowParentRunId,
+            dagName,
+            dagName + "." + task1Name,
+            NAMESPACE_NAME);
+
+    // the older airflow integration reported parentRun instead of parent. We support this as an
+    // alias for compatibility
+    RunFacet parent = airflowTask1.getRun().getFacets().getAdditionalProperties().remove("parent");
+    airflowTask1.getRun().getFacets().getAdditionalProperties().put("parentRun", parent);
+
+    RunEvent airflowTask2 =
+        createAirflowRunEvent(
+            ol,
+            startOfHour,
+            endOfHour,
+            airflowParentRunId,
+            dagName,
+            dagName + "." + task2Name,
+            NAMESPACE_NAME);
+    parent = airflowTask2.getRun().getFacets().getAdditionalProperties().remove("parent");
+    airflowTask2.getRun().getFacets().getAdditionalProperties().put("parentRun", parent);
+
+    CompletableFuture<Integer> future = sendAllEvents(airflowTask1, airflowTask2);
+    future.get(5, TimeUnit.SECONDS);
+
+    Job job = client.getJob(NAMESPACE_NAME, dagName + "." + task1Name);
+    assertThat(job)
+        .isNotNull()
+        .hasFieldOrPropertyWithValue("id", new JobId(NAMESPACE_NAME, dagName + "." + task1Name))
+        .hasFieldOrPropertyWithValue("parentJobName", dagName);
+
+    Job parentJob = client.getJob(NAMESPACE_NAME, dagName);
+    assertThat(parentJob)
+        .isNotNull()
+        .hasFieldOrPropertyWithValue("id", new JobId(NAMESPACE_NAME, dagName))
+        .hasFieldOrPropertyWithValue("parentJobName", null);
+    List<Run> runsList = client.listRuns(NAMESPACE_NAME, dagName);
+    assertThat(runsList).isNotEmpty().hasSize(1);
+  }
+
+  @Test
+  public void testOpenLineageJobHierarchyAirflowIntegrationWithParentAndParentRunFacet()
+      throws ExecutionException, InterruptedException, TimeoutException {
+    OpenLineage ol = new OpenLineage(URI.create("http://openlineage.test.com/"));
+    ZonedDateTime startOfHour =
+        Instant.now()
+            .atZone(LineageTestUtils.LOCAL_ZONE)
+            .with(ChronoField.MINUTE_OF_HOUR, 0)
+            .with(ChronoField.SECOND_OF_MINUTE, 0);
+    ZonedDateTime endOfHour = startOfHour.plusHours(1);
+    String airflowParentRunId = UUID.randomUUID().toString();
+    String task1Name = "task1";
+    String task2Name = "task2";
+    String dagName = "the_dag";
+    RunEvent airflowTask1 =
+        createAirflowRunEvent(
+            ol,
+            startOfHour,
+            endOfHour,
+            airflowParentRunId,
+            dagName,
+            dagName + "." + task1Name,
+            NAMESPACE_NAME);
+
+    // the older airflow integration reported parentRun instead of parent. The new integration
+    // reports both. They are the same in the airflow integration, but this test verifies we handle
+    // the "parentRun" field first.
+    // It would be preferable to prioritize the "parent" field, but it seems Jackson prefers the
+    // alias first.
+    RunFacet parent = airflowTask1.getRun().getFacets().getAdditionalProperties().get("parent");
+    RunFacet newParent = ol.newRunFacet();
+    Map<String, Object> runFacetProps = newParent.getAdditionalProperties();
+    runFacetProps.put("run", parent.getAdditionalProperties().get("run"));
+    runFacetProps.put(
+        "job", ImmutableMap.of("name", "a_new_dag", "namespace", "incorrect_namespace"));
+    airflowTask1.getRun().getFacets().getAdditionalProperties().put("parentRun", parent);
+    airflowTask1.getRun().getFacets().getAdditionalProperties().put("parent", newParent);
+
+    RunEvent airflowTask2 =
+        createAirflowRunEvent(
+            ol,
+            startOfHour,
+            endOfHour,
+            airflowParentRunId,
+            dagName,
+            dagName + "." + task2Name,
+            NAMESPACE_NAME);
+    parent = airflowTask2.getRun().getFacets().getAdditionalProperties().get("parent");
+    newParent = ol.newRunFacet();
+    runFacetProps = newParent.getAdditionalProperties();
+    runFacetProps.put("run", parent.getAdditionalProperties().get("run"));
+    runFacetProps.put(
+        "job", ImmutableMap.of("name", "a_new_dag", "namespace", "incorrect_namespace"));
+    airflowTask2.getRun().getFacets().getAdditionalProperties().put("parentRun", parent);
+    airflowTask2.getRun().getFacets().getAdditionalProperties().put("parent", newParent);
+
+    CompletableFuture<Integer> future = sendAllEvents(airflowTask1, airflowTask2);
+    future.get(5, TimeUnit.SECONDS);
+
+    Job job = client.getJob(NAMESPACE_NAME, dagName + "." + task1Name);
+    assertThat(job)
+        .isNotNull()
+        .hasFieldOrPropertyWithValue("id", new JobId(NAMESPACE_NAME, dagName + "." + task1Name))
+        .hasFieldOrPropertyWithValue("parentJobName", dagName);
+
+    Job parentJob = client.getJob(NAMESPACE_NAME, dagName);
+    assertThat(parentJob)
+        .isNotNull()
+        .hasFieldOrPropertyWithValue("id", new JobId(NAMESPACE_NAME, dagName))
+        .hasFieldOrPropertyWithValue("parentJobName", null);
+    List<Run> runsList = client.listRuns(NAMESPACE_NAME, dagName);
+    assertThat(runsList).isNotEmpty().hasSize(1);
+  }
+
   @Test
   public void testOpenLineageJobHierarchyAirflowIntegrationWithDagNameWithDot()
       throws ExecutionException, InterruptedException, TimeoutException {