ashb commented on code in PR #65590:
URL: https://github.com/apache/airflow/pull/65590#discussion_r3117115070
##########
airflow-core/src/airflow/jobs/triggerer_job_runner.py:
##########
@@ -1331,14 +1335,27 @@ async def run_trigger(
self.events.append((trigger_id, event))
span.set_status(Status(StatusCode.OK))
except asyncio.CancelledError as e:
- # We get cancelled by the scheduler changing the task state.
But if we do lets give a nice error
- # message about it
+ # A trigger can be cancelled for two reasons:
+ # - The user acted on the task (mark failed / clear / mark
succeeded).
+ # - The triggerer is shutting down, here cancel_triggers()
is not
+ # involved — the shutdown path cancels tasks directly
without a message.
+ # Only first case should invoke on_cancel().
+ #
+ # For timeout, raise immediately without calling on_cancel().
if timeout := timeout_after:
timeout = timeout.replace(tzinfo=timezone.utc) if not
timeout.tzinfo else timeout
if timeout < timezone.utcnow():
await self.log.aerror("Trigger cancelled due to
timeout")
span.set_status(Status(StatusCode.ERROR),
description=str(e))
raise
+ if e.args and e.args[0] == USER_ACTION_INDICATOR:
+ # Suppress exceptions from on_cancel() so they do not
digest the
+ # CancelledError errors.
+ with suppress(BaseException):
Review Comment:
Specifically the suppress call.
--
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]
