Fail deferred Cloud Composer tasks when the GCP operation errors

[SEPURI-SAI-KRISHNA]

CloudComposerExecutionTrigger.run() checked operation.done before operation.error, in an if/elif. Per the long-running operation contract, "if done == false, neither error nor response is set", and a finished operation has done=True with exactly one of error or response populated.

So the two branches were inverted relative to the API:

• the elif operation.error.message branch was only ever reached while done was false — the one state where error is guaranteed unset, making it dead code;
• a genuinely failed operation (done=True with error set) broke out of the loop and yielded operation_done: True, the success shape.

The consequence differs per operator, all three of which defer with this trigger:

Operator	execute_complete	Result when the operation failed
CloudComposerUpdateEnvironmentOperator	returns on operation_done, no state check	task marked success
CloudComposerDeleteEnvironmentOperator	pass	task marked success
CloudComposerCreateEnvironmentOperator	re-fetches the environment and validates its state	task fails, but with a misleading NotFound/wrong-state error rather than the GCP failure reason

This also made the deferrable path disagree with the synchronous one: with deferrable=False the operators call hook.wait_for_operation() → operation.result(), which raises on a failed operation.

Checking error inside the done branch fixes all three. The existing raise is relocated, not new.

CloudComposerExecutionTrigger had no test coverage, so this adds two cases: a finished-with-error operation must raise, and a finished-without-error operation must still yield operation_done: True. The first fails without this change; the second guards the new nested condition against over-raising on success.

---

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

• Yes — Claude Code (Opus 5)

Generated-by: Claude Code (Opus 5) following the guidelines

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[potiuk]

Good catch, and the diagnosis is exactly right. Because the long-running operation contract guarantees that neither error nor response is populated while done is false, the elif operation.error.message branch could never be reached — a failed operation arrives as done=True with error set, took the first branch, broke out of the loop, and the trigger then yielded

TriggerEvent({"operation_name": ..., "operation_done": True})

which the operator reads as success. So a failed environment create/delete/update reported success rather than failing the task. Silent false success is about the worst failure mode a deferrable operator can have.

Nesting the check inside the done branch is the right fix, since that is the only state in which error can be set.

The tests are well chosen: test_run_raises_when_operation_finished_with_error genuinely fails against the old code — it would yield the success event instead of raising — and the companion test pins the happy path so the fix cannot over-fire. Putting the LRO contract in make_operation's docstring is a nice touch; it puts the reasoning where the next person will look.

I rebased the branch onto main before merging, as it had drifted a long way behind.

---

Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.