kacpermuda commented on code in PR #74752:
URL: https://github.com/apache/airflow/pull/74752#discussion_r4315006986
##########
providers/openlineage/tests/system/openlineage/operator.py:
##########
@@ -351,33 +479,44 @@ def execute(self, context: Context) -> None:
else:
self.event_templates[key] = [existing, event]
errors: dict[str, str] = {}
+ events_by_key: dict[str, list[dict]] = {}
try:
for key, template in self.event_templates.items(): # type:
ignore[union-attr]
log.info("Checking key: `%s`", key)
try:
- self._check_key(key, template)
+ events_by_key[key] = self._check_key(key, template)
Review Comment:
Agreed, fixed in 7441d2d: the runId check is skipped, with a log line, once
any key failed its template.
---
Drafted-by: Claude Code (Opus 5.5) (no human review before posting)
##########
providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py:
##########
@@ -263,15 +263,17 @@ def on_running():
clear_number=clear_number,
)
- task_uuid = get_regional_task_instance_run_id(
- task_instance
- ) or self.adapter.build_task_instance_run_id(
- dag_id=task_instance.dag_id,
- task_id=task_instance.task_id,
- try_number=task_instance.try_number,
- logical_date=date,
- map_index=task_instance.map_index,
- )
+ position = get_task_instance_position(task_instance)
Review Comment:
Fixed in 7441d2d: one `get_task_instance_run_id` helper, used by the four
worker hooks, the manual state change path and the `lineage_run_id` macro.
---
Drafted-by: Claude Code (Opus 5.5) (no human review before posting)
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -1135,11 +1193,44 @@ class TaskInstanceInfo(InfoJsonEncodable):
"dag_bundle_name": lambda ti: ti.bundle_instance.name if hasattr(ti,
"bundle_instance") else None,
}
+ def __init__(self, obj, position: TaskInstancePosition | None = None):
+ # Pass the position when the caller has already placed the task
instance, e.g. from its stored region
+ self._position = position
+ super().__init__(obj)
+
def _extend_fields(self) -> None:
- map_index = get_task_instance_map_index(self.obj)
+ # A stored task instance whose task isn't attached can't be placed
without its region, and a facet must
+ # not fail the whole event, so then it describes the task instance
without its position.
+ position = self._position or _get_task_instance_position(self.obj)
Review Comment:
Fixed in 7441d2d: a position passed in is used as is; the facet only looks
up the iteration itself when no position was given.
---
Drafted-by: Claude Code (Opus 5.5) (no human review before posting)
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -1123,9 +1123,67 @@ def team_name(cls, dagrun: DagRun) -> str | None:
class TaskInstanceInfo(InfoJsonEncodable):
Review Comment:
Trimmed in 7441d2d: `TaskInstancePosition` now only describes its own fields
and points to `TaskInstanceInfo`, which keeps the full explanation for readers
of the code, while the schema keeps short per-field descriptions for consumers.
---
Drafted-by: Claude Code (Opus 5.5) (no human review before posting)
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -2049,6 +2224,11 @@ def _get_task_groups_details(dag: DAG | SerializedDAG,
edge_map: dict[str, tuple
"ui_label": tg.label,
"downstream_task_edges": task_edges,
"downstream_group_edges": group_edges,
+ **{
+ field: getattr(tg, field) # Only a loop group has these
+ for field in ("gate_task_id", "has_until", "max_iterations",
"terminal_task_id")
Review Comment:
Added in 7441d2d, to the `task_group` definition in `AirflowRunFacet.json`
and to `taskGroups` in `AirflowJobFacet.json`.
---
Drafted-by: Claude Code (Opus 5.5) (no human review before posting)
--
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]
