Fire failure listeners when TriggerDagRunOperator API call fails

[ABiscuitttt]

When TriggerDagRunOperator targets a nonexistent Dag (or the API server otherwise rejects the trigger with 403/5xx), the supervisor reports the error by raising AirflowRuntimeError from the TriggerDagRun comms round-trip. That raise happens inside run()'s except DagRunTriggerException handler, and an exception raised in an except handler is not caught by the sibling except clauses — it escaped run() entirely. main() then never reached finalize(), the only place on_task_instance_failed / on_failure_callback listeners run. The task still showed FAILED in the UI (via the supervisor's exit-code fallback), but alerting plugins were silently skipped for exactly this failure mode.

The fix routes the error through the normal failure path (_handle_current_task_failed), so the task returns a terminal state, finalize() fires the listeners, and retry policy is honored (UP_FOR_RETRY when retries remain).

Relationship to prior work on #63089: the reported UnboundLocalError symptom was already fixed by #67955 (the state variable is now initialized and the ti.finish stats emission is guarded), and #63207 was an earlier, unmerged attempt at the full fix. What remains — and what this PR addresses — is the API error escaping run() entirely, which silently skips failure listeners, failure callbacks, and retry handling.

Regression tests cover both the no-retry (FAILED) and retry (UP_FOR_RETRY) paths, asserting the terminal state reaches the supervisor instead of the error escaping run().

closes: #63089

---

Was generative AI tooling used to co-author this PR?

• Yes — Kimi Code CLI

Generated-by: Kimi Code CLI following the guidelines

[ABiscuitttt]

Some context on how we ran into this: we run a global failure-alert listener plugin that pages our on-call channel (Feishu/Lark) through the on_task_instance_failed hook when a task instance reaches its final FAILED state:

@hookimpldefon_task_instance_failed(previous_state, task_instance, error):
iftask_instance.state!=TaskInstanceState.FAILED:
return# still up_for_retry; only alert on final failuresend_alert(task_instance, error) # internal notification

When a TriggerDagRunOperator points at a non-existent Dag, the task dies from the API error without this hook ever firing, so no alert goes out — from the on-call perspective the task fails silently. That is what led to this fix: routing the API error through the normal failure path makes finalize() fire the listener (and honor the retry policy) as expected.

[ABiscuitttt]

Closing as superseded by #70719, which addresses the same TriggerDagRunOperator 404 failure-listener issue more generally.

[ABiscuitttt]

When TriggerDagRunOperator targets a nonexistent Dag (or the API server otherwise rejects the trigger with 403/5xx), the supervisor reports the error by raising AirflowRuntimeError from the TriggerDagRun comms round-trip. That raise happens inside run()'s except DagRunTriggerException handler, and an exception raised in an except handler is not caught by the sibling except clauses — it escaped run() entirely. main() then never reached finalize(), the only place on_task_instance_failed / on_failure_callback listeners run. The task still showed FAILED in the UI (via the supervisor's exit-code fallback), but alerting plugins were silently skipped for exactly this failure mode.

The fix routes the error through the normal failure path (_handle_current_task_failed), so the task returns a terminal state, finalize() fires the listeners, and retry policy is honored (UP_FOR_RETRY when retries remain).

Relationship to prior work on #63089: the reported UnboundLocalError symptom was already fixed by #67955 (the state variable is now initialized and the ti.finish stats emission is guarded), and #63207 was an earlier, unmerged attempt at the full fix. What remains — and what this PR addresses — is the API error escaping run() entirely, which silently skips failure listeners, failure callbacks, and retry handling.

Regression tests cover both the no-retry (FAILED) and retry (UP_FOR_RETRY) paths, asserting the terminal state reaches the supervisor instead of the error escaping run().

closes: #63089

---

Was generative AI tooling used to co-author this PR?

• Yes — Kimi Code CLI

Generated-by: Kimi Code CLI following the guidelines

[ABiscuitttt]

Some context on how we ran into this: we run a global failure-alert listener plugin that pages our on-call channel (Feishu/Lark) through the on_task_instance_failed hook when a task instance reaches its final FAILED state:

@hookimpldefon_task_instance_failed(previous_state, task_instance, error):
iftask_instance.state!=TaskInstanceState.FAILED:
return# still up_for_retry; only alert on final failuresend_alert(task_instance, error) # internal notification

When a TriggerDagRunOperator points at a non-existent Dag, the task dies from the API error without this hook ever firing, so no alert goes out — from the on-call perspective the task fails silently. That is what led to this fix: routing the API error through the normal failure path makes finalize() fire the listener (and honor the retry policy) as expected.

[ABiscuitttt]

Closing as superseded by #70719, which addresses the same TriggerDagRunOperator 404 failure-listener issue more generally.

