kacpermuda commented on code in PR #75328:
URL: https://github.com/apache/airflow/pull/75328#discussion_r4390401724
##########
providers/openlineage/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -1209,17 +1209,56 @@ def get_rendered_map_index(ti: TaskInstance |
RuntimeTaskInstance) -> str | None
return getattr(ti, "rendered_map_index", None)
def __init__(
- self, obj, position: TaskInstancePosition | None = None, dag: Any =
None, dag_run: Any = None
+ self,
+ obj,
+ position: TaskInstancePosition | None = None,
+ dag: Any = None,
+ dag_run: Any = None,
+ end_date: datetime.datetime | None = None,
):
# Pass the position when the caller has already placed the task
instance, e.g. from its stored region or
# with get_task_instance_position; it is then used as is.
self._position = position
# A stored task instance's bundle comes from these, as the worker's
does, without a query
self._dag = dag
self._dag_run = dag_run
+ # The end date a terminal event is timed with, used when the task
instance carries none, as a skipped
+ # Airflow 3 runtime task instance doesn't before Airflow 3.4.
+ self._end_date = end_date
super().__init__(obj)
+ def _get_end_date(self) -> datetime.datetime | None:
+ if isinstance(end_date := getattr(self.obj, "end_date", None),
datetime.datetime):
+ return end_date
+ return self._end_date
+
+ def _get_duration(
+ self, start_date: datetime.datetime | None, end_date:
datetime.datetime | None
+ ) -> float | None:
+ # The stored duration is what the UI shows, so it wins. A runtime task
instance has none -- the API
+ # server computes it after the listener ran.
+ if (duration := getattr(self.obj, "duration", None)) is not None:
+ return duration
+ if start_date is None or end_date is None:
+ return None
+ try:
+ return (end_date - start_date).total_seconds()
Review Comment:
Agreed, a value we know is wrong is worse than none. One correction to the
description: #63247 was backported to 3.2.2, so 3.0 through 3.2.1 are affected,
not everything before 3.3.
The runtime task instance can tell it resumed on every 3.x version:
`_ti_context_from_server.next_method` is set then. Before 3.2.2, when it's set,
the facet now leaves out `start_date`, because that's the resume time too.
Without a start date, no duration is computed. `end_date` is still included.
The worker's events for tasks that didn't defer are unchanged. The schema now
says when `start_date` is absent.
--
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]
