[Vamsi-klu]

What is the change?

CloudComposerDAGRunSensor and CloudComposerDAGRunTrigger now succeed only when at least one Composer Dag run falls inside the requested execution window and every in-window run is in an allowed state. An empty window means keep waiting, not success.

Why did I do it?

The old window check returned success whenever it saw no in-window run in a disallowed state, so a window with no matching runs at all passed immediately. Unrelated out-of-window runs could make the sensor succeed even though the run it was waiting for never happened.

closes: #57512

How did I do it?

I extracted shared helpers into utils/composer.py, used by both the sync sensor and the deferrable trigger: a half-open [start, end) window, null-date skipping, naive timestamps normalized to UTC, and empty allowed_states defaulting to ["success"]. The hook's get_dag_runs now exhausts REST pagination. composer_dag_run_id is templated, and the system test example pins each sensor to the run id minted by an upstream CloudComposerTriggerDAGRunOperator via XCom, with a defensive timeout. The external-task sensors stay on the legacy windowed path; that anti-pattern is tracked in #67051.

What's the impact?

No more false successes from empty windows. Users who relied on the old behavior should set a finite timeout or pin composer_dag_run_id, since an empty window now waits.

What's the test plan?

Unit tests for the sensor, trigger, hook pagination, and the new window utilities (boundary, null and naive dates, run-id precedence, template rendering). I ran ruff, the provider pytest suite for these files, and the prek ruff hooks. The live Composer system test was not run; it needs real GCP infrastructure.

---

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

• Yes - Codex (GPT-5) and Claude Code (Opus 4.7 1M)

Generated-by: Codex (GPT-5) and Claude Code (Opus 4.7 1M) following the guidelines

[shahar1]

Could you please demonstrate that system tests pass when the proposed changes are included?
See past discussion here: #61046

[nnguyen168]

Hi @shahar1 @Vamsi-klu I ran the unit tests and local simulation for this PR.

Environment:

• macOS (Darwin)
• Python 3.12.11
• uv 0.11.16

Unit Test Results:

• test_cloud_composer.py (sensors): 59/59 passed ✅
• test_cloud_composer.py (triggers): 16/16 passed ✅

Local Simulation (dev/simulate_composer_system_test.py):

• Final state: success ✅
• Both sync dag_run_sensor and deferred defer_dag_run_sensor matched via composer_dag_run_id branch
• All 18 tasks completed successfully (watcher skipped as expected)

Note: I don't have GCP credentials to run the actual system tests against real Composer infrastructure. The unit tests and local simulation validate the control flow, but real GCP validation is still needed.

[nnguyen168]

  OK: dag.test() completed for composer -- final state: success
  the sync `dag_run_sensor` and deferred `defer_dag_run_sensor` both 
  matched on dag_run_id='manual__simulated' via the composer_dag_run_id branch.

[MaksYermak]

@nnguyen168 could you please run the existing System Test for this operator and shared the screenshot for the latest run?

[shahar1]

Hi @shahar1 -- addressed the system-test concern.

Why PR #61046 had to be reverted (revert commit 14995f1): the example called CloudComposerDAGRunSensor without execution_range, so the sensor inferred the window [logical_date - 1d, logical_date]. The Composer environment was created mid-DAG-run, which meant no airflow_monitoring runs ever fell inside that past window. The buggy code returned True (false positive); the empty-window fix correctly returns False and the example then hung forever.

The example patch in the earlier revision of this PR tried to side-step the issue with execution_range=[datetime.now() - timedelta(1), datetime.now()], but that's evaluated at DAG-parse time -- roughly half an hour before the sensor pokes and before the Composer env exists. Same hang mode, just hidden.

What this commit changes:

• composer_dag_run_id added to CloudComposerDAGRunSensor.template_fields.
• Example reordered so each CloudComposerTriggerDAGRunOperator runs before its sibling Dag-run sensor; sensors pull the freshly minted dag_run_id via XCom. The run-id branch is pure id+state equality and bypasses execution_range entirely, so the hang mode cannot recur on the new path.
• Defensive timeout=60*60 on the Dag-run sensors so any future regression to a windowed code path fails CI fast rather than hanging the worker.
• External-task sensors deliberately left on the legacy windowed path -- the analogous _check_task_instances_states anti-pattern is tracked at CloudComposerExternalTaskSensor succeeds when all task instances are outside execution_range #67051, and the example carries an inline note so the CloudComposerExternalTaskSensor succeeds when all task instances are outside execution_range #67051 fix must reorder them at the same time.

Verification I can run locally without GCP credentials:

• six new sensor unit tests + two new trigger unit tests covering the composer_dag_run_id branch (template_fields, Jinja render, poke success/wait, range-vs-id precedence, parse-time source guard, trigger branch precedence, trigger polling) -- all 75 tests in providers/google/tests/unit/google/cloud/{sensors,triggers}/test_cloud_composer.py pass.
• dev/simulate_composer_system_test.py, a standalone harness that loads the example via importlib, mocks every GCP surface the example touches (googleapiclient + CloudComposerHook + CloudComposerAsyncHook + Environment / ImageVersion / ExecuteAirflowCommandResponse.to_dict), runs dag.test() end-to-end, and reports per-task state. The deferred trigger runs inline under dag.test(), so the deferrable sensor path is exercised too.

Simulation result:

$ uv run --project providers/google python dev/simulate_composer_system_test.py
loading example dag from .../example_cloud_composer.py
registering dag bundle and serializing dag
running dag.test() on composer
OK: dag.test() completed for composer -- final state: success
  create_env                       -> success
  dag_run_sensor                   -> success
  defer_create_env                 -> success
  defer_dag_run_sensor             -> success
  defer_delete_env                 -> success
  defer_external_task_sensor       -> success
  defer_run_airflow_cli_cmd        -> success
  defer_trigger_dag_run            -> success
  defer_update_env                 -> success
  delete_env                       -> success
  external_task_sensor             -> success
  get_env                          -> success
  get_project_number               -> success
  image_versions                   -> success
  list_envs                        -> success
  run_airflow_cli_cmd              -> success
  trigger_dag_run                  -> success
  update_env                       -> success
  watcher                          -> skipped

Both the sync dag_run_sensor and the deferred defer_dag_run_sensor matched on the trigger ops' run id via the composer_dag_run_id branch.

@VladaZakharova @MaksYermak -- would appreciate a confirmation pass against real Composer infra when convenient, since the simulation can only validate the control flow, not the actual REST API contracts.

Drafted-by: Claude Code (Opus 4.7 1M); reviewed by @Vamsi-klu before posting

When asking to run the system tests, I always refer to a real GCP instance - simulating it against a mock isn't helpful (maybe with exception for very specific services in GCP that have emulators).
Therefore, please remove dev/simulate_composer_system_test.py.
As you currently lack the means to run the system tests, I'll see if I'm able to help with running them (can't promise that it will happen soon).

[nnguyen168]

I've validated the unit tests (75/75 passing) and local simulation, but I don't have GCP credits remaining to run the full system test against real Composer infrastructure (~$2-4 cost per run).

Could a maintainer or someone with GCP access help run the system test and share the results? Alternatively, is there a CI workflow for system tests that can be triggered?

[potiuk]

@shahar1 I still cannot run example_cloud_composer on real Composer from here.

Your merge condition is that live run. I can leave this PR open if you or the Google team still plan to run it on the community project. I can also close it until someone actually has a run, and reopen or open a new PR then.

Which do you prefer? I don't want this sitting in the review queue if the test is not going to happen soon. I don't want to increase the review backlog.

Then we should close it @Vamsi-klu -> you seem to open a lot of PRs with Claude Code but it seems you do not test it. If you cannot run system test now, it means your changes are theorethical and basically you are throwing testing over the wall to someone else. Which is not a good idea. I am closing all Google provider ones from you in this case - feel free to reopen your PRs for providers with a proof that you actuallly run the tests manually with real Google account.

Also whenever we release a provider with such changes you are requested to installl an test such modified provider with real google service.

The providers are supposed to be run with raal services they are accessing and not "theorethically" fixed.