[ABiscuitttt]

When TriggerDagRunOperator targets a nonexistent Dag (or the API server otherwise rejects the trigger with 403/5xx), the supervisor reports the error by raising AirflowRuntimeError from the TriggerDagRun comms round-trip. That raise happens inside run()'s except DagRunTriggerException handler, and an exception raised in an except handler is not caught by the sibling except clauses — it escaped run() entirely. main() then never reached finalize(), the only place on_task_instance_failed / on_failure_callback listeners run. The task still showed FAILED in the UI (via the supervisor's exit-code fallback), but alerting plugins were silently skipped for exactly this failure mode.

The fix routes the error through the normal failure path (_handle_current_task_failed), so the task returns a terminal state, finalize() fires the listeners, and retry policy is honored (UP_FOR_RETRY when retries remain).

Relationship to prior work on #63089: the reported UnboundLocalError symptom was already fixed by #67955 (the state variable is now initialized and the ti.finish stats emission is guarded), and #63207 was an earlier, unmerged attempt at the full fix. What remains — and what this PR addresses — is the API error escaping run() entirely, which silently skips failure listeners, failure callbacks, and retry handling.

Regression tests cover both the no-retry (FAILED) and retry (UP_FOR_RETRY) paths, asserting the terminal state reaches the supervisor instead of the error escaping run().

closes: #63089

---

Was generative AI tooling used to co-author this PR?

• Yes — Kimi Code CLI

Generated-by: Kimi Code CLI following the guidelines

[ABiscuitttt]

Some context on how we ran into this: we run a global failure-alert listener plugin that pages our on-call channel (Feishu/Lark) through the on_task_instance_failed hook when a task instance reaches its final FAILED state:

@hookimpldefon_task_instance_failed(previous_state, task_instance, error):
iftask_instance.state!=TaskInstanceState.FAILED:
return# still up_for_retry; only alert on final failuresend_alert(task_instance, error) # internal notification

When a TriggerDagRunOperator points at a non-existent Dag, the task dies from the API error without this hook ever firing, so no alert goes out — from the on-call perspective the task fails silently. That is what led to this fix: routing the API error through the normal failure path makes finalize() fire the listener (and honor the retry policy) as expected.

[ABiscuitttt]

Closing as superseded by #70719, which addresses the same TriggerDagRunOperator 404 failure-listener issue more generally.

[ABiscuitttt]

When TriggerDagRunOperator targets a nonexistent Dag (or the API server otherwise rejects the trigger with 403/5xx), the supervisor reports the error by raising AirflowRuntimeError from the TriggerDagRun comms round-trip. That raise happens inside run()'s except DagRunTriggerException handler, and an exception raised in an except handler is not caught by the sibling except clauses — it escaped run() entirely. main() then never reached finalize(), the only place on_task_instance_failed / on_failure_callback listeners run. The task still showed FAILED in the UI (via the supervisor's exit-code fallback), but alerting plugins were silently skipped for exactly this failure mode.

The fix routes the error through the normal failure path (_handle_current_task_failed), so the task returns a terminal state, finalize() fires the listeners, and retry policy is honored (UP_FOR_RETRY when retries remain).

Relationship to prior work on #63089: the reported UnboundLocalError symptom was already fixed by #67955 (the state variable is now initialized and the ti.finish stats emission is guarded), and #63207 was an earlier, unmerged attempt at the full fix. What remains — and what this PR addresses — is the API error escaping run() entirely, which silently skips failure listeners, failure callbacks, and retry handling.

Regression tests cover both the no-retry (FAILED) and retry (UP_FOR_RETRY) paths, asserting the terminal state reaches the supervisor instead of the error escaping run().

closes: #63089

---

Was generative AI tooling used to co-author this PR?

• Yes — Kimi Code CLI

Generated-by: Kimi Code CLI following the guidelines

[ABiscuitttt]

Some context on how we ran into this: we run a global failure-alert listener plugin that pages our on-call channel (Feishu/Lark) through the on_task_instance_failed hook when a task instance reaches its final FAILED state:

@hookimpldefon_task_instance_failed(previous_state, task_instance, error):
iftask_instance.state!=TaskInstanceState.FAILED:
return# still up_for_retry; only alert on final failuresend_alert(task_instance, error) # internal notification

When a TriggerDagRunOperator points at a non-existent Dag, the task dies from the API error without this hook ever firing, so no alert goes out — from the on-call perspective the task fails silently. That is what led to this fix: routing the API error through the normal failure path makes finalize() fire the listener (and honor the retry policy) as expected.

[ABiscuitttt]

Closing as superseded by #70719, which addresses the same TriggerDagRunOperator 404 failure-listener issue more generally.

