jedcunningham commented on code in PR #45860:
URL: https://github.com/apache/airflow/pull/45860#discussion_r1934573873
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -2001,11 +2005,13 @@ def _find_zombies(self, *, session: Session) ->
list[tuple[TI, str]]:
self.log.warning("Failing %s TIs without heartbeat after %s",
len(zombies), limit_dttm)
return zombies
- def _purge_zombies(self, zombies: list[tuple[TI, str]], *, session:
Session) -> None:
- for ti, file_loc in zombies:
+ def _purge_zombies(self, zombies: list[TI], *, session: Session) -> None:
+ for ti in zombies:
zombie_message_details = self._generate_zombie_message_details(ti)
request = TaskCallbackRequest(
- full_filepath=file_loc,
+ filepath=str(ti.dag_model.fileloc),
+ bundle_name=ti.dag_model.bundle_name,
+ bundle_version=ti.dag_model.bundle_version,
Review Comment:
And here too.
##########
airflow/triggers/base.py:
##########
@@ -215,9 +215,10 @@ def _submit_callback_if_necessary(self, *, task_instance:
TaskInstance, session)
"""Submit a callback request if the task state is SUCCESS or FAILED."""
if self.task_instance_state in (TaskInstanceState.SUCCESS,
TaskInstanceState.FAILED):
request = TaskCallbackRequest(
- full_filepath=task_instance.dag_model.fileloc,
+ filepath=task_instance.dag_model.fileloc,
ti=task_instance,
task_callback_type=self.task_instance_state,
+ bundle_name=task_instance.dag_model.bundle_name,
Review Comment:
It's here now :)
##########
airflow/callbacks/callback_requests.py:
##########
@@ -34,8 +34,10 @@ class BaseCallbackRequest(BaseModel):
:param msg: Additional Message that can be used for logging
"""
- full_filepath: str
+ filepath: str
"""File Path to use to run the callback"""
+ bundle_name: str
+ bundle_version: str | None = None
Review Comment:
Isn't that what we want though? Like it can be None if the bundle doesn't do
versioning, but the caller should tell us. Having a default just makes it
easier to miss (like it was missed in the triggerer).
##########
airflow/jobs/scheduler_job_runner.py:
##########
@@ -850,7 +850,9 @@ def process_executor_events(
ti.task = task
if task.on_retry_callback or task.on_failure_callback:
request = TaskCallbackRequest(
- full_filepath=ti.dag_model.fileloc,
+ filepath=str(ti.dag_model.fileloc),
+ bundle_name=ti.dag_model.bundle_name,
+ bundle_version=ti.dag_model.bundle_version,
Review Comment:
This should be the from dagrun too.
##########
airflow/triggers/base.py:
##########
@@ -215,9 +215,11 @@ def _submit_callback_if_necessary(self, *, task_instance:
TaskInstance, session)
"""Submit a callback request if the task state is SUCCESS or FAILED."""
if self.task_instance_state in (TaskInstanceState.SUCCESS,
TaskInstanceState.FAILED):
request = TaskCallbackRequest(
- full_filepath=task_instance.dag_model.fileloc,
+ filepath=task_instance.dag_model.fileloc,
ti=task_instance,
task_callback_type=self.task_instance_state,
+ bundle_name=task_instance.dag_model.bundle_name,
+ bundle_version=task_instance.dag_model.bundle_version,
Review Comment:
We want the version from the dagrun, not the dag itself.
--
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]
