You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "shivaam (via GitHub)" <gi...@apache.org> on 2026/03/23 01:26:36 UTC
[PR] Fix AwsBaseWaiterTrigger losing error details on deferred task failure [airflow]
shivaam opened a new pull request, #64085:
URL: https://github.com/apache/airflow/pull/64085
When a deferred AWS task hits a terminal failure state, `async_wait()` raises
`AirflowException` with the error details. But `AwsBaseWaiterTrigger.run()` did
not catch it — the exception propagated to the triggerer framework which replaced
it with a generic "Trigger failure" message. `execute_complete()` was never called,
so operators and `on_failure_callback` lost all error context.
**Fix:** Catch `AirflowException` in `run()` and yield
`TriggerEvent(status="error", message=str(e))`, routing failures through
`execute_complete()` where operators handle non-success events.
Also adds `"JobRun.ErrorMessage"` to `GlueJobCompleteTrigger.status_queries` so the
actual Glue error text is included alongside `JobRunState`.
related: https://github.com/apache/airflow/discussions/63706
## Tested on EC2 with real Glue job
Ran a Glue job designed to fail with `deferrable=True`. Before the fix,
`on_failure_callback` received `TaskDeferralError("Trigger failure")`. After:
```
AirflowException: Error in glue job: {
'status': 'error',
'message': 'AWS Glue job failed.: FAILED - RuntimeError: GLUE_TEST: Intentional failure
to test error propagation in deferrable mode\nWaiter job_complete failed: Waiter
encountered a terminal failure state: For expression "JobRun.JobRunState" we matched
expected path: "FAILED"',
'run_id': 'jr_a485d84c34f952e1c8d9d3199455d2d6eda07529034ceb7e1b554514d367f417'
}
```
`execute_complete()` was called, error details preserved, `on_failure_callback` received
the actual Glue error.
## Operator compatibility audit — feedback requested
This change means `execute_complete()` will now be called with `status="error"` events
where previously it was never called on failure. I audited all 59 `execute_complete`
methods across AWS operators and sensors:
- **47 are safe** — use `validated_event["status"] != "success"` with raise
- **4 not affected** — their triggers override `run()` with own error handling
- **8 have missing error handling** — would silently succeed on error events:
| Operator | Issue |
|----------|-------|
| [DmsDeleteReplicationConfigOperator](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/dms.py#L512) | No status check |
| [DmsStartReplicationOperator](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/dms.py#L707) | No status check |
| [DmsStopReplicationOperator](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/dms.py#L800) | No status check |
| [NeptuneStartDbClusterOperator](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py#L189) | Reads status but never validates |
| [NeptuneStopDbClusterOperator](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py#L316) | Reads status but never validates |
| [EmrServerlessStopApplicationOperator](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py#L1636) | `== "success"` with no else/raise |
| [EmrServerlessDeleteApplicationOperator](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/operators/emr.py#L1743) | `== "success"` with no else/raise |
| [MwaaTaskSensor](https://github.com/apache/airflow/blob/main/providers/amazon/src/airflow/providers/amazon/aws/sensors/mwaa.py#L283) | `return None` unconditionally |
**These are pre-existing bugs** (they never validated the event status), but merging the
base trigger fix without addressing them would change their failure mode from "Trigger
failure" crash to silent success — which is worse.
**Reviewers:** Does the `TriggerEvent(status="error")` approach look right? If so, I'll
add commits to fix all 8 operators in this PR before merge.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (Claude Opus 4.6)
Generated-by: Claude Code (Claude Opus 4.6) following [the guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix AwsBaseWaiterTrigger losing error details on deferred task failure [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on PR #64085:
URL: https://github.com/apache/airflow/pull/64085#issuecomment-4137013502
@vincbeck / @ferruzzi let me know if you will have any comments as I have updated all the operators to handle the trigger error event
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix AwsBaseWaiterTrigger losing error details on deferred task failure [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on code in PR #64085:
URL: https://github.com/apache/airflow/pull/64085#discussion_r2972440069
##########
providers/amazon/src/airflow/providers/amazon/aws/triggers/base.py:
##########
@@ -149,13 +150,17 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
client=client,
config_overrides=self.waiter_config_overrides,
)
- await async_wait(
- waiter,
- self.waiter_delay,
- self.attempts,
- self.waiter_args,
- self.failure_message,
- self.status_message,
- self.status_queries,
- )
- yield TriggerEvent({"status": "success", self.return_key: self.return_value})
+ try:
+ await async_wait(
+ waiter,
+ self.waiter_delay,
+ self.attempts,
+ self.waiter_args,
+ self.failure_message,
+ self.status_message,
+ self.status_queries,
+ )
+ except AirflowException as e:
+ yield TriggerEvent({"status": "error", "message": str(e), self.return_key: self.return_value})
Review Comment:
@o-nikolas
Does the TriggerEvent(status="error") approach look right? If so, I'll
add commits to fix all 8 operators in this PR before we can merge.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix AwsBaseWaiterTrigger losing error details on deferred task failure [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck merged PR #64085:
URL: https://github.com/apache/airflow/pull/64085
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix AwsBaseWaiterTrigger losing error details on deferred task failure [airflow]
Posted by "eladkal (via GitHub)" <gi...@apache.org>.
eladkal commented on PR #64085:
URL: https://github.com/apache/airflow/pull/64085#issuecomment-4113226236
Just to clarify this fix also closes https://github.com/apache/airflow/issues/64095 right?
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix AwsBaseWaiterTrigger losing error details on deferred task failure [airflow]
Posted by "ferruzzi (via GitHub)" <gi...@apache.org>.
ferruzzi commented on code in PR #64085:
URL: https://github.com/apache/airflow/pull/64085#discussion_r2977946099
##########
providers/amazon/src/airflow/providers/amazon/aws/operators/neptune.py:
##########
@@ -187,14 +188,13 @@ def execute(self, context: Context, event: dict[str, Any] | None = None, **kwarg
return {"db_cluster_id": self.cluster_id}
def execute_complete(self, context: Context, event: dict[str, Any] | None = None) -> dict[str, str]:
- status = ""
- cluster_id = ""
+ validated_event = validate_execute_complete_event(event)
- if event:
- status = event.get("status", "")
- cluster_id = event.get("cluster_id", "")
+ if validated_event["status"] != "success":
+ raise AirflowException(f"Error starting Neptune cluster: {validated_event}")
- self.log.info("Neptune cluster %s available with status: %s", cluster_id, status)
Review Comment:
Is there a reason to drop `status` from the log here (and int he similar logs below)?
##########
providers/amazon/src/airflow/providers/amazon/aws/triggers/base.py:
##########
@@ -149,13 +150,17 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
client=client,
config_overrides=self.waiter_config_overrides,
)
- await async_wait(
- waiter,
- self.waiter_delay,
- self.attempts,
- self.waiter_args,
- self.failure_message,
- self.status_message,
- self.status_queries,
- )
- yield TriggerEvent({"status": "success", self.return_key: self.return_value})
+ try:
+ await async_wait(
+ waiter,
+ self.waiter_delay,
+ self.attempts,
+ self.waiter_args,
+ self.failure_message,
+ self.status_message,
+ self.status_queries,
+ )
+ except AirflowException as e:
+ yield TriggerEvent({"status": "error", "message": str(e), self.return_key: self.return_value})
Review Comment:
You don't see many try/except/else blocks. I like it.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix AwsBaseWaiterTrigger losing error details on deferred task failure [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on PR #64085:
URL: https://github.com/apache/airflow/pull/64085#issuecomment-4122710566
> Setting temporary request changes to avoid merge
>
> > Note: This PR is blocked until all underlying operators are updated to handle the new error trigger event. Otherwise, these operators will fail silently.
@eladkal I have updated all the 7 sensors to handle trigger event 'error'
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix AwsBaseWaiterTrigger losing error details on deferred task failure [airflow]
Posted by "shivaam (via GitHub)" <gi...@apache.org>.
shivaam commented on PR #64085:
URL: https://github.com/apache/airflow/pull/64085#issuecomment-4115275640
> Just to clarify this fix also closes #64095 right?
Yes, it will close that issue as well
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org