[ABiscuitttt]

When TriggerDagRunOperator targets a nonexistent Dag (or the API server otherwise rejects the trigger with 403/5xx), the supervisor reports the error by raising AirflowRuntimeError from the TriggerDagRun comms round-trip. That raise happens inside run()'s except DagRunTriggerException handler, and an exception raised in an except handler is not caught by the sibling except clauses — it escaped run() entirely. main() then never reached finalize(), the only place on_task_instance_failed / on_failure_callback listeners run. The task still showed FAILED in the UI (via the supervisor's exit-code fallback), but alerting plugins were silently skipped for exactly this failure mode.

The fix routes the error through the normal failure path (_handle_current_task_failed), so the task returns a terminal state, finalize() fires the listeners, and retry policy is honored (UP_FOR_RETRY when retries remain).

Relationship to prior work on #63089: the reported UnboundLocalError symptom was already fixed by #67955 (the state variable is now initialized and the ti.finish stats emission is guarded), and #63207 was an earlier, unmerged attempt at the full fix. What remains — and what this PR addresses — is the API error escaping run() entirely, which silently skips failure listeners, failure callbacks, and retry handling.

Regression tests cover both the no-retry (FAILED) and retry (UP_FOR_RETRY) paths, asserting the terminal state reaches the supervisor instead of the error escaping run().

closes: #63089

---

Was generative AI tooling used to co-author this PR?

• Yes — Kimi Code CLI

Generated-by: Kimi Code CLI following the guidelines

[ABiscuitttt]

Some context on how we ran into this: we run a global failure-alert listener plugin that pages our on-call channel (Feishu/Lark) through the on_task_instance_failed hook when a task instance reaches its final FAILED state:

@hookimpldefon_task_instance_failed(previous_state, task_instance, error):
iftask_instance.state!=TaskInstanceState.FAILED:
return# still up_for_retry; only alert on final failuresend_alert(task_instance, error) # internal notification

When a TriggerDagRunOperator points at a non-existent Dag, the task dies from the API error without this hook ever firing, so no alert goes out — from the on-call perspective the task fails silently. That is what led to this fix: routing the API error through the normal failure path makes finalize() fire the listener (and honor the retry policy) as expected.

[ABiscuitttt]

Closing as superseded by #70719, which addresses the same TriggerDagRunOperator 404 failure-listener issue more generally.

[ABiscuitttt]

When TriggerDagRunOperator targets a nonexistent Dag (or the API server otherwise rejects the trigger with 403/5xx), the supervisor reports the error by raising AirflowRuntimeError from the TriggerDagRun comms round-trip. That raise happens inside run()'s except DagRunTriggerException handler, and an exception raised in an except handler is not caught by the sibling except clauses — it escaped run() entirely. main() then never reached finalize(), the only place on_task_instance_failed / on_failure_callback listeners run. The task still showed FAILED in the UI (via the supervisor's exit-code fallback), but alerting plugins were silently skipped for exactly this failure mode.

The fix routes the error through the normal failure path (_handle_current_task_failed), so the task returns a terminal state, finalize() fires the listeners, and retry policy is honored (UP_FOR_RETRY when retries remain).

Relationship to prior work on #63089: the reported UnboundLocalError symptom was already fixed by #67955 (the state variable is now initialized and the ti.finish stats emission is guarded), and #63207 was an earlier, unmerged attempt at the full fix. What remains — and what this PR addresses — is the API error escaping run() entirely, which silently skips failure listeners, failure callbacks, and retry handling.

Regression tests cover both the no-retry (FAILED) and retry (UP_FOR_RETRY) paths, asserting the terminal state reaches the supervisor instead of the error escaping run().

closes: #63089

---

Was generative AI tooling used to co-author this PR?

• Yes — Kimi Code CLI

Generated-by: Kimi Code CLI following the guidelines

[ABiscuitttt]

Some context on how we ran into this: we run a global failure-alert listener plugin that pages our on-call channel (Feishu/Lark) through the on_task_instance_failed hook when a task instance reaches its final FAILED state:

@hookimpldefon_task_instance_failed(previous_state, task_instance, error):
iftask_instance.state!=TaskInstanceState.FAILED:
return# still up_for_retry; only alert on final failuresend_alert(task_instance, error) # internal notification

When a TriggerDagRunOperator points at a non-existent Dag, the task dies from the API error without this hook ever firing, so no alert goes out — from the on-call perspective the task fails silently. That is what led to this fix: routing the API error through the normal failure path makes finalize() fire the listener (and honor the retry policy) as expected.

[ABiscuitttt]

Closing as superseded by #70719, which addresses the same TriggerDagRunOperator 404 failure-listener issue more generally.

