arorasachin9 commented on issue #60517:
URL: https://github.com/apache/airflow/issues/60517#issuecomment-3750945984
@vincbeck ,
I think the method `execute_complete` will never get called as the exception
is thrown before hand only. So handling the way it is handled in the attached
PR might not work. Let me know if my understanding is wrong
```
def resume_execution(self, next_method: str, next_kwargs: dict[str, Any] |
None, context: Context):
"""Entrypoint method called by the Task Runner (instead of execute)
when this task is resumed."""
from airflow.sdk.exceptions import TaskDeferralError,
TaskDeferralTimeout
if next_kwargs is None:
next_kwargs = {}
# __fail__ is a special signal value for next_method that indicates
# this task was scheduled specifically to fail.
if next_method == TRIGGER_FAIL_REPR:
next_kwargs = next_kwargs or {}
traceback = next_kwargs.get("traceback")
if traceback is not None:
self.log.error("Trigger failed:\n%s", "\n".join(traceback))
if (error := next_kwargs.get("error", "Unknown")) ==
TriggerFailureReason.TRIGGER_TIMEOUT:
raise TaskDeferralTimeout(error)
raise TaskDeferralError(error)
# Grab the callable off the Operator/Task and add in any kwargs
execute_callable = getattr(self, next_method)
return execute_callable(context, **next_kwargs)
```
This function gets called on the resume of the task after timeout. Here
TaskDeferralTimeout(error) is raised before calling execute_complete. So It
will not help if I handle here.
Every trigger in the provider implements AwsBaseWaiterTrigger. I need to
paas the cancel method to this particular class to cleanup the task by putting
the run function inside try catch or I can implement cleanup method in class
AwsBaseWaiterTrigger.
Let me know what you think.
--
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]
