ieCecchetti commented on code in PR #75303:
URL: https://github.com/apache/airflow/pull/75303#discussion_r4336743637
##########
providers/standard/src/airflow/providers/standard/utils/openlineage.py:
##########
@@ -71,18 +72,26 @@ def _get_openlineage_parent_info(ti: TaskInstance |
RuntimeTI) -> dict[str, str]
lineage_run_id,
)
- return {
+ parent_info: dict[str, Any] = {
"parentRunId": lineage_run_id(ti),
"parentJobName": lineage_job_name(ti),
"parentJobNamespace": lineage_job_namespace(),
"rootParentRunId": lineage_root_run_id(ti),
"rootParentJobName": lineage_root_job_name(ti),
"rootParentJobNamespace": lineage_root_job_namespace(ti),
}
+ dag_run = getattr(ti, "dag_run", None)
+ if dag_run is None:
+ dag_run = ti.get_template_context()["dag_run"]
+ if isinstance(dag_run.conf, dict):
+ ol_conf = dag_run.conf.get("openlineage")
+ if isinstance(ol_conf, dict) and isinstance(root_job_type :=
ol_conf.get("rootParentJobType"), dict):
Review Comment:
Hey, I was checking this pr cause I was curios about it. I just found a case
that can create a problem:
1. An outside system starts pipeline B and passes "started by job X, run id
`abc`, type STREAMING".
2. `abc` isn't a valid id, so the lineage helpers reject X as the starter
and treat B as the start of the chain.
3. B triggers C. C is told "the chain started with B", but this PR still
adds "type STREAMING", which was X's type.
4. C's lineage now pairs B's name with X's job type.
maybe can be useful?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