[ABiscuitttt]

When TriggerDagRunOperator targets a nonexistent Dag (or the API server otherwise rejects the trigger with 403/5xx), the supervisor reports the error by raising AirflowRuntimeError from the TriggerDagRun comms round-trip. That raise happens inside run()'s except DagRunTriggerException handler, and an exception raised in an except handler is not caught by the sibling except clauses — it escaped run() entirely. main() then never reached finalize(), the only place on_task_instance_failed / on_failure_callback listeners run. The task still showed FAILED in the UI (via the supervisor's exit-code fallback), but alerting plugins were silently skipped for exactly this failure mode.

The fix routes the error through the normal failure path (_handle_current_task_failed), so the task returns a terminal state, finalize() fires the listeners, and retry policy is honored (UP_FOR_RETRY when retries remain).

Relationship to prior work on #63089: the reported UnboundLocalError symptom was already fixed by #67955 (the state variable is now initialized and the ti.finish stats emission is guarded), and #63207 was an earlier, unmerged attempt at the full fix. What remains — and what this PR addresses — is the API error escaping run() entirely, which silently skips failure listeners, failure callbacks, and retry handling.

Regression tests cover both the no-retry (FAILED) and retry (UP_FOR_RETRY) paths, asserting the terminal state reaches the supervisor instead of the error escaping run().

closes: #63089

---

Was generative AI tooling used to co-author this PR?

• Yes — Kimi Code CLI

Generated-by: Kimi Code CLI following the guidelines

[ABiscuitttt]

Some context on how we ran into this: we run a global failure-alert listener plugin that pages our on-call channel (Feishu/Lark) through the on_task_instance_failed hook when a task instance reaches its final FAILED state:

@hookimpldefon_task_instance_failed(previous_state, task_instance, error):
iftask_instance.state!=TaskInstanceState.FAILED:
return# still up_for_retry; only alert on final failuresend_alert(task_instance, error) # internal notification

When a TriggerDagRunOperator points at a non-existent Dag, the task dies from the API error without this hook ever firing, so no alert goes out — from the on-call perspective the task fails silently. That is what led to this fix: routing the API error through the normal failure path makes finalize() fire the listener (and honor the retry policy) as expected.

[ABiscuitttt]

Closing as superseded by #70719, which addresses the same TriggerDagRunOperator 404 failure-listener issue more generally.

[ABiscuitttt]

When TriggerDagRunOperator targets a nonexistent Dag (or the API server otherwise rejects the trigger with 403/5xx), the supervisor reports the error by raising AirflowRuntimeError from the TriggerDagRun comms round-trip. That raise happens inside run()'s except DagRunTriggerException handler, and an exception raised in an except handler is not caught by the sibling except clauses — it escaped run() entirely. main() then never reached finalize(), the only place on_task_instance_failed / on_failure_callback listeners run. The task still showed FAILED in the UI (via the supervisor's exit-code fallback), but alerting plugins were silently skipped for exactly this failure mode.

The fix routes the error through the normal failure path (_handle_current_task_failed), so the task returns a terminal state, finalize() fires the listeners, and retry policy is honored (UP_FOR_RETRY when retries remain).

Relationship to prior work on #63089: the reported UnboundLocalError symptom was already fixed by #67955 (the state variable is now initialized and the ti.finish stats emission is guarded), and #63207 was an earlier, unmerged attempt at the full fix. What remains — and what this PR addresses — is the API error escaping run() entirely, which silently skips failure listeners, failure callbacks, and retry handling.

Regression tests cover both the no-retry (FAILED) and retry (UP_FOR_RETRY) paths, asserting the terminal state reaches the supervisor instead of the error escaping run().

closes: #63089

---

Was generative AI tooling used to co-author this PR?

• Yes — Kimi Code CLI

Generated-by: Kimi Code CLI following the guidelines

[ABiscuitttt]

Some context on how we ran into this: we run a global failure-alert listener plugin that pages our on-call channel (Feishu/Lark) through the on_task_instance_failed hook when a task instance reaches its final FAILED state:

@hookimpldefon_task_instance_failed(previous_state, task_instance, error):
iftask_instance.state!=TaskInstanceState.FAILED:
return# still up_for_retry; only alert on final failuresend_alert(task_instance, error) # internal notification

When a TriggerDagRunOperator points at a non-existent Dag, the task dies from the API error without this hook ever firing, so no alert goes out — from the on-call perspective the task fails silently. That is what led to this fix: routing the API error through the normal failure path makes finalize() fire the listener (and honor the retry policy) as expected.

[ABiscuitttt]

Closing as superseded by #70719, which addresses the same TriggerDagRunOperator 404 failure-listener issue more generally.